Python Async VAT Validation with httpx and asyncio (FastAPI & Django Guide)

The synchronous requests library works fine for one-off VAT lookups, but it blocks the event loop when you use it inside an async Python application. VIES latency averages 400ms–1.5 seconds — during which your FastAPI or Django async view cannot process other requests. The solution is httpx.AsyncClient, which gives you the same interface as requests but with full asyncio support. This guide covers single validation, bulk concurrent validation, and integration patterns for FastAPI and Django.

Installing httpx and setting up the client

pythonrequirements.txt / pyproject.toml

# Install httpx with asyncio support
pip install httpx

# Or with Poetry
poetry add httpx

pythonlib/vat_client.py

import httpx
import os

# Reuse a single AsyncClient for connection pooling — do not create one per request
_client: httpx.AsyncClient | None = None

def get_client() -> httpx.AsyncClient:
    global _client
    if _client is None or _client.is_closed:
        _client = httpx.AsyncClient(
            base_url='https://www.taxid.dev',
            headers={'Authorization': f'Bearer {os.environ["TAXID_API_KEY"]}'},
            timeout=httpx.Timeout(10.0, connect=5.0),
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
        )
    return _client

Warning

Do not create a new httpx.AsyncClient on every request. Connection setup adds 50–200ms overhead and exhausts file descriptors under load. Create the client once at application startup (or use a module-level singleton) and reuse it across requests. The get_client() function above handles this with a lazy singleton pattern.

Single async validation

pythonlib/validate_vat.py

from typing import Literal
from lib.vat_client import get_client

StatusCode = Literal['active', 'invalid', 'inactive', 'format_invalid', 'service_unavailable']

async def validate_vat(country_code: str, vat_number: str) -> dict:
    """Validate a single EU VAT number asynchronously."""
    client = get_client()
    try:
        response = await client.get(
            f'/api/v1/validate/{country_code}/{vat_number}'
        )
        response.raise_for_status()
        return response.json()
    except httpx.TimeoutException:
        return {'status': 'service_unavailable', 'error': 'timeout'}
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 503:
            return {'status': 'service_unavailable'}
        raise

FastAPI endpoint with async validation

pythonapi/routes/checkout.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from lib.validate_vat import validate_vat

router = APIRouter()

class CheckoutVatRequest(BaseModel):
    country_code: str
    vat_number: str
    customer_id: str

@router.post('/api/checkout/validate-vat')
async def validate_vat_at_checkout(req: CheckoutVatRequest):
    result = await validate_vat(req.country_code, req.vat_number)
    status = result.get('status')

    if status == 'service_unavailable':
        # VIES is down — allow through, schedule re-validation
        await db.schedule_revalidation(req.customer_id, req.vat_number, req.country_code)
        return {'proceed': True, 'reverse_charge': True, 'pending_revalidation': True}

    if status not in ('active',):
        raise HTTPException(
            status_code=422,
            detail=f'VAT number validation failed: {status}'
        )

    # Store audit record
    await db.vat_validations.create({
        'customer_id': req.customer_id,
        'vat_number': req.vat_number,
        'status': status,
        'company_name': result.get('company_name'),
        'request_id': result.get('request_id'),
    })

    return {
        'proceed': True,
        'reverse_charge': True,
        'company_name': result.get('company_name'),
    }

Bulk async validation with asyncio.gather

For bulk validation, use asyncio.gather with asyncio.Semaphore to limit concurrency. Without a semaphore, a large list will overwhelm the VIES backend and trigger rate limit responses.

pythonlib/bulk_validate.py

import asyncio
from lib.validate_vat import validate_vat

async def bulk_validate(
    numbers: list[tuple[str, str]],  # List of (country_code, vat_number)
    concurrency: int = 8,
) -> list[dict]:
    semaphore = asyncio.Semaphore(concurrency)

    async def validate_with_semaphore(country_code: str, vat_number: str) -> dict:
        async with semaphore:
            result = await validate_vat(country_code, vat_number)
            return {'country_code': country_code, 'vat_number': vat_number, **result}

    tasks = [
        validate_with_semaphore(country_code, vat_number)
        for country_code, vat_number in numbers
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    return [
        r if not isinstance(r, Exception)
        else {'vat_number': numbers[i][1], 'status': 'error', 'error': str(r)}
        for i, r in enumerate(results)
    ]

Django async view (Django 4.1+)

pythonviews/checkout.py

# Django 4.1+ supports async views natively
from django.http import JsonResponse
from django.views import View
from lib.validate_vat import validate_vat
import json

class ValidateVatView(View):
    async def post(self, request):
        body = json.loads(request.body)
        country_code = body.get('country_code', '').upper()
        vat_number = body.get('vat_number', '').upper()

        result = await validate_vat(country_code, vat_number)
        status = result.get('status')

        if status == 'service_unavailable':
            return JsonResponse({'proceed': True, 'pending_revalidation': True})

        if status != 'active':
            return JsonResponse({'proceed': False, 'status': status}, status=422)

        return JsonResponse({
            'proceed': True,
            'reverse_charge': True,
            'company_name': result.get('company_name'),
        })

Error handling: timeout, service unavailable, invalid

Exception / Status	Cause	Recommended action
httpx.TimeoutException	Request took longer than timeout (8–10s)	Return service_unavailable — allow customer through
httpx.ConnectError	Cannot reach the API (network issue)	Return service_unavailable — allow customer through
httpx.HTTPStatusError 503	API is temporarily unavailable	Return service_unavailable — retry later
status: service_unavailable	VIES upstream is down	Allow through, schedule re-validation
status: invalid / inactive	Number does not exist or was deregistered	Reject at checkout, prompt to correct
status: format_invalid	Wrong format for the country prefix	Reject — prompt user to check the number