Async Third-Party API Integration with Clean Architecture

A guide for Python backend coding interviews covering: async design, API exploration, schema modeling, error taxonomy, retries, and clean separation of concerns with FastAPI.

Interview Context

Typical scenario: A third-party service returns data asynchronously. You need to:

1. Discover the API shape when you don’t know it
2. Handle auth errors (e.g. 401)
3. Define models and map external data to your domain
4. Decide which errors to retry and which not
5. Integrate with FastAPI using clean architecture

Phase 1: “We don’t know what comes from the API — how do we find out?”

Make a raw request and inspect the response.

import httpx
import json

async def explore_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        print(response.status_code)
        print(response.headers)
        print(json.dumps(response.json(), indent=2))  # pretty-print the shape

You’re looking at: status code, content-type header, and the JSON shape/keys.

Phase 2: “We got 401 — what do we do?”

401 = Unauthorized. Two parts:

1. Understand the cause

• Missing Authorization header
• Wrong or expired token
• Wrong auth scheme (Bearer vs Basic vs API key)

2. Handle it in code

class APIAuthError(Exception):
    pass

class APIRateLimitError(Exception):
    pass

class APIClient:
    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url

    async def get(self, endpoint: str) -> dict:
        headers = {"Authorization": f"Bearer {self.api_key}"}
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{self.base_url}{endpoint}", headers=headers)

            if response.status_code == 401:
                raise APIAuthError("Invalid or expired API key")
            if response.status_code == 429:
                raise APIRateLimitError("Rate limit hit — back off and retry")

            response.raise_for_status()  # handles 4xx/5xx generically
            return response.json()

Phase 3: Defining the Model (Pydantic)

Once you see the response shape, convert it into a typed model.

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

# Raw API response looks like:
# {"id": 123, "user_name": "alice", "score": 98.5, "created_at": "2024-01-01T00:00:00Z"}

class ThirdPartyUserResponse(BaseModel):
    """Model matching exactly what the API returns."""
    id: int
    user_name: str
    score: float
    created_at: datetime
    metadata: Optional[dict] = None

class User(BaseModel):
    """Your internal domain model — decoupled from the API."""
    id: int
    name: str
    score: float
    created_at: datetime

def map_to_domain(raw: ThirdPartyUserResponse) -> User:
    """Adapter layer: external → internal."""
    return User(
        id=raw.id,
        name=raw.user_name,  # rename
        score=raw.score,
        created_at=raw.created_at,
    )

Key point to say aloud: “I separate the API model from the domain model so that if the API changes, only the adapter breaks — not the rest of the app.”

API Model vs Domain Model — Why the Separation Matters

This is central to clean architecture. The adapter is the only boundary that knows both worlds.

The Problem Without Separation

# BAD: using raw API data directly in your business logic
@router.get("/users/{user_id}")
async def get_user(user_id: int):
    response = await httpx.get(f"https://api.example.com/users/{user_id}")
    data = response.json()

    return {
        "name": data["user_name"],        # raw API field
        "score": data["score_value"],     # raw API field
        "joined": data["created_at"],     # raw API field
    }

If the third-party renames user_name → username or score_value → score, every place that touches this data breaks.

The Solution: Two Models + One Adapter

from pydantic import BaseModel
from datetime import datetime

# ─────────────────────────────────────────
# LAYER 1: API Model
# Mirrors EXACTLY what the third-party sends.
# This is the "contract" with the outside world.
# ─────────────────────────────────────────
class ThirdPartyUserResponse(BaseModel):
    user_name: str
    score_value: float
    created_at: str
    internal_ref_id: str  # their weird internal field you don't care about

# ─────────────────────────────────────────
# LAYER 2: Domain Model
# YOUR language, YOUR naming, YOUR types.
# The rest of your app only ever sees this.
# ─────────────────────────────────────────
class User(BaseModel):
    name: str
    score: float
    created_at: datetime  # properly typed, not a raw string

# ─────────────────────────────────────────
# LAYER 3: Adapter (the translator)
# Knows about BOTH models. Lives at the boundary.
# This is the ONLY place that knows the API's quirks.
# ─────────────────────────────────────────
def to_user(raw: ThirdPartyUserResponse) -> User:
    return User(
        name=raw.user_name,
        score=raw.score_value,
        created_at=datetime.fromisoformat(raw.created_at),
        # internal_ref_id is dropped — we don't need it
    )

When the API Changes

Third-party renames user_name → username and score_value → score:

# You change ONE place only — the API model + adapter
class ThirdPartyUserResponse(BaseModel):
    username: str   # updated
    score: float    # updated
    created_at: str

def to_user(raw: ThirdPartyUserResponse) -> User:
    return User(
        name=raw.username,
        score=raw.score,
        created_at=datetime.fromisoformat(raw.created_at),
    )

Your domain model User is unchanged. Your service layer is unchanged. Your routes are unchanged. Your tests for business logic are unchanged.

The blast radius of the change is contained to the adapter.

Visual Mental Model

Third-Party API
      │
      │  raw JSON  {"user_name": "alice", "score_value": 98.5}
      ▼
ThirdPartyUserResponse      ← API Model (mirrors external world)
      │
      │  adapter / mapping function
      ▼
    User                    ← Domain Model (your internal world)
      │
      ├── Service Layer
      ├── Business Logic
      ├── Database
      └── FastAPI Response

The adapter is the only thing that crosses the boundary.
Everything to the right speaks your language, not the API's.

Switching Providers Entirely

# Provider A's model
class ProviderAResponse(BaseModel):
    user_name: str
    score_value: float

# Provider B's model (completely different shape)
class ProviderBResponse(BaseModel):
    full_name: str
    rating: float
    tier: str  # "gold", "silver" — not even a number

# Two adapters — same output
def from_provider_a(raw: ProviderAResponse) -> User:
    return User(name=raw.user_name, score=raw.score_value)

def from_provider_b(raw: ProviderBResponse) -> User:
    score = {"gold": 100.0, "silver": 75.0}.get(raw.tier, 0.0)
    return User(name=raw.full_name, score=score)

# Your entire app still only uses User — zero changes needed downstream

One-line summary for the interview: “The domain model is stable — it’s your app’s language. The API model is volatile — it’s someone else’s language. The adapter translates between them so volatility never leaks inward.”

Phase 4: Error Handling — What to Retry vs What Not To

Rule: Retry transient errors. Do not retry client/auth errors.

import asyncio

class APIServerError(Exception):
    pass

class APIClientError(Exception):
    pass

async def fetch_with_retry(client: APIClient, endpoint: str, retries: int = 3) -> dict:
    for attempt in range(retries):
        try:
            return await client.get(endpoint)

        except APIRateLimitError:
            # 429 — wait and retry (exponential backoff)
            wait = 2 ** attempt
            await asyncio.sleep(wait)

        except APIServerError:
            # 5xx — server-side, might recover, retry
            if attempt == retries - 1:
                raise
            await asyncio.sleep(1)

        except APIAuthError:
            # 401 — our fault (config issue), don't retry
            raise

        except APIClientError:
            # 400, 404, 422 — bad request, don't retry
            raise

    raise Exception("Max retries exceeded")

Phase 5: Clean Architecture + FastAPI Integration

Business logic lives in services. FastAPI routers stay thin.

# services/user_service.py — business logic, no FastAPI dependency
class UserService:
    def __init__(self, api_client: APIClient):
        self.api_client = api_client

    async def get_user(self, user_id: int) -> User:
        raw_data = await fetch_with_retry(self.api_client, f"/users/{user_id}")
        raw = ThirdPartyUserResponse(**raw_data)
        return map_to_domain(raw)

# routers/users.py — FastAPI layer, thin
from fastapi import APIRouter, Depends, HTTPException

router = APIRouter()

# get_api_client: provide APIClient via DI (e.g. from settings, singleton)
def get_user_service(api_client: APIClient = Depends(get_api_client)) -> UserService:
    return UserService(api_client=api_client)

@router.get("/users/{user_id}", response_model=User)
async def get_user_route(user_id: int, service: UserService = Depends(get_user_service)):
    try:
        return await service.get_user(user_id)
    except APIAuthError:
        raise HTTPException(status_code=502, detail="Upstream auth failed")
    except APIRateLimitError:
        raise HTTPException(status_code=429, detail="Try again later")
    except Exception:
        raise HTTPException(status_code=500, detail="Internal error")

Key Phrases to Drop in the Interview

• “I’d use Pydantic for validation so bad data from the API fails fast at the boundary.”
• “I separate the API model from the domain model — the adapter pattern protects internal logic from external changes.”
• “Retry only idempotent and transient failures — never retry 401/400 as it’ll just fail again.”
• “In clean architecture, FastAPI routers are thin — business logic lives in services, not in route handlers.”
• “I’d use httpx.AsyncClient as a singleton (via DI) to reuse the connection pool.”
• “The domain model is stable — it’s your app’s language. The API model is volatile — it’s someone else’s. The adapter translates so volatility never leaks inward.”

Interview angle

• “Walk me through your layering.” - transport (FastAPI router, no business logic), service or use case (orchestration, the only layer that knows the business rules), client or gateway (HTTP, retries, auth, provider-specific quirks), and models split into external DTOs versus domain objects. The rule to state explicitly is that dependencies point inward: the service knows an interface, not httpx.
• “Why separate the external schema from the domain model?” - so a provider adding, renaming or removing a field changes one mapping function, not your whole codebase. It is also where you enforce that a partial or malformed upstream response never reaches business logic.
• “How do you test this?” - the service against a fake client with no network at all, the client against recorded responses or a mock transport, and one integration test end to end. If testing the service requires patching httpx, the boundary is in the wrong place.
• “Is this over-engineering for one endpoint?” - for one endpoint that never changes, yes, and say so. It earns its keep at the second provider, the first retry policy, or the first time you need to test business logic without a network.