diff --git a/README.md b/README.md index 98541cf..279d690 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,9 @@ retry traffic by default), body replayability and the remaining deadline. `AdapterDeps(runtime=...)` - otherwise breaker state dies with every request. With `clientwright[dishka]`, `contrib.dishka.ClientwrightProvider` does both: APP-scope runtime and a generator provide that closes the client in `finally`. +Several upstreams in one container are Dishka components — one +`ClientwrightProvider("httpx", config, component="github-api", client_type=httpx.AsyncClient)` +per upstream, injected as `Annotated[httpx.AsyncClient, FromComponent("github-api")]`. ## Deadline budgets diff --git a/clientwright/contrib/dishka.py b/clientwright/contrib/dishka.py index f3d2b07..9632877 100644 --- a/clientwright/contrib/dishka.py +++ b/clientwright/contrib/dishka.py @@ -15,12 +15,22 @@ ClientwrightProvider("httpx", config), ..., ) - handle = await container.get(ClientHandle) + handle = await container.get(ClientHandle[Any]) client: httpx.AsyncClient = handle.client -Known limitation: the ``http_client_circuit_state`` gauge is wired when the -ADAPTER builds the runtime; a runtime built here (to be shared) has no -telemetry listener, so state-change events are not exported as a gauge. +Several upstreams in one container are Dishka components: one provider per +upstream, each in its own component, and ``client_type`` for injection sites +that want the native client rather than the handle:: + + container = make_async_container( + ClientwrightProvider("httpx", login_config, component="github-oauth", client_type=httpx.AsyncClient), + ClientwrightProvider("httpx", api_config, component="github-api", client_type=httpx.AsyncClient), + ) + api = await container.get(httpx.AsyncClient, component="github-api") + +The runtime built here carries the ``http_client_circuit_state`` listener the +way a runtime built by an adapter does; a runtime passed in through +``deps.runtime`` is handed back as it came. """ from __future__ import annotations @@ -38,29 +48,56 @@ from ..core.contracts.adapter import AdapterDeps, default_deps from ..core.plan import ClientHandle, ClientRuntime from ..core.registry import resolve_adapter +from ..core.telemetry.emitter import ClientTelemetry class ClientwrightProvider(Provider): """One async native client with an APP-scope runtime and guaranteed close. - For several upstreams, instantiate one provider per upstream in separate - containers, or subclass and add typed aliases (e.g. a provider returning - ``httpx.AsyncClient`` from the handle) for ergonomic injection. + ``component`` puts everything the provider gives out in a Dishka component, + which is how several upstreams share one container: resolve with + ``container.get(..., component="github-api")`` or inject with + ``Annotated[..., FromComponent("github-api")]``. ``client_type`` also + provides the native client under that type (``httpx.AsyncClient``, ...), + the same object the handle holds. """ scope = Scope.APP - def __init__(self, adapter: str, config: ClientConfig, deps: AdapterDeps | None = None) -> None: - super().__init__() + def __init__( + self, + adapter: str, + config: ClientConfig, + deps: AdapterDeps | None = None, + *, + component: str | None = None, + client_type: type[Any] | None = None, + ) -> None: + super().__init__(component=component) self._adapter = adapter self._config = config self._deps = deps or default_deps() + if client_type is not None: + self.provide(self._native_client, provides=client_type) @provide def client_runtime(self) -> ClientRuntime: if self._deps.runtime is not None: return self._deps.runtime - return ClientRuntime.for_config(self._config, clock=self._deps.clock) + # An adapter wires the circuit-state gauge only to a runtime it builds + # itself; this runtime is built here, so the wiring happens here. + adapter = resolve_adapter(self._adapter)() + telemetry = ClientTelemetry( + service=self._config.service_name, + adapter=adapter.name, + seam=adapter.capabilities.seam, + config=self._config.observability, + metrics=self._deps.metrics, + tracer=None, + ) + return ClientRuntime.for_config( + self._config, clock=self._deps.clock, circuit_listener=telemetry.circuit_state_changed + ) @provide async def client_handle(self, runtime: ClientRuntime) -> AsyncIterator[ClientHandle[Any]]: @@ -76,5 +113,8 @@ async def client_handle(self, runtime: ClientRuntime) -> AsyncIterator[ClientHan elif handle.close is not None: handle.close() + def _native_client(self, handle: ClientHandle[Any]) -> Any: + return handle.client + __all__ = ["ClientwrightProvider"] diff --git a/docs/agents.md b/docs/agents.md index b885ffd..04e5a1d 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -346,7 +346,7 @@ task-local and thread-local, inherited by tasks started inside it, invisible to | | `use_budget(budget)` | context manager installing a budget; `use_budget(None)` detaches | | | `current_budget()` | the installed budget or `None` | | | `DeadlineBudgetProtocol` | structural: `remaining() -> float`, `expired() -> bool` | -| `clientwright.contrib.dishka` | `ClientwrightProvider(adapter, config, deps=None)` | `Scope.APP` provider giving `ClientRuntime` and a `ClientHandle` closed in `finally` | +| `clientwright.contrib.dishka` | `ClientwrightProvider(adapter, config, deps=None, *, component=None, client_type=None)` | `Scope.APP` provider giving `ClientRuntime` (circuit-state listener wired), a `ClientHandle` closed in `finally`, and the native client under `client_type`; `component` is one per upstream, resolved with `FromComponent(...)` / `component=` | | `clientwright.contrib.settings` | `BaseClientSettings` | `ClientConfig` as a pydantic `BaseModel` — nest it in your own `BaseSettings` (`env_nested_delimiter="__"`); `.to_config(service_name)` returns the `ClientConfig`. Never a `BaseSettings` itself, so a bare `BASE_URL` cannot reach a section | | | `BaseTimeoutSettings`, `BasePoolSettings`, `BaseRetrySettings`, `BaseCircuitBreakerSettings`, `BaseTlsSettings`, `BaseProxySettings`, `BaseObservabilitySettings` | the sub-configs, same field names and defaults, each with `.to_config()`. A knob left unset stays `UNSET`; `null` is the explicit "unbounded"; `retry: None = None` in a subclass disables a section; `url_masker` is not a field | | `clientwright.core.testing` | `OriginServer()` | in-process fault-injecting origin on an ephemeral localhost port | @@ -371,7 +371,7 @@ Metric names and label sets are a frozen wire contract in | `http_client_attempts_total` | counter | once per **physical attempt** (not under `RetryMode.DELEGATED`) | | `http_client_attempt_duration_seconds` | histogram | same | | `http_client_inflight` | gauge | ±1 around each logical call | -| `http_client_circuit_state` | gauge | on a breaker transition, **only when the adapter built the runtime** | +| `http_client_circuit_state` | gauge | on a breaker transition, **only when the adapter or `ClientwrightProvider` built the runtime** | | `http_client_redirect_hops_total` | counter | per followed hop | | `http_client_retry_skipped_total` | counter | `reason=method|non_replayable|deadline|budget` | | `http_client_uninstrumented_calls_total` | counter | aiohttp only: a request that bypassed the middleware | @@ -435,9 +435,9 @@ What each one will not do: the per-origin limiters. Build it once, share it through `AdapterDeps(runtime=...)` across request-scoped clients. A runtime per request is a breaker with no memory. 2. **A runtime you build yourself has no circuit-state gauge.** The - `http_client_circuit_state` listener is wired only when the *adapter* builds the runtime. - Passing `deps.runtime` — including through `ClientwrightProvider` — keeps the breaker - working and loses that one gauge. + `http_client_circuit_state` listener is wired when the *adapter* or `ClientwrightProvider` + builds the runtime. Passing `deps.runtime` keeps the breaker working and loses that one + gauge, unless you built it with `ClientRuntime.for_config(config, circuit_listener=...)`. 3. **`UNSET` is not `None`.** `UNSET` defers to the adapter's native default and says so in the report; `None` means explicitly unbounded. Both differ from a number. 4. **`retryable_kinds` cannot retry a status.** The policy checks `retryable_status` first, @@ -647,7 +647,7 @@ Fetch a page when the task is the one named beside it. | [Proxies and TLS](guide/proxies-tls.md) | mTLS, private CAs, explicit and environment proxies | | [Native passthrough](guide/native-options.md) | a knob `ClientConfig` does not cover | | [Capability honesty](guide/capabilities.md) | reading a report, comparing adapters before a migration | -| [Dependency injection](guide/dishka.md) | wiring the runtime and the client lifecycle in a container | +| [Dependency injection](guide/dishka.md) | wiring the runtime and the client lifecycle in a container, one component per upstream | | [Deadline budgets](guide/deadline-budget.md) | propagating the inbound request's remaining time | | [Testing your service](guide/testing.md) | `OriginServer`, `RecordingMetrics`, what to mock instead | | [Choosing an adapter](adapters/index.md) | picking one, or planning a swap | diff --git a/docs/guide/dishka.md b/docs/guide/dishka.md index 473a656..76d465a 100644 --- a/docs/guide/dishka.md +++ b/docs/guide/dishka.md @@ -27,8 +27,7 @@ container = make_async_container( ) ``` -Resolve the handle (or alias the client type in your own provider for ergonomic -injection): +Resolve the handle: ```python from typing import Any @@ -41,9 +40,60 @@ handle = await container.get(ClientHandle[Any]) client: httpx.AsyncClient = handle.client ``` +Or ask for the client itself. `client_type=` makes the provider also provide +the native client under that type — the same object the handle holds — which +is what an injection site usually wants: + +```python +container = make_async_container( + ClientwrightProvider("httpx", config, client_type=httpx.AsyncClient), +) +client = await container.get(httpx.AsyncClient) +``` + The provider is `Scope.APP`: one runtime, one client, built once. The client is yielded from a generator provide whose `finally` calls `aclose()` — closing the -container closes the client, deterministically. +container closes the client, deterministically. A request scope opening and +closing does not touch it; nothing is parked on a request-scoped exit stack. + +## Several upstreams + +One provider serves one upstream. A service with several registers one provider +per upstream, each in its own Dishka component, and names the component at the +injection site: + +```python +from typing import Annotated + +import httpx +from dishka import FromComponent, make_async_container + +from clientwright.contrib.dishka import ClientwrightProvider + +container = make_async_container( + ClientwrightProvider("httpx", login_config, component="github-oauth", client_type=httpx.AsyncClient), + ClientwrightProvider("httpx", api_config, component="github-api", client_type=httpx.AsyncClient), +) + + +class GitHubOAuthProvider: + def __init__( + self, + login_client: Annotated[httpx.AsyncClient, FromComponent("github-oauth")], + api_client: Annotated[httpx.AsyncClient, FromComponent("github-api")], + ) -> None: ... + + +api = await container.get(httpx.AsyncClient, component="github-api") +``` + +Dishka resolves a component's dependencies inside that component, so each +provider's client is built on the runtime of its own component: two upstreams +are two breakers, two budgets, two clients, each closed when the container +closes. The cost is that every injection site names the upstream — +`FromComponent("github-api")` or `component="github-api"` — the same trade-off +`grpc_client_kit.dishka` makes. A single upstream stays in the default +component and needs none of this. ## Sharing a runtime across rebuilds @@ -62,22 +112,16 @@ container = make_async_container(ClientwrightProvider("httpx", config, deps)) An upstream that was failing before the rebuild is still remembered as failing after it — which is the entire point of a breaker. -## Several upstreams - -One provider serves one upstream. For several, instantiate one provider per -upstream and give each a typed alias so injection sites stay readable: - -```python -from dishka import Provider, Scope, provide +## The circuit-state gauge - -class WarehouseClient(Provider): - scope = Scope.APP - - @provide - def client(self, handle: ClientHandle[Any]) -> httpx.AsyncClient: - return handle.client -``` +An adapter wires the [`http_client_circuit_state`](observability.md#the-metric-families) +gauge to the runtime it builds. The provider builds the runtime, so it does the +same wiring: with `AdapterDeps(metrics=...)` and `observability.metrics` on, a +breaker transition on a provider-built runtime reaches the gauge exactly as it +does with `build()`. The one runtime that carries no listener is one you built +yourself and passed through `AdapterDeps(runtime=...)` — it is handed back as it +came, so give `ClientRuntime.for_config(config, circuit_listener=...)` a +listener that records into your metrics sink if you want the gauge there too. ## Without dishka diff --git a/docs/reference/contrib.md b/docs/reference/contrib.md index 08c9fb8..fecded8 100644 --- a/docs/reference/contrib.md +++ b/docs/reference/contrib.md @@ -22,14 +22,22 @@ here rather than auto-rendered. from clientwright.contrib.dishka import ClientwrightProvider ``` -**`ClientwrightProvider(adapter, config, deps=None)`** — a `dishka.Provider` -with `scope=Scope.APP` providing: +**`ClientwrightProvider(adapter, config, deps=None, *, component=None, client_type=None)`** +— a `dishka.Provider` with `scope=Scope.APP` providing: -- `ClientRuntime` — the injected `deps.runtime` if given, else - `ClientRuntime.for_config(config)`; one per container, shared. +- `ClientRuntime` — the injected `deps.runtime` if given, handed back as it + came; else `ClientRuntime.for_config(config)` with the + `http_client_circuit_state` listener wired the way an adapter wires it. One + per container (per component), shared. - `ClientHandle[Any]` — an async generator provide that builds the native client with the shared runtime and closes it (`aclose()` / `close()`) in `finally` when the container shuts down. +- the native client under `client_type`, when given — `handle.client`, the + same object, so `httpx.AsyncClient` resolves to what the handle holds. -Usage, scope rules and multi-upstream patterns: +`component` puts all three in a Dishka component: one provider per upstream in +one container, resolved with `container.get(..., component="github-api")` or +injected as `Annotated[..., FromComponent("github-api")]`. + +Usage, scope rules and several upstreams in one container: [Guide → Dependency injection](../guide/dishka.md). diff --git a/tests/integration/contrib/test_dishka.py b/tests/integration/contrib/test_dishka.py index 19cbd35..26177cd 100644 --- a/tests/integration/contrib/test_dishka.py +++ b/tests/integration/contrib/test_dishka.py @@ -65,3 +65,45 @@ async def test__two_container_generations__breaker_state_lives_in_the_runtime(or await rebuilt.client.get("/status/500") # rejected locally: state survived the rebuild assert origin.request_count("/status/500") == 1 await second.close() + + +async def test__two_components__each_upstream_has_its_own_breaker_and_the_trip_reaches_metrics( + origin: OriginServer, +) -> None: + # The reporter's shape: two upstreams in one container, each on its own + # runtime, each closed with the container - and the tripped breaker is + # exported although the provider, not the adapter, built the runtime. + metrics = RecordingMetrics() + deps = AdapterDeps(metrics=metrics) + + def config(service: str) -> ClientConfig: + return ClientConfig( + service_name=service, + base_url=origin.url, + retry=None, + circuit_breaker=CircuitBreakerConfig(fail_threshold=1, recovery_timeout=60.0), + ) + + container = make_async_container( + ClientwrightProvider( + "httpx", config("github-oauth"), deps, component="github-oauth", client_type=httpx.AsyncClient + ), + ClientwrightProvider( + "httpx", config("github-api"), deps, component="github-api", client_type=httpx.AsyncClient + ), + ) + login = await container.get(httpx.AsyncClient, component="github-oauth") + api = await container.get(httpx.AsyncClient, component="github-api") + assert login is not api + + trip = await api.get("/status/500") + assert trip.status_code == 500 # one 5xx signal arms github-api's breaker + with pytest.raises(httpx.HTTPError): + await api.get("/status/500") # rejected locally + assert (await login.get("/status/500")).status_code == 500 # github-oauth's breaker is its own + assert origin.request_count("/status/500") == 2 + assert [record["state"] for record in metrics.circuit_states] == ["open", "open"] + + await container.close() + assert login.is_closed + assert api.is_closed diff --git a/tests/unit/contrib/test_dishka.py b/tests/unit/contrib/test_dishka.py index 62776c9..f591383 100644 --- a/tests/unit/contrib/test_dishka.py +++ b/tests/unit/contrib/test_dishka.py @@ -2,18 +2,20 @@ from __future__ import annotations -from typing import Any +from typing import Annotated, Any import pytest pytest.importorskip("dishka", reason="requires the [dishka] extra") httpx = pytest.importorskip("httpx", reason="requires the [httpx] extra") -from dishka import make_async_container # noqa: E402 +from dishka import FromComponent, Provider, Scope, make_async_container, provide # noqa: E402 import clientwright # noqa: E402 +from clientwright import AdapterDeps, CircuitBreakerConfig, FailureKind, Outcome # noqa: E402 from clientwright.contrib.dishka import ClientwrightProvider # noqa: E402 from clientwright.core.plan import ClientHandle, ClientRuntime # noqa: E402 +from clientwright.core.telemetry import names # noqa: E402 async def test__provider__builds_client_and_closes_it_with_container() -> None: @@ -26,3 +28,97 @@ async def test__provider__builds_client_and_closes_it_with_container() -> None: assert not handle.client.is_closed await container.close() assert handle.client.is_closed # the generator provide's finally ran + + +async def test__request_scope__leaves_the_client_open_until_the_container_closes() -> None: + # Close travels with the APP-scope provide: a request scope coming and going + # must not touch the client (the legacy kit parked aclose on a + # request-scoped exit stack and leaked a client per request). + config = clientwright.ClientConfig(service_name="di-scope") + container = make_async_container(ClientwrightProvider("httpx", config)) + async with container() as request: + handle = await request.get(ClientHandle[Any]) + assert not handle.client.is_closed + await container.close() + assert handle.client.is_closed + + +async def test__two_components__serve_two_upstreams_from_one_container() -> None: + login = clientwright.ClientConfig(service_name="github-oauth", base_url="https://github.com") + api = clientwright.ClientConfig(service_name="github-api", base_url="https://api.github.com") + container = make_async_container( + ClientwrightProvider("httpx", login, component="github-oauth"), + ClientwrightProvider("httpx", api, component="github-api"), + ) + login_handle = await container.get(ClientHandle[Any], component="github-oauth") + api_handle = await container.get(ClientHandle[Any], component="github-api") + assert login_handle is not api_handle + assert login_handle.plan.config.service_name == "github-oauth" + assert api_handle.plan.config.service_name == "github-api" + # Each component resolves its own runtime: two upstreams are two breakers. + assert login_handle.runtime is not api_handle.runtime + assert login_handle.runtime is await container.get(ClientRuntime, component="github-oauth") + await container.close() + assert login_handle.client.is_closed + assert api_handle.client.is_closed + + +async def test__client_type__provides_the_native_client_the_handle_holds() -> None: + config = clientwright.ClientConfig(service_name="di-typed") + container = make_async_container(ClientwrightProvider("httpx", config, client_type=httpx.AsyncClient)) + client = await container.get(httpx.AsyncClient) + handle = await container.get(ClientHandle[Any]) + assert client is handle.client + await container.close() + assert client.is_closed + + +class GitHubOAuthProvider: + """The reporter's shape: two clientwright clients handed to one object.""" + + def __init__( + self, + login_client: Annotated[httpx.AsyncClient, FromComponent("github-oauth")], + api_client: Annotated[httpx.AsyncClient, FromComponent("github-api")], + ) -> None: + self.login_client = login_client + self.api_client = api_client + + +class ServiceProvider(Provider): + scope = Scope.APP + github = provide(GitHubOAuthProvider) + + +async def test__client_type_per_component__injects_each_upstream_by_name() -> None: + login = clientwright.ClientConfig(service_name="github-oauth", base_url="https://github.com") + api = clientwright.ClientConfig(service_name="github-api", base_url="https://api.github.com") + container = make_async_container( + ClientwrightProvider("httpx", login, component="github-oauth", client_type=httpx.AsyncClient), + ClientwrightProvider("httpx", api, component="github-api", client_type=httpx.AsyncClient), + ServiceProvider(), + ) + github = await container.get(GitHubOAuthProvider) + login_handle = await container.get(ClientHandle[Any], component="github-oauth") + api_handle = await container.get(ClientHandle[Any], component="github-api") + assert github.login_client is login_handle.client + assert github.api_client is api_handle.client + await container.close() + assert github.login_client.is_closed + assert github.api_client.is_closed + + +async def test__runtime_built_by_provider__moves_the_circuit_state_gauge() -> None: + prometheus_client = pytest.importorskip("prometheus_client", reason="requires the [metrics] extra") + from clientwright.adapters.observability import PrometheusClientMetrics # noqa: PLC0415 - behind the extra + + registry = prometheus_client.CollectorRegistry() + config = clientwright.ClientConfig(service_name="di-gauge", circuit_breaker=CircuitBreakerConfig(fail_threshold=1)) + deps = AdapterDeps(metrics=PrometheusClientMetrics(registry=registry)) + container = make_async_container(ClientwrightProvider("httpx", config, deps)) + handle = await container.get(ClientHandle[Any]) + assert handle.runtime.circuits is not None + handle.runtime.circuits.record("https://wh.example.com:443", Outcome(kind=FailureKind.STATUS, status_code=503)) + labels = {"service": "di-gauge", "adapter": "httpx", "key": "https://wh.example.com:443"} + assert registry.get_sample_value(names.CIRCUIT_STATE, labels) == 2.0 # open + await container.close()