# Recipe — Provider seams with `mgf.common.providers`

> **Audience:** consumers integrating external services (telematics
> SDKs, payment gateways, mapping APIs, notification providers,
> object-storage clients) where every external call can fail with
> SDK-specific exceptions that domain code shouldn't see.
> **No optional extra** — pure stdlib + `mgf.common.exceptions`.

## At a glance

```python
from mgf.common.providers import ProviderABC, translate_at_seam
from mgf.common.exceptions import AppError, ProviderError, ResourceNotFoundError

import telematics_sdk # any provider SDK

class TelematicsProvider(ProviderABC):
    def __init__(self, client: telematics_sdk.Client) -> None:
        self._client = client

    def translate_exception(self, exc: Exception) -> AppError:
        if isinstance(exc, telematics_sdk.NotFound):
            return ResourceNotFoundError(name=str(exc))
        if isinstance(exc, telematics_sdk.RateLimited):
            return ProviderError(f"telematics rate limited: {exc}")
        if isinstance(exc, telematics_sdk.AuthFailed):
            return ProviderError(f"telematics auth failed: {exc}")
        return ProviderError(f"telematics call failed: {exc}")

    @translate_at_seam
    async def get_vehicle(self, vehicle_id: str) -> dict:
        return await self._client.vehicles.get(vehicle_id)

    @translate_at_seam
    async def list_devices(self, fleet_id: str) -> list[dict]:
        return await self._client.devices.list(fleet_id=fleet_id)
```

That's the entire boilerplate replacement. Domain code that calls
`provider.get_vehicle(...)` only sees `AppError` subclasses;
SDK-specific exceptions never leak past the seam.

## What the decorator does

For every wrapped method (sync or async):

1. **`AppError` subclasses pass through unchanged.** Domain code
   that raises typed errors inside a provider method (e.g., a
   business-rule check before the external call) is never re-translated.
2. **Every other `Exception` is translated.** The decorator catches
   it, calls `self.translate_exception(exc)`, re-raises the returned
   `AppError` with the original as `__cause__`.
3. **`BaseException` subclasses pass through unchanged.**
   `KeyboardInterrupt`, `SystemExit`, `GeneratorExit`, `MemoryError`,
   and `asyncio.CancelledError` are NOT caught — catching them
   breaks Python's signal model and asyncio's cooperative cancellation.

The decorator detects sync vs async at decoration time via
`inspect.iscoroutinefunction`. The same decorator works on both;
no separate `translate_at_seam_async` exists.

## Why this exists

Without a primitive, every L2 consumer with provider seams ships
the same shape:

```python
# Without mgf.common.providers — repeat per method:
async def get_vehicle(self, vehicle_id: str) -> dict:
    try:
        return await self._client.vehicles.get(vehicle_id)
    except telematics_sdk.NotFound as e:
        raise ResourceNotFoundError(name=str(e)) from e
    except telematics_sdk.RateLimited as e:
        raise ProviderError(f"rate limited: {e}") from e
    except telematics_sdk.AuthFailed as e:
        raise ProviderError(f"auth failed: {e}") from e
    except telematics_sdk.SdkError as e:
        raise ProviderError(f"sdk error: {e}") from e
```

Multiply by ~20 methods per provider × ~5 providers = a lot of
boilerplate where every method has its own translation ladder.
Mistakes leak SDK-specific exceptions into domain code, breaking
the closed-box rule.

`@translate_at_seam` collapses this to:

```python
# With the decorator — one ladder per provider, not per method:
def translate_exception(self, exc: Exception) -> AppError:
    # The whole ladder lives here, ONCE.

@translate_at_seam
async def get_vehicle(self, vehicle_id: str) -> dict:
    return await self._client.vehicles.get(vehicle_id)
```

## The two patterns

### Pattern 1: subclass `ProviderABC` (recommended)

```python
class StripeProvider(ProviderABC):
    def translate_exception(self, exc: Exception) -> AppError:
        if isinstance(exc, stripe.error.CardError):
            return ResourceNotFoundError(name="card")
        if isinstance(exc, stripe.error.RateLimitError):
            return ProviderError(f"stripe rate limited: {exc}")
        return ProviderError(f"stripe call failed: {exc}")

    @translate_at_seam
    def charge(self, amount_cents: int, source: str) -> dict:
        return stripe.Charge.create(amount=amount_cents, source=source)
```

ABC enforcement: `StripeProvider()` fails at construction time if
you forget to implement `translate_exception` — no silent surprises
in production.

### Pattern 2: duck-typed (no subclass needed)

```python
class _LegacyProvider:
    """Existing class you don't want to make a ProviderABC subclass
    yet (refactor in flight, mixin conflicts, etc.)."""

    def translate_exception(self, exc: Exception) -> AppError:
        return ProviderError(f"upstream call failed: {exc}")

    @translate_at_seam
    def fetch(self, key: str) -> str:
        return self._raw_call(key)
```

The decorator only requires that `self.translate_exception` exists
at call time — no `isinstance(self, ProviderABC)` check. Useful for
gradual migration from existing typed-error patterns.

If `translate_exception` is missing, the decorator raises a
`TypeError` with an actionable message:

```
TypeError: @translate_at_seam used on _LegacyProvider, which has
no translate_exception method. Either subclass
mgf.common.providers.ProviderABC or implement the method on the
class.
```

## Designing your `translate_exception`

The translator is the most important piece. Two design moves keep
it maintainable:

### Move 1: route SDK exceptions in MRO order

```python
def translate_exception(self, exc: Exception) -> AppError:
    # Most-specific FIRST. Catch-all LAST.
    if isinstance(exc, sdk.NotFound):           # specific
        return ResourceNotFoundError(name=str(exc))
    if isinstance(exc, sdk.PermissionDenied):    # specific
        return ProviderError(f"permission denied: {exc}")
    if isinstance(exc, sdk.RateLimited):         # specific
        return ProviderError(f"rate limited: {exc}")
    if isinstance(exc, sdk.SdkError):            # base — catch-all
        return ProviderError(f"upstream failure: {exc}")
    # Anything else (bug? OS error? network glitch unwrapped?):
    return ProviderError(f"unexpected provider failure: {exc}")
```

The final fallback uses a generic catch-all so domain code never
sees `OSError`, `ConnectionError`, or other non-SDK exceptions
leaking from a provider method. Choose `ProviderError` (a category)
for the catch-all so the CLI exit-code translator and FastAPI
status-map both route it correctly.

### Move 2: don't drop information

The original exception is preserved as `__cause__`. Use it in your
translator's message AND keep the chain intact:

```python
def translate_exception(self, exc: Exception) -> AppError:
    return ProviderError(
        f"telematics call failed: {type(exc).__name__}: {exc}"
    )
    # The decorator does `raise translated from exc`, so:
    # caller_traceback
    # ↓
    # ProviderError("telematics call failed: NotFound: vehicle/abc")
    #   __cause__: telematics_sdk.NotFound("vehicle/abc")
    #     __cause__: <whatever the SDK chained off>
```

The full chain is available in logs, debugger, and crash reports.

## Common patterns

### Composing with `mgf.common.http`

`AsyncHttpClient` raises `HttpClientError` on transport failures.
A provider built on top of `AsyncHttpClient` can translate
`HttpClientError` to a domain-specific `AppError`:

```python
from mgf.common.http import AsyncHttpClient, HttpClientError
from mgf.common.providers import ProviderABC, translate_at_seam

class WeatherProvider(ProviderABC):
    def __init__(self, client: AsyncHttpClient) -> None:
        self._client = client

    def translate_exception(self, exc: Exception) -> AppError:
        if isinstance(exc, HttpClientError):
            return ProviderError(f"weather API failed: {exc}")
        return ProviderError(f"unexpected weather failure: {exc}")

    @translate_at_seam
    async def get_forecast(self, lat: float, lon: float) -> dict:
        response = await self._client.get(
            "/forecast",
            params={"lat": lat, "lon": lon},
        )
        response.raise_for_status()
        return response.json()
```

The chain is now: domain → `ProviderError` → `HttpClientError` →
`httpx.HTTPError`. Three layers; debugging walks `__cause__` to
find the root.

### Composing with the FastAPI translation middleware

The `AppError` subclasses raised by `@translate_at_seam` are
already what `mgf.common.fastapi.ExceptionTranslationMiddleware`
expects:

```python
@app.get("/vehicles/{vehicle_id}")
async def read_vehicle(
    vehicle_id: str,
    provider: Annotated[TelematicsProvider, Depends(get_telematics)],
) -> dict:
    # If the SDK raises NotFound, the decorator translates it to
    # ResourceNotFoundError, the middleware maps to 404, the client
    # gets a typed JSON error body. No try/except in the handler.
    return await provider.get_vehicle(vehicle_id)
```

End-to-end typed-error flow: SDK exception → `AppError` (at
decorator) → JSON response (at middleware) → consumer-app stays
focused on the happy path.

### Testing — assert the translation, not the SDK

```python
import pytest
from unittest.mock import AsyncMock, MagicMock

from myapp.providers.telematics import TelematicsProvider
from mgf.common.exceptions import ResourceNotFoundError

@pytest.fixture
def fake_sdk_client():
    client = MagicMock()
    client.vehicles.get = AsyncMock()
    return client

async def test_get_vehicle_translates_not_found(fake_sdk_client):
    fake_sdk_client.vehicles.get.side_effect = telematics_sdk.NotFound("v123")
    provider = TelematicsProvider(fake_sdk_client)

    with pytest.raises(ResourceNotFoundError) as info:
        await provider.get_vehicle("v123")

    # The original SDK exception is preserved on __cause__.
    assert isinstance(info.value.__cause__, telematics_sdk.NotFound)
```

Tests assert at the seam — the decorator is the contract, not the
SDK. Mock the SDK; assert the typed error.

## Anti-patterns

- **Don't catch `AppError` inside a wrapped method and re-raise as
  something else.** The decorator already lets `AppError` pass
  through. If you want a different `AppError`, raise it directly —
  the decorator won't double-wrap.
- **Don't use `@translate_at_seam` on methods that don't make
  external calls.** The decorator adds try/except overhead per call.
  For pure-Python helpers that already raise `AppError`, skip it.
- **Don't return non-`AppError` values from `translate_exception`.**
  The decorator runtime-checks the return type and raises `TypeError`
  if it's wrong. Bugs here fail loud, not silent.
- **Don't catch `BaseException` in your provider method.** If you
  do, you're swallowing `KeyboardInterrupt` and `asyncio.CancelledError`
  — neither the decorator nor the runtime can recover gracefully.
  Let those bubble.
- **Don't make `translate_exception` async.** It runs inside the
  decorator's `except` clause; an async translator would require
  awaiting from a sync context, which doesn't compose. Keep it
  synchronous (translation is pure dispatch — no I/O needed).

## See also

- `docs/public/02_exceptions.md` — `AppError` hierarchy + when to
  use each category.
- `docs/public/recipes/http.md` — `AsyncHttpClient` for the
  provider's transport layer.
- `docs/public/recipes/fastapi.md` — `ExceptionTranslationMiddleware`
  closes the typed-error loop at the HTTP boundary.
- `docs/public/recipes/cli.md` — `run_and_translate` closes the
  typed-error loop at the CLI exit-code boundary.
