Designing an external API integration

The take-home and live-coding exercise this repo’s owner keeps meeting: “integrate three external sources into one internal API.” This file is the structural answer; 02_schema_and_models.md covers the data, and ../02_resilience/ covers failure.

The layering that answers the question

      HTTP layer (FastAPI routers)          <- transport, no business logic
                  |
          Service / use case                <- orchestration, domain rules
                  |
     Port (Protocol)  <- the seam that makes this testable
                  |
   Adapter per provider (HotelAdapter, ...) <- HTTP, auth, their schema
                  |
              External API

The load-bearing idea: the service depends on a Protocol, not on any provider’s client. Adding a fourth provider means writing a fourth adapter and registering it — no change to the service.

from typing import Protocol

class InventorySource(Protocol):
    name: str
    async def search(self, query: SearchQuery) -> list[Offer]: ...

class HotelAdapter:
    name = "hotels"
    def __init__(self, http: httpx.AsyncClient, cfg: HotelConfig): ...
    async def search(self, query: SearchQuery) -> list[Offer]:
        raw = await self._get("/v2/search", params=self._to_their_params(query))
        return [self._to_offer(item) for item in raw["results"]]   # their shape -> ours

The adapter is where their world stops. Their field names, their status codes, their pagination and their auth scheme all terminate at _to_offer. Nothing above that line knows the provider exists.

Interviewers probe this by asking “now add a provider that paginates differently” — if your service changes, the seam is in the wrong place.

What to establish before writing code

Say these out loud; it demonstrates you’ve done this before:

Ask about rate limits and idempotency early. Both change the architecture, and both are expensive to retrofit.

The client layer

One place per provider owning HTTP concerns:

class BaseAPIClient:
    def __init__(self, base_url: str, timeout: httpx.Timeout, breaker: CircuitBreaker):
        self._client = httpx.AsyncClient(base_url=base_url, timeout=timeout)
        self._breaker = breaker

    async def request(self, method: str, path: str, **kw) -> dict:
        async with self._semaphore:                 # per-provider concurrency cap
            resp = await self._breaker.call(self._client.request, method, path, **kw)
        if resp.status_code >= 400:
            raise self._map_error(resp)             # their errors -> your exceptions
        return resp.json()

Three things belong here and nowhere else: timeout and retry policy, the circuit breaker, and error mapping. Scattering these across service methods makes total latency impossible to reason about.

Reuse one AsyncClient per provider for the lifetime of the app. Creating a client per request throws away connection pooling and TLS session reuse, and it’s a common and expensive mistake. Wire it as a singleton — see ../../backend/13_architecture_design/07_di_pattern_library.md.

Error mapping

Their failures become your domain exceptions at the adapter boundary:

def _map_error(self, resp: httpx.Response) -> Exception:
    match resp.status_code:
        case 401 | 403: return AuthFailure(self.name)
        case 404:       return NotFound(self.name)
        case 429:       return RateLimited(self.name, retry_after=resp.headers.get("Retry-After"))
        case 400 | 422: return InvalidRequest(self.name, detail=resp.text)
        case _ if resp.status_code >= 500: return UpstreamUnavailable(self.name)
        case _:         return UnexpectedResponse(self.name, resp.status_code)

Why bother: the service layer decides what to do about a failure, and it can only do that if failures are expressed in its own vocabulary. httpx.HTTPStatusError leaking into a use case couples your business logic to a transport library.

Keep the provider name on the exception — with three sources, “which one failed” is the first question during an incident.

Fan-out with partial success

The core of the three-sources exercise:

async def search_all(self, query: SearchQuery) -> SearchResult:
    async def guarded(src: InventorySource) -> tuple[str, list[Offer] | None]:
        try:
            async with asyncio.timeout(self.per_source_budget):
                return src.name, await src.search(query)
        except Exception:
            logger.warning("source failed", extra={"source": src.name}, exc_info=True)
            return src.name, None                    # failure is a value, not a raise

    pairs = await asyncio.gather(*(guarded(s) for s in self.sources))
    offers = [o for _, res in pairs if res for o in res]
    failed = [name for name, res in pairs if res is None]
    return SearchResult(offers=self._rank(offers), degraded_sources=failed)

The decisions to defend:

• Concurrent, not sequential. Total latency is the slowest source, not the sum.
• Per-source timeout below the overall request budget, so one slow provider can’t consume it.
• Each source’s failure is caught inside its own task — returning a value rather than raising, so one failure doesn’t abort the others.
• degraded_sources in the response, so the caller can say “results may be incomplete”. Silent partial results are worse than stated ones.

See ../02_resilience/03_fallbacks_and_degradation.md.

Configuration and secrets

class ProviderSettings(BaseSettings):
    base_url: HttpUrl
    api_key: SecretStr                      # SecretStr: won't print in logs or reprs
    timeout_s: float = 10.0
    max_concurrency: int = 10

    model_config = SettingsConfigDict(env_prefix="HOTELS_")

Per-provider settings from the environment, SecretStr so a stray log line doesn’t leak a key, and no hardcoded URLs — you need to point at a sandbox. See ../04_secrets_config/.

Testing

Save one real response per endpoint as a fixture. Hand-written fixtures encode what you think the API returns; recorded ones encode what it does, including the nulls and the field you assumed was always present.

The Protocol seam is what makes service tests fast: no HTTP, no mocking library, just a fake class.

Interview angle

• “How would you integrate three external sources behind one API?” — an adapter per provider implementing a shared Protocol, a service that depends on the Protocol, and normalisation into an internal model at the adapter boundary. Adding a fourth provider is a new adapter and a registration, with no service change.
• “What do you ask before designing it?” — auth and token lifetime, rate limits, pagination style, p99 latency, sandbox availability, versioning policy, and idempotency support. Rate limits and idempotency change the architecture and are expensive to retrofit.
• “One provider is slow. What happens?” — per-source timeout under the overall budget, calls run concurrently, and each failure is caught inside its own task so it becomes a value rather than aborting the fan-out. Return partial results with the failed sources named.
• “Where do you put retry and circuit-breaker logic?” — the client layer, once per provider. Scattered across service methods you can’t reason about total latency or retry amplification.
• “Why map their errors to your own exceptions?” — the service layer decides what to do about a failure and needs failures in its own vocabulary. Leaking HTTPStatusError into a use case couples business logic to a transport library. Keep the provider name on the exception.
• “How do you test it?” — recorded real payloads for adapter tests, fake Protocol implementations for service tests, contract tests replaying saved responses to catch provider drift, and scheduled sandbox runs end to end.