Greyhorse Web library

Web transport for greyhorse applications: HTTP, GraphQL and gRPC handlers written the way each of those libraries is normally written, assembled into a running server by the application itself.

The library does not replace FastAPI, Strawberry or grpclib — it wires them. A handler is an ordinary class whose collaborators arrive through __init__, so it can be built by hand in a unit test with no framework loaded and no container to reach into.

@http_handler
class UserApi:
    def __init__(self, service: UserService) -> None:
        self._service = service

    @post('/', status_code=201)
    async def create_user(self, body: CreateUserReq) -> UserResp:
        return UserResp(username=self._service.create_user(body.username))

    @get('/')
    async def list_users(self, limit: Annotated[int, Query(le=100)] = 20) -> list[UserResp]:
        return [UserResp(username=name) for name in self._service.list_users(limit)]

Those are FastAPI's own Query, its own body model, its own status_code. Nothing is re-invented; @get and @post only record the verb and the path on the method, and the router is built later, from the instance the application constructed.

Declaring that the class has routes is one line in the component, and mounting is the application's job:

class UsersComponent(Component):
    fragments: ClassVar = UsersFragment
    exports: ClassVar = UserApi
    handlers: ClassVar = (Routes(UserApi),)


app = Application(
    UsersApp,
    gateways=(UvicornHttpGateway(host='127.0.0.1', port=8000, title='users'),),
)
app.run_sync()

The component's name becomes the path prefix. UserApi never said where it lives.

Install

Two independent choices — the HTTP layer and the ASGI server — combined by you:

pip install 'greyhorse-web[fastapi,uvicorn]'

Extra	Brings
fastapi	the HTTP layer
uvicorn, hypercorn, granian	the ASGI server (pick one)
strawberry	GraphQL
betterproto	gRPC, over grpclib

All three servers are interchangeable: same gateway API, same access log, same graceful drain. Requires Python 3.14.

Usage

Every program below was run against the library as it stands, and the answers quoted under each one are what it actually returned. The longer, commented versions live in examples/ and are executed by the test suite, so they cannot rot silently.

HTTP

The handler is an ordinary class with FastAPI decorators on its methods. Routes(UserApi) in the component declares that it has routes; the router itself is built later, from the instance the application constructed.

from typing import ClassVar

from pydantic import BaseModel

from greyhorse.rock import Fragment, factory
from greyhorse.run import wrap_sync
from greyhorse.strand import Application, Component, Module, Use
from greyhorse_web.http.transport.binding import Routes
from greyhorse_web.http.transport.routing import get, http_handler, post
from greyhorse_web.servers.uvicorn import UvicornHttpGateway


class UserResp(BaseModel):
    username: str


class UserService:
    def __init__(self) -> None:
        self._names: list[str] = []

    def create(self, username: str) -> str:
        self._names.append(username)
        return username

    def list_users(self) -> list[str]:
        return list(self._names)


@http_handler
class UserApi:
    def __init__(self, service: UserService) -> None:
        self._service = service

    @post('/', status_code=201)
    async def create_user(self, body: UserResp) -> UserResp:
        return UserResp(username=self._service.create(body.username))

    @get('/')
    async def list_users(self) -> list[UserResp]:
        return [UserResp(username=name) for name in self._service.list_users()]


class UsersFragment(Fragment):
    exports: ClassVar = {UserApi}

    user_service = factory(UserService)
    user_api = factory(UserApi)


class UsersComponent(Component):
    fragments: ClassVar = UsersFragment
    exports: ClassVar = UserApi
    handlers: ClassVar = (Routes(UserApi),)


class UsersApp(Module):
    name = 'users-app'
    components: ClassVar = {'users': Use(UsersComponent)}


def main() -> None:
    Application(
        UsersApp,
        gateways=(UvicornHttpGateway(host='127.0.0.1', port=8000, title='users'),),
    ).run_sync()


wrap_sync(main)

$ curl -X POST localhost:8000/users -H 'content-type: application/json' \
       -d '{"username": "alice"}'
{"username":"alice"}
$ curl localhost:8000/users
[{"username":"alice"}]

The routes are under /users because the component is mounted there. UserApi never said where it lives, and moving it under another name moves its routes with it.

GraphQL

Same shape, strawberry instead of FastAPI. Schema(UserApi, name='users') gives the class one root field; where the component is mounted does not touch it, because a root field name is published schema and a mount path is not.

from typing import ClassVar

import strawberry

from greyhorse.rock import Fragment, factory
from greyhorse.run import wrap_sync
from greyhorse.strand import Application, Component, Module, Use
from greyhorse_web.http.transport.binding import Schema
from greyhorse_web.http.transport.graphql import graphql_handler, mutation, query
from greyhorse_web.servers.uvicorn import UvicornHttpGateway


@strawberry.type
class User:
    username: str


class UserService:
    def __init__(self) -> None:
        self._names: list[str] = []

    def create(self, username: str) -> str:
        self._names.append(username)
        return username

    def list_users(self) -> list[str]:
        return list(self._names)


@graphql_handler
class UserApi:
    def __init__(self, service: UserService) -> None:
        self._service = service

    @query(name='list')
    async def list_users(self) -> list[User]:
        # Named `list` in the schema but not in Python: a method called
        # `list` would shadow the builtin inside the class body, and under
        # PEP 649 the annotation `list[User]` is evaluated in exactly that
        # namespace.
        return [User(username=name) for name in self._service.list_users()]

    @mutation(description='Add a user and return it.')
    async def create(self, username: str) -> User:
        return User(username=self._service.create(username))


class UsersFragment(Fragment):
    exports: ClassVar = {UserApi}

    user_service = factory(UserService)
    user_api = factory(UserApi)


class UsersComponent(Component):
    fragments: ClassVar = UsersFragment
    exports: ClassVar = UserApi
    handlers: ClassVar = (Schema(UserApi, name='users'),)


class UsersApp(Module):
    name = 'users-app'
    components: ClassVar = {'users': Use(UsersComponent)}


def main() -> None:
    Application(
        UsersApp,
        gateways=(UvicornHttpGateway(host='127.0.0.1', port=8000, title='users'),),
    ).run_sync()


wrap_sync(main)

$ curl localhost:8000/graphql -H 'content-type: application/json' \
       -d '{"query": "mutation { users { create(username: \"bob\") { username } } }"}'
{"data":{"users":{"create":{"username":"bob"}}}}
$ curl localhost:8000/graphql -H 'content-type: application/json' \
       -d '{"query": "{ users { list { username } } }"}'
{"data":{"users":{"list":[{"username":"bob"}]}}}

A browser at the same address gets the GraphiQL explorer. HTTP handlers and GraphQL handlers live side by side in one component and one application.

A failing resolver, and where its text goes. Strawberry puts str(exc) into errors[].message, so an unhandled failure travels to the caller verbatim — a DSN with a password in it, if that is what the exception said. One extension stops that: graphql={'extensions': [MaskErrors]}, and the client then sees Unexpected error.

The original does not disappear, and this is worth knowing in both directions. Strawberry logs it BEFORE masking, with a traceback, on the stdlib logger strawberry.execution. So a service that does not route that logger into its own logging loses its last diagnostic point silently — and one that ships logs somewhere shared should know the unmasked text, DSN and all, is in them.

gRPC

The handler is a subclass of the *Base the betterproto2 compiler wrote, overriding the methods named in the .proto. There are no per-method decorators on purpose: the wire paths are in the .proto, and marking them again in Python would be a second source of truth free to disagree.

from typing import ClassVar

from greetpb.demo import GreeterBase, HelloReq, HelloResp

from greyhorse.rock import Fragment, factory
from greyhorse.run import wrap_sync
from greyhorse.strand import Application, Component, Module, Use
from greyhorse_web.grpc.binding import Rpc
from greyhorse_web.grpc.gateway import GrpcGateway
from greyhorse_web.grpc.service import grpc_handler


class VisitorBook:
    def __init__(self) -> None:
        self.names: list[str] = []


@grpc_handler
class Greeter(GreeterBase):
    def __init__(self, book: VisitorBook) -> None:
        self._book = book

    async def say_hello(self, message: HelloReq) -> HelloResp:
        self._book.names.append(message.name)
        return HelloResp(message=f'hello {message.name}')


class GreetFragment(Fragment):
    exports: ClassVar = {Greeter}

    visitor_book = factory(VisitorBook)
    greeter = factory(Greeter)


class GreetComponent(Component):
    fragments: ClassVar = GreetFragment
    exports: ClassVar = Greeter
    handlers: ClassVar = (Rpc(Greeter),)


class GreetApp(Module):
    name = 'greet-app'
    components: ClassVar = {'greet': Use(GreetComponent)}


def main() -> None:
    Application(
        GreetApp, gateways=(GrpcGateway(host='127.0.0.1', port=50051),)
    ).run_sync()


wrap_sync(main)

The component name gives the service no prefix — a gRPC address lives in the .proto, so this one answers at /demo.Greeter/SayHello wherever the component is mounted.

The client half is two helpers: a channel from GrpcCreds, a stub from the channel. channel_factory is a generator, shaped for a dependency graph — a fragment declares it as a factory and the channel is closed when that window closes.

Driven by hand instead, it has one rule: build the channel on the event loop that will use it. Channel.__init__ binds to the current loop, so from a thread without one it raises RuntimeError: There is no current event loop. Hence the call inside the coroutine, and closing() rather than a second next() — an abandoned generator still runs its finally and releases the socket.

import asyncio
from contextlib import closing

from greetpb.demo import GreeterStub, HelloReq

from greyhorse_web.grpc.client import channel_factory, client_factory
from greyhorse_web.schemas import GrpcCreds


async def main() -> None:
    creds = GrpcCreds(host='127.0.0.1', port=50051, tls_verification='disabled')
    with closing(channel_factory(creds)) as channels:
        greeter = client_factory(GreeterStub, next(channels), timeout=5.0)
        print((await greeter.say_hello(HelloReq(name='alice'))).message)


asyncio.run(main())

hello alice

Choosing a server

One import and one class name apart. Everything else — the handlers, the components, the application — is untouched.

from greyhorse_web.servers.uvicorn import UvicornHttpGateway
from greyhorse_web.servers.hypercorn import HypercornHttpGateway
from greyhorse_web.servers.granian import GranianHttpGateway

Each takes the same arguments and behaves the same way: same access log, same admission gate, same graceful drain. config={...} passes settings straight through to that server's own configuration object for anything this package does not have an argument for. One caveat, measured rather than assumed: granian's embedded server starts refusing new TCP connections at roughly 300+ concurrent — a capacity limit of that mode, which granian labels experimental, not a correctness problem.

Configuration

The HTTP gateways, whichever server:

Argument	Default	Means
host, port	'127.0.0.1', 8000	where to listen
title, version	'greyhorse', '0.1.0'	OpenAPI identity
debug	False	indented JSON, and the exception's own text in detail
root_path	''	mount path behind a proxy
cors	None	allowed origins. None means every origin; a NAMED list also enables credentials; [] means nobody
trusted_proxies	None	peers whose X-Forwarded-* are believed — addresses, CIDR networks, or '*'. Nothing is trusted by default, so a deployment behind an ingress must say so or every request is attributed to the proxy
docs	True	publish /docs, /redoc and /openapi.json
access_log	True	a line per request on greyhorse_web.http.access
request_id	True	settle a correlation id for every call — see below
graphql_path	'/graphql'	where a schema is served, if the application has one
graphql	{}	settings for the GraphQL layer, split by name: extensions, config, scalar_map, types, directives shape the schema, everything else (context_getter, graphql_ide, allow_queries_via_get, …) the router
middleware	()	middleware the application declares, OUTERMOST FIRST, installed inside everything the gateway owns
lifespan	None	the plain FastAPI hook, wrapped in the gateway's own frame rather than replacing it
auth_backend	None	who the caller is — see below
startup_timeout	10.0	how long start() waits for the port to answer
drain_timeout	30.0	how long stop() waits for in-flight requests
config	{}	passed through to the server's own config

GrpcGateway takes host, port (50051), access_log, request_id, startup_timeout, drain_timeout with the same meanings, plus trust_inbound_request_id (False, see below), config and listen — grpclib splits its settings between Server(...) and Server.start(...), and folding them into one bag would hide which is which.

A shutdown budget should be read as 2 * drain_timeout: a threaded server can wait that long once in the drain and again joining the server thread. The two are bounded separately because they fail for different reasons.

Correlation id

One id per call, settled at the edge and carried through everything that follows: the access log line, whatever the application logs itself, the response, and the next service this one calls. Without it, tying those together means guessing from timestamps.

It is on by default on both transports. Nothing to import to get the ordinary benefit — the access logs carry request_id as a field, beside method, path and status. Application code that wants the value asks for it:

from greyhorse_web.request_id import current_request_id

logger.info('charging the card', extra={'request_id': current_request_id()})

None outside a call, never an empty string, so a blank field never has to mean two different things.

One nuance worth stating rather than discovering: a task started inside a call inherits the id and KEEPS it, including after the response has gone out. That is ContextVar semantics — a task copies the context it was created in — and it is what makes the id reach work a handler spawns. The other face of it is that a background task left running past the request still logs under that request's id.

Where the id comes from, and who is believed. On HTTP the inbound x-request-id header is reused — but only from a peer the gateway already trusts through trusted_proxies. An id chosen by an untrusted caller is one they can reuse across requests on purpose, collapsing two unrelated traces into one in every log downstream; and since the value is echoed back and forwarded onward, it is untrusted input in the fullest sense. Anything else gets a freshly minted id instead: no header, an untrusted peer, a value carrying a control character, a value over 128 characters, or one that is not ASCII. The last of those is about the hop, not about the character: HTTP headers are latin-1 and would carry é happily, while gRPC metadata is ASCII-only and grpclib refuses it outright — so an id accepted at the HTTP edge and forwarded onward would not merely fail to correlate, it would break the downstream call. The narrower of the two alphabets decides. Every id a proxy generates is ASCII anyway: Envoy's is a UUID, nginx's $request_id is 16 random bytes in hex.

The middleware sits OUTSIDE everything else the gateway installs, which is what makes a request refused during a drain still carry one: the 503 has an x-request-id header and an access log line to match. It also rewrites the request's own headers, so a middleware or handler INSIDE it reads the settled id rather than minting a second, disagreeing one — and cannot read what an untrusted caller wrote, because that value is gone by then.

Two consequences worth knowing:

• x-request-id is the library's now. An application middleware of your own that stamps a header should pick another name, or it will find its value replaced on the way out.
• Behind a proxy over a unix socket there is no peer address to judge (scope['client'] is None), so a CIDR list can never match and every inbound id is discarded. Such a deployment wants trusted_proxies='*', which is the right answer there anyway: the socket is reachable by the proxy alone.

gRPC does the same with call metadata, with one difference: trust_inbound_request_id is False by default, so an id arriving from a caller is ignored and a fresh one minted. The HTTP gateway can judge the sender because it already keeps a trust set for X-Forwarded-*; a gRPC port has no such notion here, so this is a plain switch to flip when that port is reachable only from inside your own network.

Crossing the boundary. client_factory attaches the current id to every outgoing call, resolved when the call is made rather than when the stub was built. That distinction is the whole feature: a stub built once and shared by a dependency graph would otherwise stamp the FIRST call's id onto every later one, which is worse than no correlation at all because it is confidently wrong.

What client_factory returns is a subclass of the stub class you asked for, with betterproto2's four call helpers wrapped. That is not a detail one can wave away: a generated method's own metadata= REPLACES the stub's stored default rather than merging with it, so an id kept on the stub would be dropped by every call that passed metadata of its own — stub.say_hello(req, metadata={'authorization': ...}) being an entirely ordinary line. Wrapping the call covers the stub's default and the call's own argument alike, keeps repeated metadata keys repeated (gRPC metadata is multi-valued), reads a generator of pairs exactly once, and leaves a caller's own x-request-id alone. isinstance(stub, GreeterStub) still holds and stub.channel is still the channel you passed.

Two calls through one stub, under two ids, as the server saw them:

HANDLER call-one -> first-call-id
/demo.Greeter/SayHello OK 10.6ms request_id=first-call-id
HANDLER call-two -> second-call-id
/demo.Greeter/SayHello OK  0.2ms request_id=second-call-id

Outgoing HTTP calls are not covered, because this package ships no HTTP client to cover them. A service calling another over HTTP forwards the id itself — current_request_id() into an x-request-id header on whatever client it uses.

The paginator's limits, stated rather than discovered: there is no first/last split, and a cursor carries an identity, so sorting the query yourself and paginating it with these cursors are not compatible — the rows come out in your order while the cursor still walks identities, so page two can overlap page one. AsyncPaginator's docstring has the rest.

With structlog, the fields arrive as fields: both access logs put them in extra=, so structlog.stdlib.ExtraAdder in the processor chain lifts request_id into the event dict with no parsing.

Switch it off per transport with request_id=False — the id is then None everywhere, and no header is added.

What it does

Assembles routes. Handlers declare verbs and paths; the gateway collects them from the whole application, rejects two handlers that would answer the same request, and mounts them under their components' prefixes.

Speaks JSON fast. orjson on both directions — requests and responses, error paths included — wired in through FastAPI's own route class, with no global patching.

Logs uniformly. One line per request on greyhorse_web.http.access (greyhorse_web.grpc.access for gRPC), with the method, path, status and duration as structured fields. No server installs handlers or runs dictConfig on the host process: uvicorn's and granian's process-wide defaults are declined, so output goes wherever the application says — including through structlog.

One exception, deliberate and narrow: while a uvicorn gateway is running it keeps the uvicorn.access logger empty, because that is the only thing uvicorn reads when deciding whether to write a per-request line of its own (access_logger.hasHandlers(), not the config flag). It is snapshotted at startup and restored on shutdown. A second uvicorn app embedded in the same process loses its access log for that window.

Correlates. Every call gets an id — reused from the edge when the peer is trusted, minted when it is not — bound for the length of the call, written as a field on both access logs, echoed to the client, and attached to outgoing gRPC calls. Off in one word per transport. See Correlation id.

Drains gracefully. On shutdown the gateway stops admitting new work and waits for what is in flight; arrivals in that window get 503 (HTTP) or UNAVAILABLE (gRPC) rather than a dropped connection. Both answers carry the correlation id, which is the window where it matters most.

Serves GraphQL and gRPC the same way. @query/@mutation on a class for Strawberry, a betterproto2 service for gRPC — both collected and mounted by the same mechanism as HTTP.

What it does not do

Authentication and authorization — no policy of its own. The HTTP layer installs starlette's AuthenticationMiddleware, so request.user and request.auth always exist, and it decides nothing: without auth_backend= everyone is anonymous. An application with real authentication passes its own backend, and a GraphQL resolver reads the same decision through graphql={'context_getter': ...} — the two seams that make this transport usable by a service that authenticates. What is still absent is any authorization: nothing here decides what an authenticated caller may do, and the gRPC side has no seam for either. Anything reachable with this library needs its own check in front of it.

Proxy headers are not trusted by default. Behind a reverse proxy, say so explicitly or every request is attributed to the proxy's address:

UvicornHttpGateway(..., trusted_proxies='*')             # only ever behind a proxy
UvicornHttpGateway(..., trusted_proxies=['10.0.0.0/8'])  # or name the networks

The header anyone can set must not decide the address a service rate-limits and audit-logs by.

The API map is published by default. /docs, /redoc and /openapi.json are FastAPI's own and stay on, but /openapi.json is every path, parameter and response shape — and there is no authentication in front of it. On an untrusted network, say so:

UvicornHttpGateway(..., docs=False)

That also frees those paths for a handler of your own to claim.

No WebSocket support, no DataLoader for GraphQL N+1, and no defined transaction boundary across a resolver tree. TLS is passed through to the server but is not covered by tests.

Behaviour under load has been measured, though not by the test suite: the admission gate, the access log, proxy-header settlement, GraphQL execution and resource borrowing were each driven with several hundred concurrent requests -- including stop() racing in-flight work from another thread -- with no cross-talk, no lost or duplicated records, and the in-flight counter returning to zero every time. One caveat came out of it: granian's embedded server starts refusing new TCP connections at roughly 300+ concurrent, which is a capacity limit of that mode rather than a correctness problem -- everything that got through was correct. granian labels embedded mode experimental and so does this package.

gRPC streaming methods ARE covered: the gateway wraps a handler without touching its cardinality, and there are tests for a streaming call reaching its handler, keeping UNARY_STREAM, being counted in flight for its whole life rather than its first message, and being refused by a closed gate.

Examples

Runnable, in examples/:

File	Shows
01_http.py	resource, service layer, handler, listening server
02_graphql.py	GraphQL schema assembled from components
03_grpc.py	a betterproto2 service over grpclib
04_seams.py	middleware=, lifespan= and graphql= declared, with no gateway subclass
05_request_id.py	one correlation id across an HTTP hop into gRPC

Each is exercised by tests/test_examples.py as a real process, so none of them can rot quietly.

Development

uv venv && uv sync --all-extras
.venv/bin/python -m pytest
.venv/bin/ruff check && .venv/bin/ruff format
.venv/bin/mypy greyhorse_web

--all-extras is not optional here. Everything this package talks to -- FastAPI, the three ASGI servers, strawberry, betterproto2 -- lives in an extra, and a plain uv sync installs none of them: the environment builds without complaint, and then everything under http/, grpc/ and servers/ fails to import, so the test suite cannot even be collected.

License

MIT — see LICENSE.