API Reference¶

class Auth[THeaders: Struct, TUser: Struct](Protocol):
    """Implement ``authenticate``; raise ``HTTPError(401, ...)`` to reject.

    ``headers`` is bound from the request headers into your declared
    Struct (header names map ``x-trace-id`` -> ``x_trace_id``). The
    returned Struct is what handlers receive as ``user``.
    """

    def authenticate(self, headers: THeaders) -> TUser | Awaitable[TUser]:
        """Validate ``headers`` and return the user Struct; raise ``HTTPError(401)`` to reject."""
        ...  # pylint: disable=unnecessary-ellipsis  # Protocol stub; pyright needs the body

class BackgroundTasks:
    """A queue of fire-and-forget background work, dispatched by item type.

    Register one handler per ``Struct`` type (inferred from the handler's parameter);
    endpoints call :meth:`add` to enqueue an item, and a single serial worker dispatches
    each to its handler. Open it with ``self._aenter`` inside ``_wire`` so the worker
    starts at startup and drains/stops at shutdown.

    Enter it *after* the resources its handlers use, so reverse-order shutdown drains the
    queue before those resources are torn down.
    """

    def __init__(
        self,
        *,
        maxsize: int = 1024,
        drain_timeout: float | None = 30.0,
        allow_one_to_many: bool = False,
    ) -> None:
        self._queue: asyncio.Queue[Struct] = asyncio.Queue(maxsize)
        self._drain_timeout = drain_timeout
        self._allow_one_to_many = allow_one_to_many
        self._handlers: dict[type[Struct], list[Callable[[Any], Awaitable[None]]]] = {}
        self._worker: asyncio.Task[None] | None = None

    async def _run(self) -> None:
        """Pull items forever, dispatching each to its handler(s). Errors are isolated."""
        while True:
            item = await self._queue.get()
            try:
                handlers = self._handlers.get(type(item))
                if not handlers:
                    logger.error("background: no handler registered for %s", type(item).__name__)
                    continue
                for handler in handlers:
                    try:
                        await handler(item)
                    except Exception:  # pylint: disable=broad-exception-caught
                        logger.exception("background: handler failed for %s", type(item).__name__)
            finally:
                self._queue.task_done()

    def register[T: Struct](self, handler: Callable[[T], Awaitable[None]]) -> None:
        """Register a handler; its item type is inferred from its single parameter.

        One handler per type by default — a second for the same type is a ``WiringError``
        unless this was built with ``allow_one_to_many=True``.
        """
        item_type = _infer_item_type(handler)
        if item_type in self._handlers and not self._allow_one_to_many:
            raise WiringError(
                f"BackgroundTasks: a handler is already registered for {item_type.__name__!r}; "
                f"pass allow_one_to_many=True to register more than one",
            )
        self._handlers.setdefault(item_type, []).append(handler)

    async def add(self, item: Struct) -> None:
        """Enqueue an item for background processing (awaits if the queue is full)."""
        await self._queue.put(item)

    async def __aenter__(self) -> Self:
        self._worker = asyncio.create_task(self._run())
        return self

    async def __aexit__(self, *_exc: object) -> None:
        if self._worker is None:
            return
        if self._drain_timeout is not None:
            try:
                await asyncio.wait_for(self._queue.join(), self._drain_timeout)
            except TimeoutError:
                logger.warning(
                    "background: drain timed out after %ss; dropping %d queued item(s)",
                    self._drain_timeout,
                    self._queue.qsize(),
                )
        self._worker.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._worker

class BaseApp[FactoryT = None](_StackScope, ABC):
    """Subclass and override ``_wire`` to open resources and include resources/endpoints.

    The app owns the two exit stacks. Parameterize with a factory class —
    ``class MyApp(BaseApp[MyFactory])`` — and the app builds it at construction,
    injecting the stacks the factory's ``__init__`` names (``es`` for the
    ExitStack, ``aes`` for the AsyncExitStack). The built factory is then
    ``self._factory`` (typed as ``MyFactory``) inside ``_wire``, and any resource
    it registers on those stacks is closed at shutdown.

    Pass ``factory=`` to supply a prebuilt factory instead of building one — the
    seam for tests, which inject a ``create_autospec`` stand-in
    (``MyApp(factory=mock_factory)``) so the real services are never constructed.

    Reverse-routed ``Location`` / ``Link`` URLs are relative unless the environment sets
    ``JERO_BASE_URL`` (a static public origin) or ``JERO_TRUST_FORWARDED`` (rebuild the
    origin per request from ``X-Forwarded-*``); see :func:`_forwarded_config_from_env`.
    """

    def __init__(self, *, factory: FactoryT | None = None) -> None:
        self._static: _StaticRoutes = {}
        self._dynamic: _DynamicRoutes = {}
        self._allowed: _AllowedMethods = {}
        self._allow_cache: dict[str, bytes] = {}
        self._decoders: dict[type[Struct], Decoder[Struct]] = {}
        base_url, trust_forwarded = _forwarded_config_from_env()
        self._reverser = _Reverser(base_url=base_url, trust_forwarded=trust_forwarded)
        self._stack = ExitStack()
        self._astack = AsyncExitStack()
        self._factory: FactoryT = factory if factory is not None else self._make_factory()

    def _decoder(self, struct_type: type[Struct]) -> Decoder[Struct]:
        """The reusable typed JSON decoder for ``struct_type``, built once per app.

        Decoders are keyed by type, so models shared across handlers (a ``WidgetIn``
        used by both ``create`` and ``update``) share one decoder. Populated only at
        wiring time; the binder holds the resolved decoder, so the request path does
        no lookup.
        """
        if struct_type not in self._decoders:
            self._decoders[struct_type] = Decoder(struct_type)
        return self._decoders[struct_type]

    def _resolve_factory_type(self) -> type | None:
        """The factory class from ``BaseApp[...]``, or None if unparameterized."""
        for base in get_original_bases(type(self)):
            if get_origin(base) is BaseApp:
                args = get_args(base)
                if args and isinstance(args[0], type) and args[0] is not type(None):
                    return args[0]
        return None

    def _make_factory(self) -> FactoryT:
        factory_type = self._resolve_factory_type()
        if factory_type is None:
            return cast("FactoryT", None)
        return cast("FactoryT", instantiate_factory(factory_type, self._stack, self._astack))

    @abstractmethod
    async def _wire(self) -> None:
        """Override to open resources (via ``_enter`` / ``_aenter``) and include them.

        Runs once at startup. Anything entered via the helpers is torn
        down (in reverse order) at shutdown.

        Abstract: every ``BaseApp`` subclass must implement it. A subclass that
        omits it is flagged at its instantiation site by the type checker.
        """

    def _register(self, method: _HttpMethod, segments: list[_Segment], handler: _Handler) -> None:
        params = tuple((i, value) for i, (is_param, value) in enumerate(segments) if is_param)
        if not params:
            route_path = "/".join(value for _, value in segments)
            if (method, route_path) in self._static:
                raise WiringError(f"{method} {route_path} is already registered")
            self._static[(method, route_path)] = handler
            self._allowed.setdefault(route_path, []).append(method)
            return

        statics = tuple((i, value) for i, (is_param, value) in enumerate(segments) if not is_param)
        bucket = self._dynamic.setdefault((method, len(segments)), [])
        if any(pattern.statics == statics for pattern in bucket):
            raise WiringError(f"{method} {_template_str(segments)} is already registered")
        bucket.append(_Pattern(statics, params, handler))

    @staticmethod
    def _check_user_source(
        resource_cls: type,
        name: str,
        user_type: type[Struct] | None,
        auth: _CompiledAuth | None,
    ) -> None:
        if user_type is None:
            return
        if auth is None:
            raise WiringError(
                f"{resource_cls.__name__}.{name} declares 'user' but no auth was given",
            )
        if not issubclass(auth.returns, user_type):
            raise WiringError(
                f"{resource_cls.__name__}.{name}: 'user' expects {user_type.__name__} "
                f"but {auth.owner}.authenticate returns {auth.returns.__name__}",
            )

    def _include(
        self,
        obj: Resource | Endpoint,
        methods: dict[str, _Verb],
        *,
        auth: Auth[Any, Any] | None,
    ) -> None:
        cls = type(obj)
        path = getattr(cls, "path", None)
        if path is None:
            raise WiringError(
                f"{cls.__name__}: no path — declare it on the class, "
                f"e.g. `class {cls.__name__}(..., path='/...')`.",
            )
        template = _parse_template(path)
        compiled_auth = _CompiledAuth(auth) if auth is not None else None

        registered = False
        for name, verb in methods.items():
            fn = getattr(obj, name, None)
            if fn is None:
                continue
            sources = _bind_sources(cls, name, fn, verb.method, self._decoder)
            self._check_user_source(cls, name, sources.user, compiled_auth)
            segments = _route_segments(
                cls, name, template, sources.path, extends_path=verb.extends_path
            )
            handler = _Route(fn, verb.success_status, sources, compiled_auth, self._reverser)
            self._register(verb.method, segments, handler)
            self._reverser.register(
                fn.__func__, cls.ref, name, _RouteRef(tuple(segments), sources.path)
            )
            registered = True

        if not registered:
            raise WiringError(f"{cls.__name__} defines none of: {', '.join(methods)}")

    def _include_resource[THeaders: Struct, TUser: Struct](
        self,
        resource: Resource,
        *,
        auth: Auth[THeaders, TUser] | None = None,
    ) -> None:
        self._include(resource, Resource.METHODS, auth=auth)

    def _include_endpoint[THeaders: Struct, TUser: Struct](
        self,
        endpoint: Endpoint,
        *,
        auth: Auth[THeaders, TUser] | None = None,
    ) -> None:
        self._include(endpoint, Endpoint.METHODS, auth=auth)

    def _resolve(self, method: str, path: str) -> tuple[_Handler, dict[str, str]] | None:
        # Static hit is the hot path: look it up directly, before narrowing the verb
        # (a non-route method simply misses). The cast is paid only on the dynamic path.
        handler = self._static.get((method, path))
        if handler is not None:
            return handler, {}
        segments = path.split("/")
        verb = cast("_HttpMethod", method)
        for pattern in self._dynamic.get((verb, len(segments)), ()):
            if pattern.matches(segments):
                values = {name: unquote(segments[i]) for i, name in pattern.params}
                return pattern.handler, values
        return None

    def _allowed_methods(self, path: str) -> tuple[_HttpMethod, ...]:
        allowed = list(self._allowed.get(path, ()))
        segments = path.split("/")
        for (method, count), bucket in self._dynamic.items():
            if (
                count == len(segments)
                and method not in allowed
                and any(pattern.matches(segments) for pattern in bucket)
            ):
                allowed.append(method)
        return tuple(allowed)

    def _allow_for(self, path: str) -> bytes | None:
        """The Allow header for a path, or None if no route shape matches it."""
        cached = self._allow_cache.get(path)
        if cached is not None:
            return cached
        allowed = self._allowed_methods(path)
        return _allow_header(allowed) if allowed else None

    def _finalize(self) -> None:
        """Precompute Allow headers for all static paths; runs once after wiring."""
        self._allow_cache = {
            path: _allow_header(self._allowed_methods(path)) for path in self._allowed
        }

    async def _close_resources(self) -> None:
        await self._astack.aclose()
        self._stack.close()

    async def _handle_lifespan(self, receive: Receive, send: Send) -> None:
        await receive()  # lifespan.startup
        try:
            await self._wire()
        except BaseException as exc:
            await self._close_resources()  # release anything entered before the failure
            await send(
                {
                    "type": "lifespan.startup.failed",
                    "message": f"{type(exc).__name__}: {exc}",
                },
            )
            raise
        self._finalize()
        await send({"type": "lifespan.startup.complete"})

        await receive()  # lifespan.shutdown
        try:
            await self._close_resources()
        except BaseException as exc:
            await send(
                {
                    "type": "lifespan.shutdown.failed",
                    "message": f"{type(exc).__name__}: {exc}",
                },
            )
            raise
        await send({"type": "lifespan.shutdown.complete"})

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            if scope["type"] == "lifespan":
                await self._handle_lifespan(receive, send)
                return
            raise RuntimeError(f"unsupported scope type {scope['type']!r}")

        # HTTP is the hot path; inlined here (was _handle_http) to save a coroutine hop.
        method: str = scope["method"]
        path: str = scope["path"]
        resolved = self._resolve("GET" if method == "HEAD" else method, path)
        if resolved is not None:
            handler, path_values = resolved
            await handler(
                scope, receive, _SuppressBody(send) if method == "HEAD" else send, path_values
            )
            return

        allow = self._allow_for(path)
        if allow is None:
            await _send_json(send, 404, b'{"error":"not found"}')
        elif method == "OPTIONS":
            await send(
                {"type": "http.response.start", "status": 204, "headers": [(b"allow", allow)]}
            )
            await send({"type": "http.response.body", "body": b""})
        else:
            await _send_json(send, 405, b'{"error":"method not allowed"}', [(b"allow", allow)])

class BaseFactory(_StackScope):
    """Base for an app's factory. Subclass and add ``create_*`` methods that
    build services with ``self._enter`` / ``self._aenter``.

    The app injects its exit stacks (``es`` / ``aes``); anything opened via the
    helpers is closed when the app shuts down.
    """

    def __init__(self, es: ExitStack, aes: AsyncExitStack) -> None:
        self._stack = es
        self._astack = aes

@dataclass(kw_only=True, slots=True)
class BytesResponse[H: Struct | None = None](BaseResponse[H]):
    """Raw bytes; content-type defaults to application/octet-stream."""

    content: bytes

class Endpoint(_Routable):
    """One HTTP endpoint at a single path: subclass and define any of
    ``get`` / ``post`` / ``put`` / ``patch`` / ``delete``.

    Unlike :class:`Resource` there are no CRUD semantics — the method name
    *is* the verb, every verb returns 200, and the path is exact (no
    trailing extension). A different path is a different ``Endpoint``.

    Optional OpenAPI metadata is declared at class definition: ``meta`` applies to every
    operation, ``meta_<verb>`` to one (``meta_get``, ``meta_post``, …).
    """

    METHODS: ClassVar[dict[str, _Verb]] = {
        "get": _Verb("GET", 200, extends_path=False),
        "post": _Verb("POST", 200, extends_path=False),
        "put": _Verb("PUT", 200, extends_path=False),
        "patch": _Verb("PATCH", 200, extends_path=False),
        "delete": _Verb("DELETE", 200, extends_path=False),
    }

    meta: ClassVar[EndpointMeta | None] = None
    meta_get: ClassVar[OperationMeta | None] = None
    meta_post: ClassVar[OperationMeta | None] = None
    meta_put: ClassVar[OperationMeta | None] = None
    meta_patch: ClassVar[OperationMeta | None] = None
    meta_delete: ClassVar[OperationMeta | None] = None

    def __init_subclass__(
        cls,
        *,
        path: str,
        ref: str | None = None,
        meta: EndpointMeta | None = None,
        meta_get: OperationMeta | None = None,
        meta_post: OperationMeta | None = None,
        meta_put: OperationMeta | None = None,
        meta_patch: OperationMeta | None = None,
        meta_delete: OperationMeta | None = None,
        **kwargs: object,
    ) -> None:
        # path / ref handling lives on _Routable
        super().__init_subclass__(path=path, ref=ref, **kwargs)
        _validate_meta(
            cls,
            meta,
            EndpointMeta,
            {
                "meta_get": meta_get,
                "meta_post": meta_post,
                "meta_put": meta_put,
                "meta_patch": meta_patch,
                "meta_delete": meta_delete,
            },
        )
        cls.meta = meta
        cls.meta_get = meta_get
        cls.meta_post = meta_post
        cls.meta_put = meta_put
        cls.meta_patch = meta_patch
        cls.meta_delete = meta_delete

class EndpointMeta(Struct):
    """OpenAPI metadata shared by all of an ``Endpoint``'s operations."""

    tags: Sequence[str] = ()

class FactoryHarness[FactoryT: BaseFactory]:
    """Build a factory in isolation and exercise its ``create_*`` methods.

    The factory-level sibling of :class:`TestClient`. It owns the exit stacks the
    factory writes to via ``_enter`` / ``_aenter`` and runs async ``create_*``
    methods on a background loop, so services are built — and their resources
    opened and torn down — exactly as under a live app, but with no app, routes,
    or server. Use it to test the real factory wiring that an app's ``factory=``
    seam mocks away.

        with FactoryHarness(Factory) as harness:
            service = harness.run(harness.factory.create_widget_service())
            assert isinstance(service, WidgetService)
        # everything opened on the stacks is closed here

    Synchronous ``create_*`` methods can be called directly on ``harness.factory``;
    ``run`` awaits the async ones on the harness's loop.
    """

    def __init__(self, factory_cls: type[FactoryT]) -> None:
        self._loop_thread = _LoopThread()
        self._stack = ExitStack()
        self._astack = AsyncExitStack()
        self.factory: FactoryT = instantiate_factory(factory_cls, self._stack, self._astack)

    def run[T](self, coro: Coroutine[Any, Any, T]) -> T:
        """Await an async ``create_*`` coroutine on the harness's loop."""
        return self._loop_thread.submit(coro)

    async def _close_stacks(self) -> None:
        await self._astack.aclose()
        self._stack.close()

    def close(self) -> None:
        """Close everything the factory opened on its exit stacks, then stop the loop."""
        self._loop_thread.submit(self._close_stacks())
        self._loop_thread.close()

    def __enter__(self) -> Self:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()

class FilePart[H: Struct | None = None](FormPart[bytes, H]):
    """A file upload part with a required filename."""

    filename: str

class FormPart[T, H: Struct | None = None](Struct):
    """One multipart form part with envelope metadata."""

    data: T
    content_type: str | None
    headers: H
    raw_headers: _RawHeaders

class HTTPError(Exception):
    """Raise from a handler to return a JSON error response."""

    def __init__(self, status: int, detail: str) -> None:
        super().__init__(detail)
        self.status = status
        self.detail = detail

@dataclass(kw_only=True, slots=True)
class JSONResponse[T: Struct, H: Struct | None = None](BaseResponse[H]):
    """A Struct encoded as JSON; content-type defaults to application/json."""

    json: T

@dataclass(frozen=True, slots=True)
class Link:
    """An RFC 8288 web link. A list of links joins into one ``Link`` header. ``rel`` is
    required; ``title`` and ``media_type`` (emitted as ``type=``) are optional."""

    target: Target
    rel: str
    title: str | None = None
    media_type: str | None = None

    @classmethod
    def from_operation(
        cls,
        operation: Callable[..., object],
        *,
        rel: str,
        path: Struct | None = None,
        title: str | None = None,
        media_type: str | None = None,
    ) -> Self:
        """Link to a mounted operation with relation ``rel``; ``path`` fills its slots."""
        _validate_operation_path(operation, path)
        return cls(OperationTarget(operation, path), rel, title, media_type)

    @classmethod
    def from_url(
        cls, url: str, *, rel: str, title: str | None = None, media_type: str | None = None
    ) -> Self:
        """Link to a fully-qualified URL, used verbatim — never rewritten."""
        return cls(URLTarget(url), rel, title, media_type)

    @classmethod
    def from_path(
        cls, path: str, *, rel: str, title: str | None = None, media_type: str | None = None
    ) -> Self:
        """Link to a root-relative path; it picks up the app's URL base the same way a
        reversed operation does."""
        return cls(PathTarget(path), rel, title, media_type)

    @classmethod
    def from_ref(
        cls,
        ref: str,
        *,
        rel: str,
        path: Struct | None = None,
        title: str | None = None,
        media_type: str | None = None,
    ) -> Self:
        """Link to an operation by its class ``ref`` (``"name.operation"``) — the
        import-cycle hatch; prefer ``from_operation`` otherwise."""
        return cls(_parse_ref(ref, path), rel, title, media_type)

@dataclass(frozen=True, slots=True)
class Location:
    """An RFC 9110 ``Location`` on a response — 201 Created, a redirect target, or the
    status URL on a 202. Build with a constructor; resolution happens at response send."""

    target: Target

    @classmethod
    def from_operation(
        cls, operation: Callable[..., object], *, path: Struct | None = None
    ) -> Self:
        """Point at a mounted operation; ``path`` (type-checked here) fills its slots."""
        _validate_operation_path(operation, path)
        return cls(OperationTarget(operation, path))

    @classmethod
    def from_url(cls, url: str) -> Self:
        """Point at a fully-qualified URL, used verbatim — never rewritten."""
        return cls(URLTarget(url))

    @classmethod
    def from_path(cls, path: str) -> Self:
        """Point at a root-relative path; it picks up the app's URL base (absolute origin /
        prefix) the same way a reversed operation does."""
        return cls(PathTarget(path))

    @classmethod
    def from_ref(cls, ref: str, *, path: Struct | None = None) -> Self:
        """Point at an operation by its class ``ref`` (``"name.operation"``) — the
        import-cycle hatch; prefer ``from_operation`` otherwise."""
        return cls(_parse_ref(ref, path))

@dataclass(kw_only=True, slots=True)
class NDJSONStreamingResponse[T: Struct, H: Struct | None = None](_StreamingResponse[T, H]):
    """A response streamed as newline-delimited JSON — one ``T`` Struct per line
    (``application/x-ndjson``)."""

class OperationMeta(Struct):
    """OpenAPI metadata for a single operation (``meta_get``, ``meta_create``, …).

    ``operation_id`` lives here, never on the class-level ``meta`` — operation ids must
    be unique, so they can't sensibly cascade to every operation.
    """

    tags: Sequence[str] = ()
    operation_id: str | None = None

@dataclass(frozen=True, slots=True, repr=False)
class RawHeaders:
    """Immutable, case-insensitive view of the request headers, preserving as-sent
    names and order.

    For forwarding the whole header bag upstream or for diagnostics — *not* for
    reading values you act on (model those in a typed ``headers`` Struct). Lookups
    are case-insensitive (``raw["X-Trace-Id"] == raw["x-traceid"]``); iteration,
    ``items`` and ``repr`` keep the casing as sent. Registers as
    ``Mapping[str, str]`` so it drops straight into ``niquests(headers=...)``; pass
    :meth:`multi_items` instead when repeated headers must survive.
    """

    _pairs: list[tuple[str, str]]  # decoded, original casing, in order

    def _unique(self) -> list[tuple[str, str]]:
        """First-seen pair for each name, compared case-insensitively (Mapping contract)."""
        seen: set[str] = set()
        out: list[tuple[str, str]] = []
        for name, value in self._pairs:
            lower = name.lower()
            if lower not in seen:
                seen.add(lower)
                out.append((name, value))
        return out

    def __getitem__(self, key: str) -> str:
        lower = key.lower()
        for name, value in self._pairs:
            if name.lower() == lower:
                return value
        raise KeyError(key)

    def get(self, key: str, default: str | None = None) -> str | None:
        """The first value for ``key`` (case-insensitive), or ``default`` if absent."""
        try:
            return self[key]
        except KeyError:
            return default

    def getlist(self, key: str) -> list[str]:
        """Every value sent under ``key`` (case-insensitive), in order."""
        lower = key.lower()
        return [value for name, value in self._pairs if name.lower() == lower]

    def __contains__(self, key: object) -> bool:
        if not isinstance(key, str):
            return False
        lower = key.lower()
        return any(name.lower() == lower for name, _ in self._pairs)

    def __iter__(self) -> Iterator[str]:
        return (name for name, _ in self._unique())

    def keys(self) -> list[str]:
        """Unique header names, first-seen casing."""
        return [name for name, _ in self._unique()]

    def values(self) -> list[str]:
        """The value of the first occurrence of each unique header name."""
        return [value for _, value in self._unique()]

    def items(self) -> list[tuple[str, str]]:
        """First-seen ``(name, value)`` pair per unique header name (Mapping contract)."""
        return self._unique()

    def multi_items(self) -> list[tuple[str, str]]:
        """Every header pair, repeats included — use for faithful forwarding."""
        return list(self._pairs)

    def __len__(self) -> int:
        return len(self._unique())

    def __repr__(self) -> str:
        return f"RawHeaders({self._pairs!r})"

class Resource(_Routable):
    """One REST resource: subclass and define any of the CRUD methods.

    ``read_one`` is the item route (its ``path`` may extend the mount with
    the item id); ``read_many`` is the collection (its path is exact).

    Optional OpenAPI metadata is declared at class definition: ``meta`` applies to every
    operation, ``meta_<op>`` to one (``meta_create``, ``meta_read_one``, …).
    """

    METHODS: ClassVar[dict[str, _Verb]] = {
        "create": _Verb("POST", 201, extends_path=True),
        "read_one": _Verb("GET", 200, extends_path=True),
        "read_many": _Verb("GET", 200, extends_path=False),
        "update": _Verb("PUT", 200, extends_path=True),
        "partial_update": _Verb("PATCH", 200, extends_path=True),
        "delete": _Verb("DELETE", 200, extends_path=True),
    }

    meta: ClassVar[ResourceMeta | None] = None
    meta_create: ClassVar[OperationMeta | None] = None
    meta_read_one: ClassVar[OperationMeta | None] = None
    meta_read_many: ClassVar[OperationMeta | None] = None
    meta_update: ClassVar[OperationMeta | None] = None
    meta_partial_update: ClassVar[OperationMeta | None] = None
    meta_delete: ClassVar[OperationMeta | None] = None

    def __init_subclass__(
        cls,
        *,
        path: str,
        ref: str | None = None,
        meta: ResourceMeta | None = None,
        meta_create: OperationMeta | None = None,
        meta_read_one: OperationMeta | None = None,
        meta_read_many: OperationMeta | None = None,
        meta_update: OperationMeta | None = None,
        meta_partial_update: OperationMeta | None = None,
        meta_delete: OperationMeta | None = None,
        **kwargs: object,
    ) -> None:
        # path / ref handling lives on _Routable
        super().__init_subclass__(path=path, ref=ref, **kwargs)
        _validate_meta(
            cls,
            meta,
            ResourceMeta,
            {
                "meta_create": meta_create,
                "meta_read_one": meta_read_one,
                "meta_read_many": meta_read_many,
                "meta_update": meta_update,
                "meta_partial_update": meta_partial_update,
                "meta_delete": meta_delete,
            },
        )
        cls.meta = meta
        cls.meta_create = meta_create
        cls.meta_read_one = meta_read_one
        cls.meta_read_many = meta_read_many
        cls.meta_update = meta_update
        cls.meta_partial_update = meta_partial_update
        cls.meta_delete = meta_delete

class ResourceMeta(Struct):
    """OpenAPI metadata shared by all of a ``Resource``'s operations."""

    tags: Sequence[str] = ()

@dataclass(kw_only=True, slots=True)
class SSEResponse[T: Struct | str = str, H: Struct | None = None](
    _StreamingResponse[T | ServerSentEvent[T], H]
):
    """A Server-Sent Events response (``text/event-stream``, GET-only). Yield a
    Struct/str (sent as ``data``) or a :class:`ServerSentEvent`. ``keepalive``, if
    set, emits a comment ping every N idle seconds."""

    keepalive: float | None = None

@dataclass(kw_only=True, slots=True)
class ServerSentEvent[T: Struct | str]:
    """One Server-Sent Event. Yield from an :class:`SSEResponse` stream to control
    the ``event`` / ``id`` / ``retry`` fields; ``data`` is a Struct (encoded as
    JSON) or a raw ``str``."""

    data: T
    event: str | None = None
    id: str | None = None
    retry: int | None = None

@dataclass(kw_only=True, slots=True)
class StreamingResponse[H: Struct | None = None](_StreamingResponse[bytes, H]):
    """A response streamed as raw ``bytes`` chunks (``application/octet-stream`` by
    default; override via ``raw_headers``)."""

class TestClient:
    """Synchronous in-process client. Prefer ``with TestClient(app) as c``."""

    __test__ = False  # stop pytest from collecting this as a test case

    def __init__(self, app: BaseApp[Any]) -> None:
        self._app = app
        self._loop_thread = _LoopThread()
        self._to_app: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
        self._from_app: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
        self._lifespan_task: asyncio.Task[None]
        try:
            self._submit(self._start_lifespan())
        except BaseException:
            self._loop_thread.close()
            raise

    def _submit[T](self, coro: Coroutine[Any, Any, T]) -> T:
        return self._loop_thread.submit(coro)

    @staticmethod
    def _part_content(value: str | bytes) -> bytes:
        return value if isinstance(value, bytes) else value.encode()

    @staticmethod
    def _disposition(name: str, filename: str | None = None) -> bytes:
        escaped_name = name.replace("\\", "\\\\").replace('"', '\\"')
        value = f'Content-Disposition: form-data; name="{escaped_name}"'
        if filename is not None:
            escaped_filename = filename.replace("\\", "\\\\").replace('"', '\\"')
            value += f'; filename="{escaped_filename}"'
        return value.encode()

    @staticmethod
    def _iter_data_values(value: _DataValues) -> Sequence[_DataValue]:
        return value if isinstance(value, list) else [value]

    @staticmethod
    def _iter_file_values(value: _FileValues) -> Sequence[_FileValue]:
        return value if isinstance(value, list) else [value]

    def _encode_multipart(
        self,
        data: dict[str, _DataValues] | None,
        files: dict[str, _FileValues] | None,
    ) -> tuple[bytes, str]:
        boundary = "jero-test-boundary"
        chunks: list[bytes] = []
        for name, value in (data or {}).items():
            for item in self._iter_data_values(value):
                chunks += [
                    f"--{boundary}\r\n".encode(),
                    self._disposition(name),
                    b"\r\n\r\n",
                    self._part_content(item),
                    b"\r\n",
                ]
        for name, value in (files or {}).items():
            for item in self._iter_file_values(value):
                filename, content = item[:2]
                content_type = item[2] if len(item) == 3 else None
                chunks += [
                    f"--{boundary}\r\n".encode(),
                    self._disposition(name, filename),
                    b"\r\n",
                ]
                if content_type is not None:
                    chunks += [f"Content-Type: {content_type}\r\n".encode()]
                chunks += [b"\r\n", content, b"\r\n"]
        chunks.append(f"--{boundary}--\r\n".encode())
        return b"".join(chunks), f"multipart/form-data; boundary={boundary}"

    async def _start_lifespan(self) -> None:
        self._lifespan_task = asyncio.create_task(
            self._app({"type": "lifespan"}, self._to_app.get, self._from_app.put)
        )
        await self._to_app.put({"type": "lifespan.startup"})
        message = await self._from_app.get()
        if message["type"] == "lifespan.startup.failed":
            # The app re-raises after reporting; retrieve it so asyncio
            # doesn't warn about an unretrieved task exception.
            with contextlib.suppress(Exception):
                await self._lifespan_task
            raise RuntimeError(f"lifespan startup failed: {message.get('message')}")

    async def _stop_lifespan(self) -> None:
        await self._to_app.put({"type": "lifespan.shutdown"})
        await self._from_app.get()
        await self._lifespan_task

    async def _request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, str] | None,
        json: Any,
        content: bytes | None,
        data: dict[str, _DataValues] | None,
        files: dict[str, _FileValues] | None,
        headers: dict[str, str] | None,
    ) -> TestResponse:
        body = b""
        wire_headers = {k.lower(): v for k, v in (headers or {}).items()}
        if json is not None:
            body = msgspec_encoder.encode(json)
            wire_headers.setdefault("content-type", "application/json")
        elif content is not None:
            body = content
            wire_headers.setdefault("content-type", "application/octet-stream")
        elif data is not None or files is not None:
            body, content_type = self._encode_multipart(data, files)
            wire_headers.setdefault("content-type", content_type)

        scope: dict[str, Any] = {
            "type": "http",
            "method": method,
            "path": path,
            "query_string": urlencode(params or {}).encode("latin-1"),
            "headers": [
                (k.encode("latin-1"), v.encode("latin-1")) for k, v in wire_headers.items()
            ],
        }

        cycle = _RequestCycle(body)
        await self._app(scope, cycle.receive, cycle.send)
        return TestResponse(
            status_code=cycle.status,
            headers=cycle.headers,
            content=b"".join(cycle.chunks),
            multi_headers=cycle.multi_headers,
        )

    async def _stream_request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, str] | None,
        json: Any,
        content: bytes | None,
        data: dict[str, _DataValues] | None,
        files: dict[str, _FileValues] | None,
        headers: dict[str, str] | None,
    ) -> _StreamSession:
        body = b""
        wire_headers = {k.lower(): v for k, v in (headers or {}).items()}
        if json is not None:
            body = msgspec_encoder.encode(json)
            wire_headers.setdefault("content-type", "application/json")
        elif content is not None:
            body = content
            wire_headers.setdefault("content-type", "application/octet-stream")
        elif data is not None or files is not None:
            body, content_type = self._encode_multipart(data, files)
            wire_headers.setdefault("content-type", content_type)

        scope: dict[str, Any] = {
            "type": "http",
            "method": method,
            "path": path,
            "query_string": urlencode(params or {}).encode("latin-1"),
            "headers": [
                (k.encode("latin-1"), v.encode("latin-1")) for k, v in wire_headers.items()
            ],
        }
        cycle = _StreamCycle(body)
        task = asyncio.create_task(self._app(scope, cycle.receive, cycle.send))
        while True:
            message = await asyncio.to_thread(cycle.chunks.get)
            if message["type"] == "http.response.start":
                headers = {k.decode("latin-1"): v.decode("latin-1") for k, v in message["headers"]}
                return _StreamSession(self._submit, cycle, task, message["status"], headers)
            cycle.chunks.put(message)

    def request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, str] | None = None,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue a request and return the buffered response."""
        return self._submit(
            self._request(
                method.upper(),
                path,
                params=params,
                json=json,
                content=content,
                data=data,
                files=files,
                headers=headers,
            )
        )

    def stream_request(
        self,
        method: str,
        path: str,
        *,
        params: dict[str, str] | None = None,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        headers: dict[str, str] | None = None,
    ) -> _StreamSession:
        """Issue a request and return a streaming session for its chunks."""
        return self._submit(
            self._stream_request(
                method.upper(),
                path,
                params=params,
                json=json,
                content=content,
                data=data,
                files=files,
                headers=headers,
            )
        )

    def get(
        self,
        path: str,
        *,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue a GET request."""
        return self.request("GET", path, params=params, headers=headers)

    def stream_get(
        self,
        path: str,
        *,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> _StreamSession:
        """Open a streaming GET request."""
        return self.stream_request("GET", path, params=params, headers=headers)

    def head(
        self,
        path: str,
        *,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue a HEAD request."""
        return self.request("HEAD", path, params=params, headers=headers)

    def options(
        self,
        path: str,
        *,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue an OPTIONS request."""
        return self.request("OPTIONS", path, params=params, headers=headers)

    def delete(
        self,
        path: str,
        *,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue a DELETE request."""
        return self.request("DELETE", path, params=params, headers=headers)

    def stream_delete(
        self,
        path: str,
        *,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> _StreamSession:
        """Open a streaming DELETE request."""
        return self.stream_request("DELETE", path, params=params, headers=headers)

    def post(
        self,
        path: str,
        *,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue a POST request (JSON, raw bytes, or multipart form)."""
        return self.request(
            "POST",
            path,
            json=json,
            content=content,
            data=data,
            files=files,
            params=params,
            headers=headers,
        )

    def stream_post(
        self,
        path: str,
        *,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> _StreamSession:
        """Open a streaming POST request."""
        return self.stream_request(
            "POST",
            path,
            json=json,
            content=content,
            data=data,
            files=files,
            params=params,
            headers=headers,
        )

    def put(
        self,
        path: str,
        *,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue a PUT request (JSON, raw bytes, or multipart form)."""
        return self.request(
            "PUT",
            path,
            json=json,
            content=content,
            data=data,
            files=files,
            params=params,
            headers=headers,
        )

    def stream_put(
        self,
        path: str,
        *,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> _StreamSession:
        """Open a streaming PUT request."""
        return self.stream_request(
            "PUT",
            path,
            json=json,
            content=content,
            data=data,
            files=files,
            params=params,
            headers=headers,
        )

    def patch(
        self,
        path: str,
        *,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> TestResponse:
        """Issue a PATCH request (JSON, raw bytes, or multipart form)."""
        return self.request(
            "PATCH",
            path,
            json=json,
            content=content,
            data=data,
            files=files,
            params=params,
            headers=headers,
        )

    def stream_patch(
        self,
        path: str,
        *,
        json: Any = None,
        content: bytes | None = None,
        data: dict[str, _DataValues] | None = None,
        files: dict[str, _FileValues] | None = None,
        params: dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
    ) -> _StreamSession:
        """Open a streaming PATCH request."""
        return self.stream_request(
            "PATCH",
            path,
            json=json,
            content=content,
            data=data,
            files=files,
            params=params,
            headers=headers,
        )

    def close(self) -> None:
        """Run the app's lifespan shutdown and stop the background loop."""
        self._submit(self._stop_lifespan())
        self._loop_thread.close()

    def __enter__(self) -> Self:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()