VAT API Python: Integrating Tax ID Validation with Django & Flask

# Sync (requests) — works with standard Django/Flask
pip install requests

# Async (httpx) — for async Django 4.1+ or Flask with async views
pip install httpx

# Store API key in environment
export TAXID_API_KEY=vat_xxxxxxxxxxxxxxxx
# Or in .env (use python-dotenv):
# TAXID_API_KEY=vat_xxxxxxxxxxxxxxxx

import os
import requests
from typing import TypedDict, Literal

VatStatus = Literal['active', 'inactive', 'format_invalid', 'service_unavailable']

class VatResult(TypedDict):
    valid: bool
    status: VatStatus
    company_name: str | None
    address: str | None
    country_code: str
    cached: bool
    request_id: str

BASE_URL = 'https://taxid.dev/api/v1'

def validate_vat(vat_number: str) -> VatResult:
    """Validate a VAT number via TaxID API. Raises requests.RequestException on network error."""
    country = vat_number[:2].upper()
    vat = vat_number.replace(' ', '').upper()

    resp = requests.get(
        f'{BASE_URL}/validate/{country}/{vat}',
        headers={'Authorization': f'Bearer {os.environ["TAXID_API_KEY"]}'},
        timeout=5,
    )
    resp.raise_for_status()
    return resp.json()


def validate_vat_safe(vat_number: str) -> VatResult:
    """Same as validate_vat but returns service_unavailable on network errors."""
    try:
        return validate_vat(vat_number)
    except Exception:
        return {
            'valid': False,
            'status': 'service_unavailable',
            'company_name': None,
            'address': None,
            'country_code': vat_number[:2].upper(),
            'cached': False,
            'request_id': '',
        }

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .client import validate_vat_safe


class VatValidationMixin:
    def get_vat_result(self, vat_number: str) -> dict | None:
        """Returns error Response or None if validation can proceed."""
        result = validate_vat_safe(vat_number)

        if result['status'] == 'format_invalid':
            return Response(
                {'error': 'format_invalid', 'message': 'Invalid VAT number format.'},
                status=status.HTTP_422_UNPROCESSABLE_ENTITY,
            )
        if result['status'] == 'inactive':
            return Response(
                {'error': 'vat_inactive', 'message': 'VAT number is not currently registered.'},
                status=status.HTTP_422_UNPROCESSABLE_ENTITY,
            )
        return None  # valid or unavailable — let checkout proceed


class B2BCheckoutView(VatValidationMixin, APIView):
    def post(self, request):
        vat_number = request.data.get('vat_number', '').strip()

        if vat_number:
            error_response = self.get_vat_result(vat_number)
            if error_response:
                return error_response

        # VAT is valid (or blank for B2C) — proceed with order creation
        result = validate_vat_safe(vat_number) if vat_number else None
        company_name = result['company_name'] if result and result['status'] == 'active' else None

        # ... create order logic here ...
        return Response({'status': 'ok', 'company_name': company_name})

from flask import Blueprint, request, jsonify
from .client import validate_vat_safe

vat_bp = Blueprint('vat', __name__, url_prefix='/api')


@vat_bp.route('/validate-vat', methods=['POST'])
def validate_vat_endpoint():
    data = request.get_json(silent=True) or {}
    vat_number = (data.get('vat_number') or '').strip()

    if not vat_number:
        return jsonify({'error': 'vat_number_required'}), 400

    result = validate_vat_safe(vat_number)

    if result['status'] == 'format_invalid':
        return jsonify({'error': 'format_invalid'}), 422

    if result['status'] == 'inactive':
        return jsonify({'error': 'vat_inactive'}), 422

    return jsonify({
        'valid': result['valid'],
        'status': result['status'],
        'company_name': result['company_name'],
        'unavailable': result['status'] == 'service_unavailable',
    })


# Register in your app factory:
# app.register_blueprint(vat_bp)

import os
import httpx
from .client import VatResult

BASE_URL = 'https://taxid.dev/api/v1'

async def validate_vat_async(vat_number: str) -> VatResult:
    country = vat_number[:2].upper()
    vat = vat_number.replace(' ', '').upper()

    async with httpx.AsyncClient(timeout=5.0) as client:
        resp = await client.get(
            f'{BASE_URL}/validate/{country}/{vat}',
            headers={'Authorization': f'Bearer {os.environ["TAXID_API_KEY"]}'},
        )
        resp.raise_for_status()
        return resp.json()


# Usage in an async Django view:
# from rest_framework.views import APIView
# class AsyncB2BView(APIView):
#     async def post(self, request):
#         result = await validate_vat_async(request.data['vat_number'])
#         ...

import pytest
from unittest.mock import patch, MagicMock
from vat_validation.client import validate_vat_safe


def make_response(data):
    m = MagicMock()
    m.json.return_value = data
    m.raise_for_status = MagicMock()
    return m


@patch('vat_validation.client.requests.get')
def test_active_vat_number(mock_get):
    mock_get.return_value = make_response({
        'valid': True, 'status': 'active',
        'company_name': 'Acme GmbH', 'address': 'Berlin, DE',
        'country_code': 'DE', 'cached': False, 'request_id': 'r1',
    })
    result = validate_vat_safe('DE123456789')
    assert result['valid'] is True
    assert result['company_name'] == 'Acme GmbH'


@patch('vat_validation.client.requests.get')
def test_inactive_vat_number(mock_get):
    mock_get.return_value = make_response({
        'valid': False, 'status': 'inactive',
        'company_name': None, 'address': None,
        'country_code': 'DE', 'cached': False, 'request_id': 'r2',
    })
    result = validate_vat_safe('DE000000000')
    assert result['status'] == 'inactive'
    assert result['valid'] is False


@patch('vat_validation.client.requests.get', side_effect=Exception('timeout'))
def test_network_error_returns_unavailable(mock_get):
    result = validate_vat_safe('DE123456789')
    assert result['status'] == 'service_unavailable'
    assert result['valid'] is False