diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 8502ba7..b977029 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -2,7 +2,7 @@ version: 2 updates: # No Cargo block, and no pip one either. This distribution has exactly one - # runtime requirement — `dynamic-config-py` — and seven framework extras, + # runtime requirement — `dynamic-config-py` — and nine framework extras, # none of which is locked: a library that pinned its dependencies would pin # its users'. `scripts/resolve-web-audit.py` is what resolves them, and the # OSV job in security.yml is what watches them. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d959b06..8cce551 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -281,6 +281,35 @@ jobs: python "$example" > /dev/null done + coverage: + needs: [changes] + if: needs.changes.outputs.python == 'true' || github.event_name != 'pull_request' + name: coverage report + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.12" + cache: pip + cache-dependency-path: pyproject.toml + # Every framework, so the number covers the adapters too — the same + # environment the `types` job builds. + - run: pip install -e ".[dev,all,robyn,django-bolt]" pytest-cov + - run: | + python -m pytest tests -q \ + --cov=dynamic_config_web --cov-report=term \ + --cov-report=json:coverage.json + - name: the number, where reviews can see it + run: | + python - <<'REPORT' >> "$GITHUB_STEP_SUMMARY" + import json + + covered = json.load(open("coverage.json"))["totals"]["percent_covered"] + print(f"line coverage: {covered:.1f}%") + REPORT + book: needs: [changes] if: needs.changes.outputs.docs == 'true' || github.event_name != 'pull_request' @@ -343,7 +372,7 @@ jobs: runs-on: ubuntu-latest timeout-minutes: 5 if: always() - needs: [changes, lint, core, adapters, django-app, types, examples, book, docs-links, actionlint] + needs: [changes, lint, core, adapters, django-app, types, examples, coverage, book, docs-links, actionlint] steps: - name: every job succeeded run: | diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index fd86298..6141644 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -7,7 +7,7 @@ name: Security # # That matters more here than anywhere else in the organisation: this is the # one distribution whose dependency graph is somebody else's web framework — -# seven of them — and a web framework is where advisories actually land. +# nine of them — and a web framework is where advisories actually land. # # No push trigger for `dev` — same reasoning as ci.yml: dev travels through # pull requests, and a push twin under required checks poisons the gate. @@ -89,6 +89,36 @@ jobs: fail-on-severity: low comment-summary-in-pr: on-failure + published: + name: the published wheel is still pure + # Schedule and dispatch only: this asks PyPI, not the tree, so a pull + # request cannot change its answer — but a compromised upload can, and + # a weekly read is how that is noticed here rather than in an issue. + if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch' + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.12" + - name: what PyPI serves imports no framework + run: | + pip install --quiet dynamic-config-py-web + python - <<'CHECK' + import sys + + import dynamic_config_web + + frameworks = { + "django", "django_bolt", "fastapi", "flask", + "litestar", "quart", "rest_framework", "robyn", "starlette", + } + arrived = sorted(n for n in sys.modules if n.split(".")[0] in frameworks) + + assert not arrived, f"the published wheel pulled in {arrived}" + print(f"pure, at {dynamic_config_web.__version__}") + CHECK + # The one status branch protection requires from this workflow — same # reasoning as CI's gate. `supply-chain` IS in the needs even though it only # exists on pull requests: the gate's own check tolerates `skipped`, so a @@ -96,7 +126,7 @@ jobs: # — which is the whole point of running it. security-ok: name: Security is green - needs: [osv, supply-chain] + needs: [osv, supply-chain, published] if: always() runs-on: ubuntu-latest timeout-minutes: 5 diff --git a/CHANGELOG.md b/CHANGELOG.md index afb3946..99a4bf4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,45 @@ for an adapter and an adapter fix should not drag the wheels behind it. ## [Unreleased] +### Changed + +- **The route table is written once.** Six adapters — FastAPI, Litestar, + Flask, Quart, Robyn and django-bolt — had each re-declared `/healthz`, + `/readyz`, `/metrics` and the two guarded diagnostics routes with the + same bodies; they now loop over one shared table and translate only + what is genuinely theirs: the path syntax, the response type, and the + refusal convention (the WSGI-shaped adapters keep 404, DRF keeps 403, + Django Ninja keeps 401). The raw-ASGI scope middleware FastAPI and + Litestar carried twice is one module now. No public signature moved; + the conformance suite is the proof, thirteen cases against every + adapter, unchanged before and after. + + The Django family stays off the table on purpose: its views late-bind + the installation per request and are individually routable public API. + They share the same `_health`/`_metrics`/`_diagnostics` bodies one + level down. + +### Fixed + +- **A scope over several configurations can no longer tear across a + reload.** Each configuration has its own atomic cell and the engine + keeps no epoch across them, so opening a scope was N independent reads — + and a reload landing between two of them put two generations in one + request. `enter()` now reads every install counter before and after, + and starts over when anything moved, with the same retry budget as the + Rust web core's `Sections::take`. One configuration pays nothing. + + The conformance suite gained the case that would have caught it: a + wiring over two configurations, read, moved underneath the request, and + read again — thirteen cases now, asked of all nine adapters. + +### Added + +- **A standalone DRF example.** `examples/09_django_drf.py` — the health + surface as APIViews and the diagnostics behind + `ConfigDiagnosticsPermission`, beside the django-ninja example instead + of folded into the Django one. + ## 0.1.0 — 2026-08-18 ### Added diff --git a/book/src/SUMMARY.md b/book/src/SUMMARY.md index b6ceb05..7bf2f56 100644 --- a/book/src/SUMMARY.md +++ b/book/src/SUMMARY.md @@ -4,6 +4,7 @@ # Guide +- [Quick Start](quick-start.md) - [The Rules](rules.md) - [Wiring & Lifetime](wiring.md) - [The Request Scope](scope.md) diff --git a/book/src/introduction.md b/book/src/introduction.md index 99796bb..a56bd51 100644 --- a/book/src/introduction.md +++ b/book/src/introduction.md @@ -1,5 +1,9 @@ # Web Integrations +> **Python.** This book covers the *Python* web adapters +> (`dynamic-config-py-web`). The Rust web crates — axum, Actix, Loco, +> tower — have [their own book](https://dynamic-config-rs.github.io/rust-web/). + `dynamic-config-py` resolves configuration and hands a program a validated model that a file edit can replace while the process serves. Everything a *web* application needs around that — where the watcher starts, how a diff --git a/book/src/quick-start.md b/book/src/quick-start.md new file mode 100644 index 0000000..aa00bf7 --- /dev/null +++ b/book/src/quick-start.md @@ -0,0 +1,57 @@ +# Quick Start + +```sh +pip install "dynamic-config-py[fastapi]" +``` + +```python +from dataclasses import dataclass + +from fastapi import Depends, FastAPI + +from dynamic_config import DynamicConfig +from dynamic_config_web.fastapi import config_dependency, setup + + +@dataclass +class Database: + host: str = "localhost" + port: int = 5432 + + +config = DynamicConfig(Database, key="db").file("config.toml").env("APP_") + +app = FastAPI() +setup(app, config) # lifecycle + request scope + routes +database = config_dependency(config) + + +@app.get("/") +def index(db: Database = Depends(database)) -> dict[str, str]: + return {"host": db.host} +``` + +Run it — `uvicorn main:app` — and you have: + +| | | +|---|---| +| `GET /` | your handler, reading one pinned snapshot per request | +| `GET /healthz` | 200 while the process lives | +| `GET /readyz` | 200 serving, 503 when nothing loaded or reloads are failing | +| `GET /metrics` | the engine's series, Prometheus text | + +`setup` did four things: loaded before the first request (a broken +document fails startup, not traffic), started the watcher and stops it on +shutdown, opened a request scope around every request, and mounted the +routes above. Every adapter here is those same four things +through its own framework's seams — the [Introduction](introduction.md) +has the table of nine. + +Edit `config.toml` while it serves: the *next* request answers with the +new document, and no request ever straddles the change — that is the +request scope, and [The Rules](rules.md) is the page that spells out +what it promises and what it refuses to. + +Diagnostics (`/_config/explain`, `/_config/check`) exist only when you +pass a guard: `setup(app, config, guard=token_guard("s3cret"))` — see +[Diagnostics](diagnostics.md). diff --git a/examples/09_django_drf.py b/examples/09_django_drf.py new file mode 100644 index 0000000..5f49015 --- /dev/null +++ b/examples/09_django_drf.py @@ -0,0 +1,144 @@ +"""Django REST Framework: the health surface inside DRF's own auth. + + pip install django djangorestframework + python examples/09_django_drf.py + +The Django adapter supplies everything that is not routing — `AppConfig.ready` +loads and watches, the middleware opens the request scope — and the plain +Django views would already work in a DRF project. What this module adds is +the *permission seam*: the diagnostics sit behind +`ConfigDiagnosticsPermission`, which defers to the installation's guard and +can be swapped for `IsAdminUser`, a scope check, or anything else DRF +offers. + +A real project writes: + + INSTALLED_APPS = [..., "dynamic_config_web.django", "rest_framework"] + MIDDLEWARE = ["dynamic_config_web.django.middleware.DynamicConfigMiddleware", ...] + DYNAMIC_CONFIG = {"target": "myproject.config:database", "guard": "..."} + + from dynamic_config_web.django.drf import urls as config_urls + urlpatterns = [path("internal/", include(config_urls())), ...] + +One convention to know: a request DRF refuses gets **403**, where the plain +views answer 404 and django-ninja answers 401. Each adapter keeps its +framework's own convention. +""" + +from __future__ import annotations + +from typing import Any + +from _shared import Database, show, workspace +from dynamic_config import DynamicConfig +from dynamic_config_web import token_guard + +try: + import django + from django.conf import settings +except ImportError: # pragma: no cover - the example says how to fix it + raise SystemExit("this example needs Django: pip install django") from None + +try: + import rest_framework # noqa: F401 +except ImportError: # pragma: no cover + raise SystemExit( + "this example needs DRF: pip install djangorestframework" + ) from None + +#: Rewritten in `main()`, once the routes exist. +urlpatterns: list[Any] = [] + +database: DynamicConfig[Database] +guard = token_guard("s3cret") + + +def configure() -> None: + """What a `settings.py` would say, said in memory.""" + if settings.configured: + return + + settings.configure( + DEBUG=False, + SECRET_KEY="example", + ALLOWED_HOSTS=["*"], + ROOT_URLCONF=__name__, + INSTALLED_APPS=["dynamic_config_web.django", "rest_framework"], + MIDDLEWARE=["dynamic_config_web.django.middleware.DynamicConfigMiddleware"], + DATABASES={}, + USE_TZ=True, + REST_FRAMEWORK={"UNAUTHENTICATED_USER": None}, + DYNAMIC_CONFIG={ + "target": f"{__name__}:database", + "debounce": 0.05, + "guard": f"{__name__}:guard", + }, + ) + + +def main() -> None: + """Runs the DRF example end to end.""" + global database + + with workspace() as path: + database = DynamicConfig(Database, key="db").file(str(path)).env("APP_") + + configure() + django.setup() + + from django.test import Client + from rest_framework.decorators import api_view + from rest_framework.response import Response + + from dynamic_config_web.django import snapshot + from dynamic_config_web.django.drf import urls as config_urls + + @api_view(["GET"]) + def index(request: Any) -> Response: + """`snapshot()` needs no argument — the middleware scoped it.""" + del request + + db: Database = snapshot() + + return Response( + {"host": db.host, "port": db.port, "pool": db.pool.max_size} + ) + + from django.urls import path as route + + urlpatterns[:] = [route("", index), *config_urls()] + + client = Client() + + show("serving") + print(f" GET / → {client.get('/').json()}") + + show("the health surface, as APIViews") + print(f" GET /healthz → {client.get('/healthz').status_code}") + + ready = client.get("/readyz") + print(f" GET /readyz → {ready.status_code} {ready.json()['status']}") + print(f" GET /metrics → {client.get('/metrics').status_code}") + + show("a deployment edits the file") + path.write_text('[db]\nhost = "db.replica"\nport = 6543\n') + database.reload() + + print(f" GET / → {client.get('/').json()}") + + show("diagnostics, behind DRF's permission") + # 403 rather than ninja's 401 or the plain views' 404: DRF's + # convention, kept on purpose. + print(f" no token → {client.get('/_config/check').status_code}") + + answer = client.get( + "/_config/explain/port", headers={"x-config-token": "s3cret"} + ) + print(f" with one → {answer.status_code}") + + for line in answer.content.decode().splitlines()[:4]: + print(f" {line}") + + +if __name__ == "__main__": + main() diff --git a/src/dynamic_config_web/__init__.py b/src/dynamic_config_web/__init__.py index 3d4a190..4539e8d 100644 --- a/src/dynamic_config_web/__init__.py +++ b/src/dynamic_config_web/__init__.py @@ -1,6 +1,6 @@ """Native web-framework integrations for `dynamic-config-py`. -Seven frameworks, one shape. Whatever the framework calls its startup +Nine adapters, one shape. Whatever the framework calls its startup hook, its dependency injection and its router, an integration here does the same five things: diff --git a/src/dynamic_config_web/_asgi.py b/src/dynamic_config_web/_asgi.py new file mode 100644 index 0000000..8fe123d --- /dev/null +++ b/src/dynamic_config_web/_asgi.py @@ -0,0 +1,52 @@ +"""The raw-ASGI request scope, shared by every ASGI adapter. + +FastAPI and Litestar carried the same fifteen lines each; this is that +middleware once. Raw ASGI rather than either framework's middleware +class, because a scope opened around `receive`/`send` is opened around +*everything* — routing included — and depends on nothing the frameworks +disagree about. + +Only `http` gets a scope. A `websocket` scope IS the connection, and a +connection that lives an hour pinned to the configuration it opened with +would be the opposite of what this package is for — `latest()` is the +documented read there (the book's Limitations page owns that decision). +`lifespan` and anything else pass through untouched. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from . import _scope + +if TYPE_CHECKING: # pragma: no cover - typing only + from ._wiring import Wiring + +__all__ = ["ScopeMiddleware"] + + +class ScopeMiddleware: + """One reading per request, entered before routing, left after send.""" + + def __init__(self, app: Any, wiring: Wiring) -> None: + self._app = app + self._wiring = wiring + + # `Any` throughout the ASGI triple, deliberately: one framework types + # `scope` as a TypedDict union, another as a MutableMapping, and this + # middleware must satisfy both protocols structurally. + async def __call__(self, scope: Any, receive: Any, send: Any) -> None: + # `str()` on purpose: one framework types `scope["type"]` as a + # Literal union, and comparing through `str` satisfies both it + # and the plain-dict world without a cast. + if str(scope.get("type")) != "http": + await self._app(scope, receive, send) + + return + + token = _scope.enter(self._wiring.configs) + + try: + await self._app(scope, receive, send) + finally: + _scope.leave(token) diff --git a/src/dynamic_config_web/_health.py b/src/dynamic_config_web/_health.py index 499a67b..9829fe4 100644 --- a/src/dynamic_config_web/_health.py +++ b/src/dynamic_config_web/_health.py @@ -4,7 +4,7 @@ conditions, and a service that conflates them either refuses traffic it could serve or accepts traffic on a configuration nobody has been able to reload for an hour. The Python book states the pair; this builds the two -answers so seven adapters do not each write them. +answers so nine adapters do not each write them. **No value ever reaches the body.** Generations, counts, kinds, paths and seconds — the same rule the engine's own diagnostics follow, and for the diff --git a/src/dynamic_config_web/_routes.py b/src/dynamic_config_web/_routes.py new file mode 100644 index 0000000..9f73fd3 --- /dev/null +++ b/src/dynamic_config_web/_routes.py @@ -0,0 +1,225 @@ +"""The route table: what the health surface serves, written once. + +Every adapter mounts the same five routes — `/healthz`, `/readyz`, +`/metrics`, and behind a guard `/_config/explain/{path}` and +`/_config/check` — and until 0.2 each adapter re-declared all five with +the same bodies. Six adapters loop over this table now — FastAPI, +Litestar, Flask, Quart, Robyn and django-bolt — translating each +:class:`Reply` into their framework's response type, which is the only +part that differs. + +The Django family stays off the table, on purpose: its views late-bind +the *installation* per request (Django settings own the wiring there, +and `AppConfig.ready` may re-run in tests), each view is individually +routable public API, and DRF and Ninja wrap them in their frameworks' +own permission machinery. They consume the same `_health`, `_metrics` +and `_diagnostics` functions this table does — the bodies are shared +one level down. + +The table knows no framework. A route handler takes a +:class:`RouteContext` (the two request-shaped things any route here +needs: the path tail and the query string) and answers a :class:`Reply` +(status, rendered body, content type). Refusals are :class:`RouteError`, +which each adapter maps to its framework's own error idiom — that is +where DRF's 403, Ninja's 401 and the plain views' 404 stay theirs. +""" + +from __future__ import annotations + +import json +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Optional + +from ._diagnostics import Guard, check, check_async, explain, explain_async, never +from ._health import liveness, readiness +from ._metrics import CONTENT_TYPE, metrics_body + +if TYPE_CHECKING: # pragma: no cover - typing only + from ._wiring import Wiring + +__all__ = [ + "Reply", + "Route", + "RouteContext", + "RouteError", + "allowed", + "named", + "route_table", +] + +JSON = "application/json" +TEXT = "text/plain; charset=utf-8" + + +class RouteError(Exception): + """A refusal, framework-neutrally: the status and a safe detail. + + Adapters translate this into their framework's error response. The + detail never carries a configuration value — the same rule every + diagnostic in this package follows. + """ + + def __init__(self, status: int, detail: str) -> None: + super().__init__(detail) + self.status = status + self.detail = detail + + +@dataclass(frozen=True) +class Reply: + """A rendered response: what is left after the framework is gone.""" + + status: int + body: str + content_type: str + + +@dataclass(frozen=True) +class RouteContext: + """The two request-shaped inputs any route in the table needs.""" + + path_param: Optional[str] = None + query: Mapping[str, str] = None # type: ignore[assignment] + + def query_get(self, key: str) -> Optional[str]: + return None if self.query is None else self.query.get(key) + + +@dataclass(frozen=True) +class Route: + """One mounted path: where, whether it is guarded, and what it does.""" + + path: str + name: str + guarded: bool + handle: Callable[[RouteContext], Reply] + handle_async: Callable[[RouteContext], Awaitable[Reply]] + + +def allowed(request: Any, guard: Optional[Guard], *, refused: int = 403) -> None: + """Refuses a request the guard does not accept. + + The guard receives the *framework's* request object — that is the + guard protocol, and `_header` inside `token_guard` already speaks + every framework's header idiom. What is shared is only the refusal; + its *status* stays each adapter's documented convention (`refused`): + the WSGI-shaped adapters answer 404 so a guarded route is + indistinguishable from an absent one, DRF answers its own 403, and + Django Ninja's `auth=` answers 401 before any of this runs. + """ + if guard is not None and not guard(request): + detail = "not found" if refused == 404 else "not permitted" + + raise RouteError(refused, detail) + + +def named(wiring: Wiring, key: Optional[str]) -> Any: + """The configuration a diagnostics request names, or the only one.""" + configs = wiring.configs + + if key is None: + if len(configs) == 1: + return configs[0] + + raise RouteError( + 400, + "this application has more than one configuration; name one " + f"with ?config=, from: {', '.join(c.key for c in configs)}", + ) + + for config in configs: + if config.key == key: + return config + + raise RouteError(404, f"no configuration named {key!r}") + + +def route_table( + wiring: Wiring, + *, + metrics: bool = True, + stale_after: Optional[float] = None, + guard: Optional[Guard] = None, + diagnostics_prefix: str = "/_config", +) -> tuple[Route, ...]: + """The five routes over `wiring`, as data. + + The two diagnostics routes exist only when a real guard was given: + `None` and the shipped `never` both mean *do not build these* — the + first because nobody asked, the second because somebody wrote the + refusal out. + """ + + def healthz(_: RouteContext) -> Reply: + report = liveness() + + return Reply(report.status_code, json.dumps(report.body), JSON) + + def readyz(_: RouteContext) -> Reply: + report = readiness(*wiring.configs, stale_after=stale_after) + + return Reply(report.status_code, json.dumps(report.body), JSON) + + def prometheus(_: RouteContext) -> Reply: + return Reply(200, metrics_body(*wiring.configs), CONTENT_TYPE) + + def explain_path(context: RouteContext) -> Reply: + config = named(wiring, context.query_get("config")) + + return Reply(200, explain(config, context.path_param or ""), TEXT) + + async def explain_path_async(context: RouteContext) -> Reply: + config = named(wiring, context.query_get("config")) + + return Reply(200, await explain_async(config, context.path_param or ""), TEXT) + + def check_all(_: RouteContext) -> Reply: + report = {config.key: check(config) for config in wiring.configs} + + return Reply(200, json.dumps(report), JSON) + + async def check_all_async(_: RouteContext) -> Reply: + report = {config.key: await check_async(config) for config in wiring.configs} + + return Reply(200, json.dumps(report), JSON) + + def as_async( + handler: Callable[[RouteContext], Reply], + ) -> Callable[[RouteContext], Awaitable[Reply]]: + async def twin(context: RouteContext) -> Reply: + return handler(context) + + return twin + + routes = [ + Route("/healthz", "healthz", False, healthz, as_async(healthz)), + Route("/readyz", "readyz", False, readyz, as_async(readyz)), + ] + + if metrics: + routes.append( + Route("/metrics", "metrics", False, prometheus, as_async(prometheus)) + ) + + if guard is not None and guard is not never: + routes.append( + Route( + f"{diagnostics_prefix}/explain/{{path}}", + "explain", + True, + explain_path, + explain_path_async, + ) + ) + routes.append( + Route( + f"{diagnostics_prefix}/check", + "check", + True, + check_all, + check_all_async, + ) + ) + + return tuple(routes) diff --git a/src/dynamic_config_web/_scope.py b/src/dynamic_config_web/_scope.py index b6c28fa..760db18 100644 --- a/src/dynamic_config_web/_scope.py +++ b/src/dynamic_config_web/_scope.py @@ -97,6 +97,37 @@ def _members(targets: Iterable[Any]) -> list[Any]: return configs +#: How many times :func:`enter` re-reads when a reload lands mid-read. +#: The same constant, for the same reason, as the Rust web core's +#: `ATTEMPTS`: past this, the last read is served — no worse than not +#: checking, which is what every caller had before the check existed. +_ATTEMPTS = 8 + + +def _read_once(configs: list[Any]) -> _Snapshot: + """One `try_current()` per configuration, into a snapshot.""" + by_object: dict[Any, Any] = {} + by_key: dict[str, Any] = {} + + for config in configs: + model = config.try_current() + by_object[config] = model + by_key[config.key] = model + + return _Snapshot(by_object, by_key) + + +def _generations(configs: list[Any]) -> tuple[int, ...]: + """Every configuration's install counter, in list order. + + The engine bumps the counter *after* the model is readable, so a + counter can lag its model but never lead it — the comparison in + :func:`enter` therefore errs only toward a harmless extra read, + never toward accepting a torn one. + """ + return tuple(config.generation for config in configs) + + def enter(targets: Iterable[DynamicConfig[Any] | ConfigGroup]) -> Token[Any]: """Opens a scope holding one read of each configuration in `targets`. @@ -106,17 +137,33 @@ def enter(targets: Iterable[DynamicConfig[Any] | ConfigGroup]) -> Token[Any]: 503 the health surface should answer, not an exception from the middleware that opened the scope. + With more than one configuration the reads have to *agree*: each has + its own atomic cell and the engine keeps no epoch across them, so N + reads are N independent loads, and a reload landing between two of + them would put two generations in one scope — exactly the tear this + module exists to prevent, one level up. So the counters are read + before and after, and the read starts over if anything moved. The + Rust web core's `Sections::take` makes the same check with the same + retry budget. + Answers the token :func:`leave` restores. """ - by_object: dict[Any, Any] = {} - by_key: dict[str, Any] = {} + configs = _members(targets) - for config in _members(targets): - model = config.try_current() - by_object[config] = model - by_key[config.key] = model + # A single configuration cannot straddle anything. + if len(configs) < 2: + return _SNAPSHOT.set(_read_once(configs)) + + for _ in range(_ATTEMPTS): + before = _generations(configs) + snapshot = _read_once(configs) + + if _generations(configs) == before: + return _SNAPSHOT.set(snapshot) - return _SNAPSHOT.set(_Snapshot(by_object, by_key)) + # Reloading faster than a read completes, eight times running. The + # last read is served: no worse than not checking. + return _SNAPSHOT.set(_read_once(configs)) def leave(token: Token[Any]) -> None: diff --git a/src/dynamic_config_web/django_bolt.py b/src/dynamic_config_web/django_bolt.py index 8162712..8fa7961 100644 --- a/src/dynamic_config_web/django_bolt.py +++ b/src/dynamic_config_web/django_bolt.py @@ -34,14 +34,14 @@ async def index(): from __future__ import annotations +import json from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import TYPE_CHECKING, Any, Callable, Optional -from ._diagnostics import Guard, check_async, explain_async, never +from ._diagnostics import Guard from ._errors import MissingFrameworkError -from ._health import liveness, readiness -from ._metrics import CONTENT_TYPE, metrics_body +from ._routes import RouteContext, RouteError, allowed, route_table from ._scope import current, enter, get, leave from ._wiring import Wiring @@ -153,58 +153,72 @@ class — which a dict body then fails. `Any` is the annotation that """ routes: Any = Router(prefix=prefix) + table = { + entry.name: entry + for entry in route_table( + running, + metrics=metrics, + stale_after=stale_after, + guard=guard, + diagnostics_prefix=diagnostics_prefix, + ) + } + + # Declared one by one rather than in a loop: django-bolt validates a + # handler against its *signature*, and the path parameter's presence + # changes it. The bodies are the shared table's; only the signatures + # are this framework's. + + async def answer(entry: Any, request: Any, path: Optional[str]) -> Any: + if entry.guarded: + try: + allowed(request, guard, refused=404) + except RouteError as stopped: + return Response({"detail": stopped.detail}, status_code=stopped.status) + + context = RouteContext( + path_param=None if path is None else path.lstrip("/"), + query={} if request is None else request.query, + ) + + try: + reply = await entry.handle_async(context) + except RouteError as stopped: + return Response({"detail": stopped.detail}, status_code=stopped.status) + + # Bolt's `Response` serialises its body itself, so a JSON reply + # goes back to a dict here — handing it the rendered string would + # double-encode it. + body: Any = reply.body + + if reply.content_type == "application/json": + body = json.loads(reply.body) + + return Response(body, status_code=reply.status, media_type=reply.content_type) + @routes.get("/healthz") async def healthz() -> Any: - """The process is up. Configuration has no say in this one.""" - report = liveness() - - return Response(report.body, status_code=report.status_code) + return await answer(table["healthz"], None, None) @routes.get("/readyz") async def readyz() -> Any: - """Serving something, and the reloads since have worked.""" - report = readiness(*running.configs, stale_after=stale_after) + return await answer(table["readyz"], None, None) - return Response(report.body, status_code=report.status_code) - - if metrics: + if "metrics" in table: @routes.get("/metrics") async def prometheus() -> Any: - """The engine's series, built per scrape.""" - return Response( - metrics_body(*running.configs), - media_type=CONTENT_TYPE, - ) + return await answer(table["metrics"], None, None) - if guard is not None and guard is not never: + if "explain" in table: @routes.get(f"{diagnostics_prefix}/explain/{{path:path}}") async def explain_path(request: Any, path: str) -> Any: - """Every layer's answer for one dotted path, off the loop.""" - if not guard(request): - return Response({"detail": "not found"}, status_code=404) - - try: - # `request.query`, which is django-bolt's name for it. - config = running.config(request.query.get("config")) - except LookupError as unknown: - return Response({"detail": str(unknown)}, status_code=400) - - return Response( - await explain_async(config, path.lstrip("/")), - media_type="text/plain", - ) + return await answer(table["explain"], request, path) @routes.get(f"{diagnostics_prefix}/check") async def check_all(request: Any) -> Any: - """Would each configuration load, and any unknown keys.""" - if not guard(request): - return Response({"detail": "not found"}, status_code=404) - - return Response( - {config.key: await check_async(config) for config in running.configs} - ) + return await answer(table["check"], request, None) return routes diff --git a/src/dynamic_config_web/fastapi.py b/src/dynamic_config_web/fastapi.py index 38c465d..ed836ba 100644 --- a/src/dynamic_config_web/fastapi.py +++ b/src/dynamic_config_web/fastapi.py @@ -39,16 +39,15 @@ def index(db: Database = Depends(database)): from contextlib import asynccontextmanager from typing import TYPE_CHECKING, Any, Callable, Optional, TypeVar -from ._diagnostics import Guard, check_async, explain_async, never +from ._asgi import ScopeMiddleware as _SharedScopeMiddleware +from ._diagnostics import Guard from ._errors import MissingFrameworkError -from ._health import liveness, readiness -from ._metrics import CONTENT_TYPE, metrics_body -from ._scope import current, enter, leave +from ._routes import RouteContext, RouteError, allowed, route_table +from ._scope import current from ._wiring import Wiring try: - from fastapi import APIRouter, HTTPException, Request - from fastapi.responses import JSONResponse, PlainTextResponse + from fastapi import APIRouter, HTTPException, Request, Response except ImportError as absent: # pragma: no cover - exercised in a subprocess raise MissingFrameworkError("FastAPI", "fastapi") from absent @@ -134,34 +133,9 @@ async def lifespan(app): wiring.stop() -class _ScopeMiddleware: - """Opens one request scope per request, in raw ASGI. - - Raw rather than `BaseHTTPMiddleware`: the base class runs the - downstream app in a task of its own, which is exactly the boundary a - `ContextVar` set here would not cross — and it costs a queue per - request for a job that is two dictionary writes. - """ - - def __init__(self, app: Any, wiring: Wiring) -> None: - self.app = app - self.wiring = wiring - - async def __call__(self, scope: Any, receive: Any, send: Any) -> None: - if scope["type"] != "http": - # A websocket lives longer than a "request" and a lifespan - # message is not one at all; neither should be pinned to a - # snapshot taken at connect time. - await self.app(scope, receive, send) - - return - - token = enter(self.wiring.configs) - - try: - await self.app(scope, receive, send) - finally: - leave(token) +# The raw-ASGI scope middleware lives in `_asgi` now — FastAPI and +# Litestar mount the same fifteen lines, so the lines exist once. +_ScopeMiddleware = _SharedScopeMiddleware def router( @@ -178,57 +152,55 @@ def router( `setup` includes this; taking it directly is for an application that wants them under its own prefix, behind its own dependencies, or in a sub-application it mounts elsewhere. + + The routes themselves are the shared table — nine adapters, one + definition — and this function is only the translation into FastAPI: + a path parameter for `{path}`, `Response` from a `Reply`, and + `HTTPException` from a refusal. """ routes = APIRouter(prefix=prefix) - @routes.get("/healthz", include_in_schema=False) - async def healthz() -> JSONResponse: - """The process is up. Configuration has no say in this one.""" - report = liveness() - - return JSONResponse(report.body, status_code=report.status_code) - - @routes.get("/readyz", include_in_schema=False) - async def readyz() -> JSONResponse: - """Serving something, and the reloads since have worked.""" - report = readiness(*wiring.configs, stale_after=stale_after) - - return JSONResponse(report.body, status_code=report.status_code) - - if metrics: - - @routes.get("/metrics", include_in_schema=False) - async def prometheus() -> PlainTextResponse: - """The engine's twelve series, built per scrape.""" - return PlainTextResponse( - metrics_body(*wiring.configs), media_type=CONTENT_TYPE - ) + table = route_table( + wiring, + metrics=metrics, + stale_after=stale_after, + guard=guard, + diagnostics_prefix=diagnostics_prefix, + ) - # `None` and the shipped `never` both mean *do not build these* — the - # first because nobody asked, the second because somebody wrote the - # refusal out. Anything else is a decision, and the routes exist. - if guard is not None and guard is not never: - diagnostics = APIRouter(prefix=diagnostics_prefix) + for route in table: + mount = route.path.replace("{path}", "{path:path}") - @diagnostics.get("/explain/{path:path}", include_in_schema=False) - async def explain_path(request: Request, path: str) -> PlainTextResponse: - """Every layer's answer for one dotted path.""" - _allowed(request, guard) + def make(entry: Any) -> Callable[..., Any]: + async def endpoint(request: Request, path: str = "") -> Response: + if entry.guarded: + try: + allowed(request, guard) + except RouteError as refused: + raise HTTPException( + status_code=refused.status, detail=refused.detail + ) from None - config = _named(wiring, request.query_params.get("config")) + context = RouteContext( + path_param=path or None, query=request.query_params + ) - return PlainTextResponse(await explain_async(config, path)) + try: + reply = await entry.handle_async(context) + except RouteError as refused: + raise HTTPException( + status_code=refused.status, detail=refused.detail + ) from None - @diagnostics.get("/check", include_in_schema=False) - async def check_all(request: Request) -> JSONResponse: - """Would each configuration load, and any unknown keys.""" - _allowed(request, guard) + return Response( + content=reply.body, + status_code=reply.status, + media_type=reply.content_type, + ) - return JSONResponse( - {config.key: await check_async(config) for config in wiring.configs} - ) + return endpoint - routes.include_router(diagnostics) + routes.get(mount, include_in_schema=False, name=route.name)(make(route)) return routes @@ -324,32 +296,3 @@ async def combined(application: FastAPI) -> AsyncIterator[Any]: ) return wiring - - -def _allowed(request: Request, guard: Guard) -> None: - """Refuses a request the guard does not accept, as a 403.""" - if not guard(request): - raise HTTPException(status_code=403, detail="not permitted") - - -def _named(wiring: Wiring, key: Optional[str]) -> Any: - """The configuration a diagnostics request names, or the only one.""" - configs = wiring.configs - - if key is None: - if len(configs) == 1: - return configs[0] - - raise HTTPException( - status_code=400, - detail=( - "this application has more than one configuration; name one " - f"with ?config=, from: {', '.join(c.key for c in configs)}" - ), - ) - - for config in configs: - if config.key == key: - return config - - raise HTTPException(status_code=404, detail=f"no configuration named {key!r}") diff --git a/src/dynamic_config_web/flask.py b/src/dynamic_config_web/flask.py index 4366d32..d801187 100644 --- a/src/dynamic_config_web/flask.py +++ b/src/dynamic_config_web/flask.py @@ -34,10 +34,9 @@ def index(): import threading from typing import TYPE_CHECKING, Any, Optional -from ._diagnostics import Guard, check, explain, never +from ._diagnostics import Guard from ._errors import MissingFrameworkError -from ._health import liveness, readiness -from ._metrics import CONTENT_TYPE, metrics_body +from ._routes import RouteContext, RouteError, allowed, route_table from ._scope import current, enter, get, leave from ._wiring import Wiring @@ -261,69 +260,60 @@ def blueprint( guard: Optional[Guard] = None, diagnostics_prefix: str = "/_config", ) -> Blueprint: - """The health, metrics and diagnostics routes, as a blueprint.""" + """The health, metrics and diagnostics routes, as a blueprint. + + The routes are the shared table; this translates them into Flask — + `` for the tail, a `Response` from each `Reply`, and the + WSGI adapters' own convention for a refused guard: **404**, so a + guarded route is indistinguishable from an absent one. Synchronous + handlers, and that is right here: WSGI has no event loop to keep + free. + """ routes = Blueprint("dynamic_config", __name__, url_prefix=prefix or None) - @routes.get("/healthz") - def healthz() -> Response: - """The process is up. Configuration has no say in this one.""" - report = liveness() - - return _json(report.body, report.status_code) - - @routes.get("/readyz") - def readyz() -> Response: - """Serving something, and the reloads since have worked.""" - report = readiness(*wiring.configs, stale_after=stale_after) - - return _json(report.body, report.status_code) - - if metrics: - - @routes.get("/metrics") - def prometheus() -> Response: - """The engine's series, built per scrape.""" - return Response(metrics_body(*wiring.configs), mimetype=CONTENT_TYPE) + table = route_table( + wiring, + metrics=metrics, + stale_after=stale_after, + guard=guard, + diagnostics_prefix=diagnostics_prefix, + ) - if guard is not None and guard is not never: - diagnostics = Blueprint( - "dynamic_config_diagnostics", __name__, url_prefix=diagnostics_prefix - ) - - @diagnostics.get("/explain/") - def explain_path(path: str) -> Response: - """Every layer's answer for one dotted path. + def make(entry: Any) -> Any: + def handler(path: str = "") -> Response: + if entry.guarded: + try: + allowed(request, guard, refused=404) + except RouteError as stopped: + return _reply_of(stopped) - Synchronous, and that is fine here: WSGI has no event loop to - keep free, and this is the one place in the package where the - blocking form is the right one. - """ - if not guard(request): - return _json({"detail": "not found"}, 404) + context = RouteContext(path_param=path or None, query=request.args) try: - config = wiring.config(request.args.get("config")) - except LookupError as unknown: - return _json({"detail": str(unknown)}, 400) + reply = entry.handle(context) + except RouteError as stopped: + return _reply_of(stopped) - return Response(explain(config, path), mimetype="text/plain") + return Response( + reply.body, status=reply.status, mimetype=reply.content_type + ) - @diagnostics.get("/check") - def check_all() -> Response: - """Would each configuration load, and any unknown keys.""" - if not guard(request): - return _json({"detail": "not found"}, 404) + handler.__name__ = f"config_{entry.name}" - return _json( - {config.key: check(config) for config in wiring.configs}, - 200, - ) + return handler - routes.register_blueprint(diagnostics) + for entry in table: + rule = entry.path.replace("{path}", "") + routes.get(rule)(make(entry)) return routes +def _reply_of(stopped: RouteError) -> Response: + """A refusal, in Flask's shape.""" + return _json({"detail": stopped.detail}, stopped.status) + + def _json(body: Any, status: int) -> Response: """A JSON response with a status, in the one place Flask needs both.""" answer = jsonify(body) diff --git a/src/dynamic_config_web/litestar.py b/src/dynamic_config_web/litestar.py index 72e675d..87a3291 100644 --- a/src/dynamic_config_web/litestar.py +++ b/src/dynamic_config_web/litestar.py @@ -36,17 +36,17 @@ async def index(db: NamedDependency[Database]) -> dict[str, object]: from contextlib import asynccontextmanager from typing import TYPE_CHECKING, Any, Optional -from ._diagnostics import Guard, check_async, explain_async, never +from ._asgi import ScopeMiddleware as _SharedScopeMiddleware +from ._diagnostics import Guard from ._errors import MissingFrameworkError -from ._health import liveness, readiness -from ._metrics import CONTENT_TYPE, metrics_body -from ._scope import current, enter, leave +from ._routes import RouteContext, RouteError, allowed, route_table +from ._scope import current from ._wiring import Wiring try: from litestar import Response, Router, get from litestar.di import Provide - from litestar.exceptions import HTTPException, NotFoundException + from litestar.exceptions import HTTPException # `FromPath` rather than a bare `path: str`: the inferred style is # deprecated in 2.x and gone in 3, and an adapter must not be the @@ -73,7 +73,6 @@ async def index(db: NamedDependency[Database]) -> dict[str, object]: from litestar import Litestar from litestar.config.app import AppConfig from litestar.connection import Request - from litestar.types import Receive, Scope, Send from dynamic_config import ConfigGroup, DynamicConfig @@ -104,33 +103,13 @@ def dependency() -> Any: return Provide(dependency, sync_to_thread=False, use_cache=False) -class ScopeMiddleware: - """Opens one request scope per request, in raw ASGI. +class ScopeMiddleware(_SharedScopeMiddleware): + """The shared raw-ASGI scope middleware, under this adapter's name. - Litestar's `AbstractMiddleware` would serve as well, but raw ASGI is - the same thing without a base class and is what the other adapters - use — one shape to review rather than seven. + A subclass rather than an alias so the class's qualname says which + adapter mounted it in a traceback. """ - def __init__(self, app: Any, wiring: Wiring) -> None: - self.app = app - self.wiring = wiring - - async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - # `str(...)`, because Litestar types the key as its own enum and - # the value on the wire is the plain ASGI string either way. - if str(scope["type"]) != "http": - await self.app(scope, receive, send) - - return - - token = enter(self.wiring.configs) - - try: - await self.app(scope, receive, send) - finally: - leave(token) - def router( wiring: Wiring, @@ -141,66 +120,71 @@ def router( guard: Optional[Guard] = None, diagnostics_path: str = "/_config", ) -> Router: - """The health, metrics and diagnostics routes, as a Litestar router.""" + """The health, metrics and diagnostics routes, as a Litestar router. - @get("/healthz", include_in_schema=False, sync_to_thread=False) - def healthz() -> Response[Any]: - """The process is up. Configuration has no say in this one.""" - report = liveness() - - return Response(report.body, status_code=report.status_code) - - @get("/readyz", include_in_schema=False, sync_to_thread=False) - def readyz() -> Response[Any]: - """Serving something, and the reloads since have worked.""" - report = readiness(*wiring.configs, stale_after=stale_after) - - return Response(report.body, status_code=report.status_code) - - handlers: list[Any] = [healthz, readyz] - - if metrics: - - @get("/metrics", include_in_schema=False, sync_to_thread=False) - def prometheus() -> Response[str]: - """The engine's series, built per scrape.""" - return Response( - metrics_body(*wiring.configs), - media_type=CONTENT_TYPE, - ) - - handlers.append(prometheus) - - if guard is not None and guard is not never: - - @get( - f"{diagnostics_path}/explain/{{path:path}}", - include_in_schema=False, + The routes are the shared table; this function only translates — + Litestar's `{path:path}` parameter, its `Response`, and its + `HTTPException` from a refusal. + """ + table = route_table( + wiring, + metrics=metrics, + stale_after=stale_after, + guard=guard, + diagnostics_prefix=diagnostics_path, + ) + + handlers: list[Any] = [] + + def make(entry: Any) -> Any: + route_path = entry.path.replace("{path}", "{path:path}") + + if "{path:path}" in route_path: + + @get(route_path, include_in_schema=False, name=f"config_{entry.name}") + async def handler( + request: Request[Any, Any, Any], path: FromPath[str] + ) -> Response[str]: + return await _answer(entry, request, path) + + else: + + @get(route_path, include_in_schema=False, name=f"config_{entry.name}") + async def handler(request: Request[Any, Any, Any]) -> Response[str]: + return await _answer(entry, request, None) + + return handler + + async def _answer( + entry: Any, request: Request[Any, Any, Any], raw_path: Optional[str] + ) -> Response[str]: + if entry.guarded: + try: + allowed(request, guard) + except RouteError as refused: + raise HTTPException( + status_code=refused.status, detail=refused.detail + ) from None + + context = RouteContext( + # A Litestar path parameter arrives with its leading slash. + path_param=None if raw_path is None else raw_path.lstrip("/"), + query=request.query_params, ) - async def explain_path( - request: Request[Any, Any, Any], path: FromPath[str] - ) -> Response[str]: - """Every layer's answer for one dotted path.""" - _allowed(request, guard) - - config = _named(wiring, request.query_params.get("config")) - # A path parameter arrives with its leading slash. - dotted = path.lstrip("/") - - return Response( - await explain_async(config, dotted), media_type="text/plain" - ) - - @get(f"{diagnostics_path}/check", include_in_schema=False) - async def check_all(request: Request[Any, Any, Any]) -> Response[Any]: - """Would each configuration load, and any unknown keys.""" - _allowed(request, guard) - return Response( - {config.key: await check_async(config) for config in wiring.configs} - ) + try: + reply = await entry.handle_async(context) + except RouteError as refused: + raise HTTPException( + status_code=refused.status, detail=refused.detail + ) from None + + return Response( + reply.body, status_code=reply.status, media_type=reply.content_type + ) - handlers.extend([explain_path, check_all]) + for entry in table: + handlers.append(make(entry)) return Router(path=path, route_handlers=handlers) @@ -303,20 +287,6 @@ def _middleware(self, app: Any) -> ScopeMiddleware: return ScopeMiddleware(app, self.wiring) -def _allowed(request: Request[Any, Any, Any], guard: Guard) -> None: - """Refuses a request the guard does not accept.""" - if not guard(request): - raise NotFoundException() - - -def _named(wiring: Wiring, key: Optional[str]) -> Any: - """The configuration a diagnostics request names, or the only one.""" - try: - return wiring.config(key) - except LookupError as unknown: - raise HTTPException(status_code=400, detail=str(unknown)) from unknown - - def plugins( target: DynamicConfig[Any] | ConfigGroup | Wiring, **options: Any ) -> Sequence[InitPlugin]: diff --git a/src/dynamic_config_web/quart.py b/src/dynamic_config_web/quart.py index 72d0a1e..924a5fe 100644 --- a/src/dynamic_config_web/quart.py +++ b/src/dynamic_config_web/quart.py @@ -31,10 +31,9 @@ async def index(): from collections.abc import AsyncIterator from typing import TYPE_CHECKING, Any, Optional -from ._diagnostics import Guard, check_async, explain_async, never +from ._diagnostics import Guard from ._errors import MissingFrameworkError -from ._health import liveness, readiness -from ._metrics import CONTENT_TYPE, metrics_body +from ._routes import RouteContext, RouteError, allowed, route_table from ._scope import current, enter, get, leave from ._wiring import Wiring @@ -229,64 +228,57 @@ def blueprint( guard: Optional[Guard] = None, diagnostics_prefix: str = "/_config", ) -> Blueprint: - """The health, metrics and diagnostics routes, as a Quart blueprint.""" - routes = Blueprint("dynamic_config", __name__, url_prefix=prefix or None) - - @routes.get("/healthz") - async def healthz() -> Response: - """The process is up. Configuration has no say in this one.""" - report = liveness() - - return _json(report.body, report.status_code) + """The health, metrics and diagnostics routes, as a Quart blueprint. - @routes.get("/readyz") - async def readyz() -> Response: - """Serving something, and the reloads since have worked.""" - report = readiness(*wiring.configs, stale_after=stale_after) - - return _json(report.body, report.status_code) - - if metrics: + The shared table, translated: async handlers (Quart's whole point), + `` for the tail, and the WSGI-family refusal convention — + **404**, a guarded route indistinguishable from an absent one. + """ + routes = Blueprint("dynamic_config", __name__, url_prefix=prefix or None) - @routes.get("/metrics") - async def prometheus() -> Response: - """The engine's series, built per scrape.""" - return Response(metrics_body(*wiring.configs), mimetype=CONTENT_TYPE) + table = route_table( + wiring, + metrics=metrics, + stale_after=stale_after, + guard=guard, + diagnostics_prefix=diagnostics_prefix, + ) - if guard is not None and guard is not never: - diagnostics = Blueprint( - "dynamic_config_diagnostics", __name__, url_prefix=diagnostics_prefix - ) + def make(entry: Any) -> Any: + async def handler(path: str = "") -> Response: + if entry.guarded: + try: + allowed(request, guard, refused=404) + except RouteError as stopped: + return _reply_of(stopped) - @diagnostics.get("/explain/") - async def explain_path(path: str) -> Response: - """Every layer's answer for one dotted path, off the loop.""" - if not guard(request): - return _json({"detail": "not found"}, 404) + context = RouteContext(path_param=path or None, query=request.args) try: - config = wiring.config(request.args.get("config")) - except LookupError as unknown: - return _json({"detail": str(unknown)}, 400) + reply = await entry.handle_async(context) + except RouteError as stopped: + return _reply_of(stopped) - return Response(await explain_async(config, path), mimetype="text/plain") + return Response( + reply.body, status=reply.status, mimetype=reply.content_type + ) - @diagnostics.get("/check") - async def check_all() -> Response: - """Would each configuration load, and any unknown keys.""" - if not guard(request): - return _json({"detail": "not found"}, 404) + handler.__name__ = f"config_{entry.name}" - return _json( - {config.key: await check_async(config) for config in wiring.configs}, - 200, - ) + return handler - routes.register_blueprint(diagnostics) + for entry in table: + rule = entry.path.replace("{path}", "") + routes.get(rule)(make(entry)) return routes +def _reply_of(stopped: RouteError) -> Response: + """A refusal, in Quart's shape.""" + return _json({"detail": stopped.detail}, stopped.status) + + def _json(body: Any, status: int) -> Response: """A JSON response with a status. diff --git a/src/dynamic_config_web/robyn.py b/src/dynamic_config_web/robyn.py index 288646a..5d110a6 100644 --- a/src/dynamic_config_web/robyn.py +++ b/src/dynamic_config_web/robyn.py @@ -47,10 +47,9 @@ async def index(request): import threading from typing import TYPE_CHECKING, Any, Callable, Optional, TypeVar -from ._diagnostics import Guard, check_async, explain_async, never +from ._diagnostics import Guard from ._errors import MissingFrameworkError -from ._health import liveness, readiness -from ._metrics import CONTENT_TYPE, metrics_body +from ._routes import RouteContext, RouteError, allowed, route_table from ._scope import current, enter, get, leave from ._wiring import Wiring @@ -215,70 +214,55 @@ def router( # reason a user's own suite prints a deprecation warning. routes = SubRouter(prefix=prefix) - async def healthz(request: Any) -> Response: - """The process is up. Configuration has no say in this one.""" - del request - - report = liveness() - - return _json(report.body, report.status_code) - - async def readyz(request: Any) -> Response: - """Serving something, and the reloads since have worked.""" - del request - - report = readiness(*running.configs, stale_after=stale_after) - - return _json(report.body, report.status_code) - - routes.add_route(HttpMethod.GET, "/healthz", healthz) - routes.add_route(HttpMethod.GET, "/readyz", readyz) - - if metrics: - - async def prometheus(request: Any) -> Response: - """The engine's series, built per scrape.""" - del request - - return _text(metrics_body(*running.configs), CONTENT_TYPE) - - routes.add_route(HttpMethod.GET, "/metrics", prometheus) - - if guard is not None and guard is not never: + table = route_table( + running, + metrics=metrics, + stale_after=stale_after, + guard=guard, + diagnostics_prefix=diagnostics_prefix, + ) - async def explain_path(request: Any) -> Response: - """Every layer's answer for one dotted path, off the loop.""" - if not guard(request): - return _json({"detail": "not found"}, 404) + def make(entry: Any) -> Any: + async def handler(request: Any) -> Response: + if entry.guarded: + try: + allowed(request, guard, refused=404) + except RouteError as stopped: + return _json({"detail": stopped.detail}, stopped.status) + + query = dict((request.query_params.to_dict() or {}).items()) + # Robyn's query values arrive as lists. + flat = { + key: value[0] if isinstance(value, list) else value + for key, value in query.items() + } + + context = RouteContext( + path_param=request.path_params.get("path", "") or None, + query=flat, + ) try: - config = running.config( - request.query_params.get("config", None) or None - ) - except LookupError as unknown: - return _json({"detail": str(unknown)}, 400) - - path = request.path_params.get("path", "") - - return _text(await explain_async(config, path), "text/plain") + reply = await entry.handle_async(context) + except RouteError as stopped: + return _json({"detail": stopped.detail}, stopped.status) + + return Response( + status_code=reply.status, + headers=Headers({"content-type": reply.content_type}), + description=reply.body, + ) - async def check_all(request: Any) -> Response: - """Would each configuration load, and any unknown keys.""" - if not guard(request): - return _json({"detail": "not found"}, 404) + handler.__name__ = f"config_{entry.name}" - return _json( - {config.key: await check_async(config) for config in running.configs}, - 200, - ) + return handler + for entry in table: # `*path` rather than `:path`: a dotted path is one segment, but a # caller who writes `database.pool.size` should not have to know # that, and the catch-all is what makes a slash in it harmless. - routes.add_route( - HttpMethod.GET, f"{diagnostics_prefix}/explain/*path", explain_path - ) - routes.add_route(HttpMethod.GET, f"{diagnostics_prefix}/check", check_all) + rule = entry.path.replace("{path}", "*path") + routes.add_route(HttpMethod.GET, rule, make(entry)) return routes diff --git a/tests/conformance/suite.py b/tests/conformance/suite.py index 01b974e..12282a6 100644 --- a/tests/conformance/suite.py +++ b/tests/conformance/suite.py @@ -1,6 +1,6 @@ """The contract every adapter passes, written once. -Seven frameworks, one set of promises. What differs between them is the +Nine frameworks, one set of promises. What differs between them is the vocabulary — a `Depends`, an extension object, a plugin, a middleware — and what must not differ is the behaviour: one reading per request, one watcher per app lifetime, a readiness endpoint that tells *serving @@ -134,6 +134,54 @@ def case_a_request_never_tears_across_a_reload( assert client.get("/probe").json()["host"] == "reloaded-mid-request" +def case_a_scope_covers_every_configuration_it_was_given( + driver: Driver, wiring: Wiring, config_file: Path +) -> None: + """Two configurations, one scope, one reading of each. + + Every other case wires one configuration, which leaves the scope's + whole multi-configuration side — the by-key read, and the pinning of + *several* sections at once — resting on unit tests. This one builds a + second configuration beside the first, hands the driver a wiring over + a group of both, and asks the handler to read both, reload both + underneath itself, and read both again: four reads, one scope, no + movement inside the request — and the next request sees the installs. + """ + from dynamic_config import ConfigGroup, DynamicConfig + from helpers import Database + + extra_file = config_file.parent / "extra.toml" + extra_file.write_text( + '[extra]\nhost = "second.internal"\nport = 1\npool_size = 2\n' + ) + + extra = DynamicConfig(Database, key="extra").file(str(extra_file)) + pair = Wiring(ConfigGroup(wiring.configs[0], extra), watch=False) + + with pair: + with driver.client(pair) as client: + answer = client.get("/pair") + + assert answer.status_code == 200 + + body = answer.json() + + assert body["db_first"] == body["db_second"], ( + "the first configuration moved inside one request" + ) + assert body["extra_first"] == body["extra_second"], ( + "the second configuration moved inside one request" + ) + assert body["extra_first"] == "second.internal" + + # A fresh request sees what the handler's reloads installed: the + # scope pinned the last request, it did not become a cache. + with driver.client(pair) as client: + after = client.get("/pair").json() + + assert after["extra_first"] == "moved.internal" + + def case_a_missing_scope_is_refused( driver: Driver, wiring: Wiring, config_file: Path ) -> None: @@ -336,6 +384,7 @@ def case_a_watched_file_reaches_the_handler( _CASES = ( case_a_request_reads_the_configuration, case_a_request_never_tears_across_a_reload, + case_a_scope_covers_every_configuration_it_was_given, case_a_missing_scope_is_refused, case_the_watcher_is_paired_with_the_app, case_building_the_app_twice_does_not_collide, diff --git a/tests/conformance/test_django.py b/tests/conformance/test_django.py index a405bd3..0d10c5d 100644 --- a/tests/conformance/test_django.py +++ b/tests/conformance/test_django.py @@ -89,6 +89,30 @@ def tear(request: Any) -> JsonResponse: ) +def pair(request: Any) -> JsonResponse: + """Every configuration in the wiring, pinned by one scope.""" + del request + + wiring = dj.wiring() + first = {c.key: current(c).host for c in wiring.configs} + + for member in wiring.configs: + if member.key != "db": + member.set_override("host", "moved.internal") + member.reload() + + second = {c.key: current(c).host for c in wiring.configs} + + return JsonResponse( + { + "db_first": first["db"], + "db_second": second["db"], + "extra_first": first.get("extra"), + "extra_second": second.get("extra"), + } + ) + + class DjangoDriver: """What the shared suite needs to know about Django.""" @@ -114,6 +138,7 @@ def client( *type(self).routes(install), path("probe", probe), path("tear", tear), + path("pair", pair), ] ) diff --git a/tests/conformance/test_django_bolt.py b/tests/conformance/test_django_bolt.py index 0abd450..c1fd6dd 100644 --- a/tests/conformance/test_django_bolt.py +++ b/tests/conformance/test_django_bolt.py @@ -74,6 +74,25 @@ async def tear() -> dict[str, Any]: "same_object": first is second, } + @bolt.get("/pair") + async def pair() -> dict[str, Any]: + """Every configuration in the wiring, pinned by one scope.""" + first = {c.key: current(c).host for c in wiring.configs} + + for member in wiring.configs: + if member.key != "db": + member.set_override("host", "moved.internal") + member.reload() + + second = {c.key: current(c).host for c in wiring.configs} + + return { + "db_first": first["db"], + "db_second": second["db"], + "extra_first": first.get("extra"), + "extra_second": second.get("extra"), + } + with BoltClient(bolt) as client: yield client diff --git a/tests/conformance/test_fastapi.py b/tests/conformance/test_fastapi.py index 07d440e..59c80bd 100644 --- a/tests/conformance/test_fastapi.py +++ b/tests/conformance/test_fastapi.py @@ -73,6 +73,30 @@ def tear() -> dict[str, Any]: "same_object": first is second, } + @app.get("/pair") + def pair() -> dict[str, Any]: + """Every configuration in the wiring, read under one scope. + + Reads both, moves both underneath itself, reads both again: + a scope over several configurations pins all of them, not + just the first. + """ + first = {c.key: current(c).host for c in wiring.configs} + + for member in wiring.configs: + if member.key != "db": + member.set_override("host", "moved.internal") + member.reload() + + second = {c.key: current(c).host for c in wiring.configs} + + return { + "db_first": first["db"], + "db_second": second["db"], + "extra_first": first.get("extra"), + "extra_second": second.get("extra"), + } + with TestClient(app) as client: yield client diff --git a/tests/conformance/test_flask.py b/tests/conformance/test_flask.py index e2ef5d3..abf6a31 100644 --- a/tests/conformance/test_flask.py +++ b/tests/conformance/test_flask.py @@ -102,6 +102,25 @@ def tear() -> dict[str, Any]: "same_object": first is second, } + @app.get("/pair") + def pair() -> dict[str, Any]: + """Every configuration in the wiring, pinned by one scope.""" + first = {c.key: current(c).host for c in wiring.configs} + + for member in wiring.configs: + if member.key != "db": + member.set_override("host", "moved.internal") + member.reload() + + second = {c.key: current(c).host for c in wiring.configs} + + return { + "db_first": first["db"], + "db_second": second["db"], + "extra_first": first.get("extra"), + "extra_second": second.get("extra"), + } + try: with app.test_client() as inner: yield Client(inner) diff --git a/tests/conformance/test_litestar.py b/tests/conformance/test_litestar.py index 1141e24..c1bc60f 100644 --- a/tests/conformance/test_litestar.py +++ b/tests/conformance/test_litestar.py @@ -66,8 +66,27 @@ def tear() -> dict[str, Any]: "same_object": first is second, } + @get("/pair", sync_to_thread=False) + def pair() -> dict[str, Any]: + """Every configuration in the wiring, pinned by one scope.""" + first = {c.key: current(c).host for c in wiring.configs} + + for member in wiring.configs: + if member.key != "db": + member.set_override("host", "moved.internal") + member.reload() + + second = {c.key: current(c).host for c in wiring.configs} + + return { + "db_first": first["db"], + "db_second": second["db"], + "extra_first": first.get("extra"), + "extra_second": second.get("extra"), + } + app = Litestar( - [probe, tear], + [probe, tear, pair], plugins=[ DynamicConfigPlugin( wiring, diff --git a/tests/conformance/test_quart.py b/tests/conformance/test_quart.py index 8961881..91b5333 100644 --- a/tests/conformance/test_quart.py +++ b/tests/conformance/test_quart.py @@ -110,6 +110,25 @@ async def tear() -> dict[str, Any]: "same_object": first is second, } + @app.get("/pair") + async def pair() -> dict[str, Any]: + """Every configuration in the wiring, pinned by one scope.""" + first = {c.key: current(c).host for c in wiring.configs} + + for member in wiring.configs: + if member.key != "db": + member.set_override("host", "moved.internal") + member.reload() + + second = {c.key: current(c).host for c in wiring.configs} + + return { + "db_first": first["db"], + "db_second": second["db"], + "extra_first": first.get("extra"), + "extra_second": second.get("extra"), + } + loop = asyncio.new_event_loop() try: diff --git a/tests/conformance/test_robyn.py b/tests/conformance/test_robyn.py index fc98a15..feef95a 100644 --- a/tests/conformance/test_robyn.py +++ b/tests/conformance/test_robyn.py @@ -102,6 +102,28 @@ async def tear(request: Any) -> dict[str, Any]: "same_object": first is second, } + @app.get("/pair") + @scoped + async def pair(request: Any) -> dict[str, Any]: + """Every configuration in the wiring, pinned by one scope.""" + del request + + first = {c.key: current(c).host for c in wiring.configs} + + for member in wiring.configs: + if member.key != "db": + member.set_override("host", "moved.internal") + member.reload() + + second = {c.key: current(c).host for c in wiring.configs} + + return { + "db_first": first["db"], + "db_second": second["db"], + "extra_first": first.get("extra"), + "extra_second": second.get("extra"), + } + client = RobynClient(app) # What the server does around the request loop, and what the test diff --git a/tests/test_core.py b/tests/test_core.py index 56143ad..8f36447 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -545,3 +545,248 @@ def test_importing_the_core_imports_no_framework() -> None: def test_the_surface_is_sorted() -> None: assert list(web.__all__) == sorted(web.__all__) assert all(hasattr(web, name) for name in web.__all__) + + +# ── the scope's generation check ───────────────────────────────────────── +# +# With two configurations a scope is two reads over two independent atomic +# cells, and a reload landing between them puts two generations in one +# request. `enter()` reads the install counters before and after, and +# starts over when anything moved — the same check, with the same retry +# budget, as the Rust web core's `Sections::take`. These fakes script the +# counter so both branches run deterministically; the conformance suite's +# `multi_config_scope` case covers the same property through every +# adapter, and the stress test below covers it under real threads. + + +class _Scripted: + """A configuration whose generation moves on a script. + + `try_current` answers a model stamped with the generation it was read + at, so a torn snapshot is visible as two different stamps. + """ + + def __init__(self, key: str, moves_during_reads: int) -> None: + self.key = key + self._generation = 1 + self._reads = 0 + self._moves = moves_during_reads + + @property + def generation(self) -> int: + return self._generation + + def try_current(self) -> tuple[str, int]: + self._reads += 1 + # A "reload" lands after this read and before the re-check, as + # many times as the script says. + if self._moves > 0: + self._moves -= 1 + self._generation += 1 + return (self.key, self._generation) + + +def test_a_scope_over_two_configs_retries_until_the_reads_agree() -> None: + from dynamic_config_web import _scope + + steady = _Scripted("a", moves_during_reads=0) + moving = _Scripted("b", moves_during_reads=1) + + token = _scope.enter([steady, moving]) + try: + snapshot = _scope._SNAPSHOT.get() + assert snapshot is not None + # The disturbed first read was refused; the second agreed. + assert snapshot.by_key("b")[1] == moving.generation + assert moving._reads == 2, "one retry, exactly" + finally: + _scope.leave(token) + + +def test_a_scope_that_cannot_win_serves_the_last_read() -> None: + from dynamic_config_web import _scope + + steady = _Scripted("a", moves_during_reads=0) + # Move on every read the budget allows, and then one more for the + # final unchecked read: the scope still answers rather than raising. + restless = _Scripted("b", moves_during_reads=_scope._ATTEMPTS + 1) + + token = _scope.enter([steady, restless]) + try: + snapshot = _scope._SNAPSHOT.get() + assert snapshot is not None + assert snapshot.by_key("a") is not None + assert snapshot.by_key("b") is not None + assert restless._reads == _scope._ATTEMPTS + 1 + finally: + _scope.leave(token) + + +def test_a_single_config_scope_never_pays_for_the_check() -> None: + from dynamic_config_web import _scope + + alone = _Scripted("a", moves_during_reads=0) + + token = _scope.enter([alone]) + try: + assert alone._reads == 1, "one section cannot straddle anything" + finally: + _scope.leave(token) + + +def test_two_real_configs_never_tear_under_a_reload_storm( + tmp_path: Path, +) -> None: + """Real engine, real threads: a scope never *mixes* across an install. + + What the generation check promises — and all it promises — is that no + install lands between a scope's reads. Two configurations reloaded at + different moments may legitimately sit at different versions, and a + scope opened in that window correctly reports the split world; no + check short of a cross-config epoch could promise otherwise. + + So the assertion is the invariant tearing alone can break. The writer + always installs `left` before `right`; the reader reads `left` before + `right` too. Every stable world therefore has `right <= left` — the + only way a scope can see `right` NEWER than `left` is by reading + `left` before an install pair and `right` after it, which is exactly + the mixed read the check retries away. + """ + import json + import threading + import time + from dataclasses import dataclass + + path = tmp_path / "config.json" + + def write_counter(n: int) -> None: + path.write_text(json.dumps({"left": {"n": n}, "right": {"n": n}})) + + write_counter(0) + + @dataclass + class Half: + n: int = 0 + + left = DynamicConfig(Half, key="left").file(str(path)) + right = DynamicConfig(Half, key="right").file(str(path)) + left.init() + right.init() + + stop = threading.Event() + inversions: list[tuple[int, int]] = [] + + # Two watchers never fire at the same instant, so the writer installs + # the halves with a real gap between them — the window a torn scope + # falls into. A short switch interval makes the scheduler actually + # interleave the readers with it. + previous_interval = sys.getswitchinterval() + sys.setswitchinterval(1e-5) + + def reader() -> None: + while not stop.is_set(): + with web.scope(left, right): + a = web.current(left).n + b = web.current(right).n + if b > a: + inversions.append((a, b)) + + def writer() -> None: + n = 0 + while not stop.is_set(): + n += 1 + write_counter(n) + # Back to back: a reader descheduled between its two reads can + # then straddle the whole install pair, which is the mix the + # check exists to refuse. + left.reload() + right.reload() + + threads = [threading.Thread(target=reader) for _ in range(4)] + threads.append(threading.Thread(target=writer)) + for thread in threads: + thread.start() + + try: + time.sleep(0.5) + finally: + stop.set() + for thread in threads: + thread.join(timeout=5) + sys.setswitchinterval(previous_interval) + + assert not inversions, f"a scope mixed reads across an install: {inversions[:5]}" + + +def test_a_scope_over_an_unloaded_config_defers_to_the_engine( + tmp_path: Path, +) -> None: + """A config that had not loaded when the request began is not pinned. + + The scope holds `None` for it, and `current()` falls through to the + engine — raising `NotInitialisedError` exactly as an unscoped read + would, and answering the live model once a load lands mid-request. + Deliberate, and worth a test: the fallback means such a request is + *not* isolated from a reload, which is different from every loaded + configuration in the same scope. + """ + from dynamic_config import NotInitialisedError + + path = tmp_path / "late.toml" + path.write_text('[late]\nhost = "late.internal"\nport = 5432\npool_size = 8\n') + + late = DynamicConfig(Database, key="late").file(str(path)) + + with web.scope(late): + # Nothing loaded yet: the engine's own refusal comes through. + with pytest.raises(NotInitialisedError): + web.current(late) + + # A load landing mid-request becomes visible — the unloaded slot + # defers, it does not pin. + late.init() + + assert web.current(late).host == "late.internal" + + +def test_concurrent_async_requests_each_hold_their_own_scope( + wiring: Wiring, config_file: Path +) -> None: + """Scopes are task-local: N tasks, N snapshots, no bleed. + + `contextvars` promises it; this holds the promise under an actual + reload landing while half the tasks are mid-"request". + """ + import asyncio + + config = wiring.configs[0] + + async def request(delay: float) -> tuple[str, str]: + with web.scope(config): + first = web.current(config).host + await asyncio.sleep(delay) + second = web.current(config).host + + return first, second + + async def storm() -> list[tuple[str, str]]: + early = [asyncio.create_task(request(0.1)) for _ in range(25)] + await asyncio.sleep(0.02) + + write(config_file, host="moved-mid-flight") + config.reload() + + late = [asyncio.create_task(request(0.0)) for _ in range(25)] + + return await asyncio.gather(*early, *late) + + results = asyncio.run(storm()) + early, late = results[:25], results[25:] + + for first, second in results: + assert first == second, "a task's scope moved under it" + + assert all(first == "db.internal" for first, _ in early), ( + "a task that began before the reload saw the new document" + ) + assert all(first == "moved-mid-flight" for first, _ in late) diff --git a/tests/test_events.py b/tests/test_events.py new file mode 100644 index 0000000..7a42fe3 --- /dev/null +++ b/tests/test_events.py @@ -0,0 +1,183 @@ +"""The reload log lines and the event stream — `_events.py`, from outside. + +The module had shipped without a test of its own: `log_reloads` was read +but never asserted, and `stream_events` was exercised only by an example. +These pin the two things a consumer relies on: what a line *says* (paths, +never values) and what an event *carries*. +""" + +from __future__ import annotations + +import asyncio +import logging +from pathlib import Path + +import pytest + +from dynamic_config import ConfigGroup, DynamicConfig +from dynamic_config_web import log_reloads, stream_events +from helpers import Database, write + + +def test_log_reloads_names_the_paths_that_moved( + wiring, config_file: Path, caplog: pytest.LogCaptureFixture +) -> None: + config = wiring.configs[0] + + guards = log_reloads(config) + + try: + with caplog.at_level(logging.INFO, logger="dynamic_config"): + write(config_file, host="moved", port=5433) + config.reload() + + assert len(caplog.records) == 1 + line = caplog.records[0].getMessage() + + assert "db" in line + assert "host" in line + assert "port" in line + finally: + for guard in guards: + guard.close() + + +def test_log_reloads_carries_no_value( + wiring, config_file: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Paths, never values — the module's own first rule.""" + config = wiring.configs[0] + + guards = log_reloads(config) + + try: + with caplog.at_level(logging.INFO, logger="dynamic_config"): + write(config_file, host="hunter2.internal", port=4242) + config.reload() + + text = "\n".join(record.getMessage() for record in caplog.records) + + assert "hunter2" not in text + assert "4242" not in text + finally: + for guard in guards: + guard.close() + + +def test_log_reloads_takes_a_logger_and_a_level( + wiring, config_file: Path, caplog: pytest.LogCaptureFixture +) -> None: + config = wiring.configs[0] + mine = logging.getLogger("test.reloads") + + guards = log_reloads(config, mine, level=logging.WARNING) + + try: + with caplog.at_level(logging.WARNING, logger="test.reloads"): + write(config_file, host="elsewhere") + config.reload() + + assert caplog.records, "the line went to the wrong logger" + assert caplog.records[0].levelno == logging.WARNING + assert caplog.records[0].name == "test.reloads" + finally: + for guard in guards: + guard.close() + + +def test_log_reloads_covers_every_member_of_a_group( + tmp_path: Path, dynamic_config_wiring, caplog: pytest.LogCaptureFixture +) -> None: + first_file = tmp_path / "first.toml" + second_file = tmp_path / "second.toml" + first_file.write_text('[db]\nhost = "a"\nport = 1\npool_size = 1\n') + second_file.write_text('[extra]\nhost = "b"\nport = 2\npool_size = 2\n') + + group = ConfigGroup( + DynamicConfig(Database, key="db").file(str(first_file)), + DynamicConfig(Database, key="extra").file(str(second_file)), + ) + wiring = dynamic_config_wiring(group, watch=False) + + guards = log_reloads(group) + + try: + assert len(guards) == 2, "one hook per member" + + with caplog.at_level(logging.INFO, logger="dynamic_config"): + second_file.write_text('[extra]\nhost = "c"\nport = 2\npool_size = 2\n') + wiring.configs[1].reload() + + lines = [record.getMessage() for record in caplog.records] + + assert any("extra" in line for line in lines) + assert not any('"c"' in line for line in lines) + finally: + for guard in guards: + guard.close() + + +def test_stream_events_reports_an_install(wiring, config_file: Path) -> None: + config = wiring.configs[0] + + async def one_event() -> dict: + stream = stream_events(config, failure_poll=None) + + async def consume() -> dict: + async for event in stream: + return event + raise AssertionError("the stream ended") + + task = asyncio.ensure_future(consume()) + + # Let the consumer subscribe before the install lands. + await asyncio.sleep(0.05) + write(config_file, host="streamed") + config.reload() + + return await asyncio.wait_for(task, timeout=5) + + event = asyncio.run(one_event()) + + assert event["type"] == "reloaded" + assert event["key"] == "db" + assert event["generation"] == 2 + assert "host" in event["changed"] + assert "streamed" not in str(event), "an event carried a value" + + +def test_stream_events_reports_a_refusal(wiring, config_file: Path) -> None: + """A load that installed nothing still reaches the stream. + + Nothing bumps a generation on a refusal, so `failure_poll` is the + only wake-up — this is the path that exists for it. + """ + config = wiring.configs[0] + + async def first_failure() -> dict: + stream = stream_events(config, failure_poll=0.05) + + async def consume() -> dict: + async for event in stream: + if event["type"] == "reload_failed": + return event + raise AssertionError("the stream ended") + + task = asyncio.ensure_future(consume()) + + await asyncio.sleep(0.05) + config_file.write_text("this is not toml [") + + import contextlib + + with contextlib.suppress(Exception): + config.reload() # the refusal is the point + + return await asyncio.wait_for(task, timeout=5) + + event = asyncio.run(first_failure()) + + assert event["type"] == "reload_failed" + assert event["key"] == "db" + assert event["kind"] == "parse" + assert event["consecutive"] >= 1