sillo-graphql

Production GraphQL for Sillo. Installs as sillo-graphql, imports as sillo_graphql.

Strawberry owns the schema. This package owns everything around it — the transports, the safety, and the observability.

pip install sillo-graphql

import strawberry
from sillo import Depend, HttpContext, SilloApp
from sillo_graphql import Graph, Limits, field

@strawberry.type
class Query:
    @field
    async def me(ctx: HttpContext, db=Depend(get_db)) -> User:
        return await db.users.get(ctx.user.id)

app = SilloApp()
Graph(strawberry.Schema(query=Query), limits=Limits(depth=8)).mount(app)

Resolvers that read like handlers

A sillo route handler takes the context first and declares what else it needs. So does a resolver here:

@field
async def posts(ctx: HttpContext, db=Depend(get_db), limit: int = 10) -> list[Post]:
    return await db.posts.recent(limit)

One rule: ctx and anything defaulted to Depend are injected and never appear in the schema; every other parameter is a GraphQL argument. So this field takes exactly one argument, limit.

Dependencies are resolved by the framework's own solver, with the framework's own pre-flattened execution plan — two resolvers in one operation that both ask for Depend(get_db) are handed the same session.

Configuration

graph = Graph(
    schema,
    path="/graphql",
    ide=False,                                       # explorer, off by default
    introspection=False,                             # off by default
    subscriptions=True,
    auth=Bearer(),                                   # the route's auth= gate
    limits=Limits(depth=10, cost=1_000, aliases=15),
    errors=ErrorPolicy(mask=True),
    transport=Transport(get_queries=True, batch=10),
    uploads=Uploads(enabled=True, max_size="10MB"),
    persisted=Persisted(apq=True, trusted="operations.json"),
)
graph.mount(app)

Common knobs are keyword arguments; the deeper ones are policy objects, the same split the framework makes between arguments on SilloApp and objects like CSRFConfig.

Every default is chosen for a public endpoint.

What it does

Cost limits, enforced before execution

Depth, aliases, breadth and document size, plus a weighted cost that understands lists — a field returning a list multiplies everything under it, by the page size the caller asked for when that is knowable.

@field(cost=25)
async def search(ctx: HttpContext, term: str) -> list[Hit]: ...

An operation over budget is refused with OPERATION_TOO_COMPLEX and the limit it passed, before a single resolver runs. Refusing afterwards would mean having already done the work.

Errors that say what happened, and no more

from sillo_graphql import forbidden, not_found

@field
async def post(ctx: HttpContext, id: int) -> Post:
    found = await Post.objects.get_or_none(id=id)
    if found is None:
        raise not_found("No such post")     # extensions.code == "NOT_FOUND"
    return found

Free builders, like the framework's json() and text(). Errors raised this way are deliberate and reach the client. An exception that escapes a resolver is masked, logged with its traceback, and reported as INTERNAL_SERVER_ERROR — because what it said may name a host, a table or a credential.

Map your own:

@graph.on_error(RecordNotFound)
def _(exc): return not_found(str(exc))

Batching, so a graph query is not a table scan per node

@graph.loader
async def load_author(keys: list[int]) -> list[User]:
    rows = await User.objects.filter(id__in=keys).all()
    return align(rows, keys)

@field
async def author(ctx: HttpContext, root: Post) -> User:
    return await load_author(root.author_id)

Keys asked for by sibling fields in the same tick become one call. State is per operation, so two concurrent requests never share a cache.

Subscriptions that exist

graphql-transport-ws over the framework's own WebSocket layer, with an initialisation timeout, ping/pong keepalive, and cancellation in a finally so an operation cannot outlive its socket. Authentication belongs in connection_init, because a browser cannot set headers on a WebSocket handshake:

@graph.on_connect
async def authenticate(socket, params):
    token = params.get("authorization")
    if not token:
        raise unauthenticated("A token is required")
    return {"user": await user_for(token)}

The same subscriptions are available over text/event-stream with sse=True, for clients that cannot hold a socket open.

The rest of the HTTP surface

Batched operations (capped, sequential), GET for queries with mutations refused, application/graphql bodies, file uploads per the multipart request spec, and content negotiation between application/graphql-response+json — the spec's status codes — and legacy application/json, which stays always-200 for the clients that expect it.

Persisted operations

APQ saves bandwidth. A trusted-document manifest is the one that matters: with Persisted(trusted="operations.json") the endpoint executes nothing else, so the workload becomes finite and known.

Knowing what it is doing

graph.on_operation(OperationLog(slower_than=0.5))

metrics = Metrics()
graph.on_operation(metrics)

Per operation, not per path: p99 on POST /graphql averages over work that has nothing in common.

Testing

from sillo_graphql.testing import GraphClient

def test_me():
    with GraphClient(app) as gql:
        result = gql.query("{ me { email } }")
        assert result.ok
        assert result["me"]["email"] == "a@b.c"

async def test_prices():
    async with GraphClient(app).subscribe(PRICES, symbol="ACME") as stream:
        assert (await stream.next())["prices"]["last"] == 10

Migrating from sillo.graphql in the framework

info.context["ctx"] still works — the context is a Mapping — so a schema can migrate one resolver at a time.

Requirements

Python 3.10+, sillo-framework 1.0 or newer, strawberry-graphql.

1.0 is the floor because the resolver bridge is built on the context-handler API. Versions before 1.0 also shipped a sillo.graphql of their own, which is unrelated to this package and is what the table above migrates from.

License

BSD-3-Clause.