Spartan GraphQL

Spartan, adj. … distinguished by simplicity, frugality, courage, or brevity of speech.

— Oxford English Dictionary

Spartan GraphQL compiles GraphQL documents into a typed Python package, and stops there:

• no imposed transport: bytes in, bytes out;
• no runtime dependencies¹: nothing to install, nothing to upgrade;
• no redundant validation: just dicts, statically checked².

Built-in:

• query, mutation, and subscription with their variables and selected fields typed;
• union, interface, enum, and @oneOf input typed as Python unions;
• fragments typed on their own;
• custom scalars converted to and from Python types;
• partial responses raised as errors, or returned with their data typed.

Opt-in:

• merging operations into a single transport request;
• non-null assertions on fields the schema declares nullable;
• structs typing a JSON scalar as an input;
• injectors supplying a value from your transport rather than at every call;
• colocation of each Python module with its GraphQL document.

Paste-in:

• tracing with OpenTelemetry;
• batching of concurrent queries;
• file uploads with the GraphQL multipart request spec;
• runtime validation of responses with Pydantic;
• fuzzing of your code with Hypothesis.

Installation

pip install spartan-graphql

Or, with uv:

uv add --dev spartan-graphql

The generator's only dependency is graphql-core¹.

Quick start

Every example of this README is a file of bookshop, whose schema is schema.graphqls:

type Query {
  # …
  "The order with this identifier."
  order(id: ID!): Order
  # …
}

enum OrderStatus {
  PENDING
  SHIPPED
  DELIVERED
  CANCELED
}

type Order {
  id: ID!
  status: OrderStatus!
  # …
}

Write your operations in GraphQL document files:

query GetOrder($id: ID!) {
  order(id: $id) {
    id
    status
  }
}

The generator reads graphql-config, the same file GraphQL editor extensions and linters already use³:

# An SDL file here, but globs, introspection results, and server URLs work too.
schema: schema.graphqls
documents: "*.graphql"
extensions:
  spartan:
    # Where imports start, relative to this file's directory.
    moduleRoot: ..
    # Dotted name from the module root, so written to `bookshop/graphql`.
    package: bookshop.graphql

Generate the package⁴:

spartan-graphql bookshop/graphql.config.yml

The generated package:

• depends on nothing but the standard library¹;
• holds only the enum and input types your operations reach, so it grows with your documents rather than with the schema;
• puts each document's operations in a module of its own.

Your type checker verifies every variable and every field of the response. The example below shows what the type checker can prove with assert_type() and assert_never():

from typing import Literal, assert_never, assert_type

from bookshop.get_order_graphql import GetOrder
from bookshop.graphql.schema import OrderStatus
from bookshop.transport import send

data = send(GetOrder({"id": "o1"}))
order = data["order"]

# `Query.order`'s type is nullable, so the type checker requires this check.
if order is None:
    print("No such order.")
else:
    # `Order.id: ID!` is a `str` on the wire.
    assert_type(order["id"], str)

    # An `enum` gets a generated alias of the `Literal` of its values.
    assert_type(order["status"], OrderStatus)
    assert_type(order["status"], Literal["PENDING", "SHIPPED", "DELIVERED", "CANCELED"])

    if "total" in order:
        # A field not selected in the GraphQL operation can never be there.
        assert_never(order)

    print(f"Order {order['id']} is {order['status']}.")

Types

Operation types

Each operation gets a type for its variables and one for its data, both keyed by the names the GraphQL document uses. These types are for your type checker: a response stays as decoded from JSON, and only custom scalars with a codec are converted.

Missing keys

An optional variable or input field, or a selected field with @include or @skip, becomes a NotRequired key:

query ListTitles(
  $genres: [Genre!]
  $first: Int
  $withGenres: Boolean! = false
) {
  books(genres: $genres, first: $first) {
    ...TitleCard
  }
}

fragment TitleCard on Book {
  title
  genre @include(if: $withGenres)
}

def titles(
    *genres: Genre, first: int | None = None, with_genres: bool | None = None
) -> list[str]:
    # `$first` is nullable and `$withGenres` has a default, so both keys are `NotRequired`.
    variables: ListTitlesVariables = {
        # A GraphQL list variable takes any `Sequence`, such as this `tuple`.
        "genres": genres,
    }

    # A key left out sends nothing, while `first: None` would send `null`.
    if first is not None:
        variables["first"] = first

    if with_genres is not None:
        variables["withGenres"] = with_genres

    data = send(ListTitles(variables))
    # A GraphQL list field in data is a Python `list`, as JSON decodes it.
    assert_type(data["books"], list[TitleCard])
    labels: list[str] = []

    for book in data["books"]:
        # Selected with `@include`, `genre` is a `NotRequired` key too.
        if "genre" in book:
            assert_type(book["genre"], Genre)
            labels.append(f"{book['title']} ({book['genre'].lower()})")
        else:
            labels.append(book["title"])

    return labels

Unions

Wherever GraphQL lets a value be one of several things, it becomes a Python union type:

• a selection on a union or an interface is one of several types, and becomes one type per concrete type, told apart by __typename (which the generator selects for you);
• an enum is one of several values, and becomes a Literal;
• a @oneOf input is one of several fields, and becomes a union of single-key types, so a value with two keys fails type checking.

These unions are closed for the type checker, but nothing enforces them at runtime, since responses are not validated.

A server may add a type to a union, an implementation to an interface, or a member to an enum at any time, since none of these is a breaking change. A validating client would reject the new value before your code ever sees it. Here, your code receives it as sent and decides what to do with it.

One way is to accept the unknown value in a case _: arm:

def status_label(status: OrderStatus, /) -> str:
    match status:
        case "PENDING":
            return "Being prepared"
        case "SHIPPED":
            return "On its way"
        case "DELIVERED":
            return "Delivered"
        case "CANCELED":
            return "Canceled"
        case _:
            # A member added after codegen.
            return status.replace("_", " ").capitalize()

The other way is to reject it with case _ as never: assert_never(never), an arm that:

• raises an AssertionError at runtime;
• makes your type checker point at every missing case next time you run the codegen.

You can do the same on an interface, such as the bookshop's Publication:

"Something the bookshop sells."
interface Publication {
  id: ID!
  title: String!
  price: Money!
}

interface Printed implements Publication {
  # …
  pageCount: Int!
}

type Book implements Publication & Printed {
  # …
}

type Magazine implements Publication & Printed {
  # …
}

type Audiobook implements Publication {
  # …
  "Its length, in minutes."
  duration: Int!
}

app.graphql selects a publication's length, in pages when it is printed and in minutes when it is an audiobook:

query GetPublication($id: ID!) {
  publication(id: $id) {
    title
    ... on Printed {
      pages: pageCount
      ... on Book {
        isbn
      }
    }
    ... on Audiobook {
      duration
    }
  }
}

GetPublicationData.publication is then a union of one TypedDict per concrete type: Book, Magazine, and Audiobook. Checking for the pages key narrows it to the TypedDicts with that key, and at runtime admits any implementation of Printed added after codegen. A match on __typename narrows it to one, and assert_never() rejects any other:

def length(publication_id: str, /) -> str:
    data = send(GetPublication({"id": publication_id}))
    publication = data["publication"]

    if publication is None:
        return "No such publication."

    if "pages" in publication:
        return f"{publication['title']} has {publication['pages']} pages."

    match publication["__typename"]:
        case "Audiobook":
            return f"{publication['title']} lasts {publication['duration']} minutes."
        case _ as never:
            assert_never(never)

Type checking over runtime validation

GraphQL is strongly typed, and the server:

• validates each operation against its schema before executing it;
• responds with the selected fields and nothing else.

Client-side validation is therefore redundant when the server's schema has not changed since codegen.

Even changes made to the schema after codegen rarely break your code:

• well-behaved APIs avoid breaking changes and deprecate fields instead;
• a member added to a union, an interface, or an enum is not breaking, and you can handle it as shown above;
• a removed field makes the server reject the operation, so parse() raises a RequestError.

The few breaking changes left, such as a field becoming nullable or its leaf type changing, surface deeper in your code until you rerun the codegen against the deployed schema and your type checker detects them. If you can neither trust the server to avoid breaking changes nor rerun the codegen before it deploys one, see runtime validation.

Fragments

A fragment becomes a TypedDict of its own, extended by the TypedDict of each selection spreading it:

fragment BookCard on Book {
  id
  title
  price
  author {
    name
  }
}

query ListBooks($filter: BookFilter, $first: Int) {
  books(filter: $filter, first: $first) {
    ...BookCard
    genre
  }
}

def card(book: BookCard, /) -> str:
    author = book["author"]
    by = "an anthology" if author is None else f"by {author['name']}"
    return f"{book['title']}, {by}"


def catalog(first: int, /) -> list[str]:
    data = send(ListBooks({"first": first}))
    return [card(book) for book in data["books"]]

A fragment on an interface or a union becomes a union of one TypedDict per concrete type it covers, told apart by __typename:

query GetPages($id: ID!) {
  publication(id: $id) {
    ...PrintedLength
  }
}

fragment PrintedLength on Printed {
  title
  pages: pageCount
  ... on Book {
    isbn
  }
}

def printed_length(printed: PrintedLength, /) -> str:
    assert_type(printed["__typename"], Literal["Book", "Magazine"])
    return f"{printed['title']} has {printed['pages']} pages."


def pages(publication_id: str, /) -> str:
    data = send(GetPages({"id": publication_id}))
    publication = data["publication"]

    if publication is None or "pages" not in publication:
        return "No pages."

    return printed_length(publication)

No name clashes

Nothing prevents a schema or a document from using names that clash with Python keywords (class, from), standard library names the generated code uses (list, Sequence), or the generator's own helpers.

Since TypedDict keys are strings rather than identifiers, a GraphQL field named from keeps the key "from" in Python. An input named list keeps its name too, since the generated code refers to builtins as _builtins.list. Only the names the generator adds, such as that _builtins import alias or _GetBookData_book for the book selection of GetBookData, are chosen to avoid every other name in their module, so none can shadow another.

Runtime

Sans-IO

Shipping a transport is a slippery slope: an HTTP client to pick, then headers, authentication, timeouts, and retries. Every project has its own opinion on each of these, and the GraphQL specification itself leaves the transport to each service. The only winning move is not to play.

The small sans-IO runtime copied into the generated package therefore has a public API limited to:

from . import http as http
from .error import (
    Error as Error,
    ExecutionError as ExecutionError,
    GraphqlError as GraphqlError,
    Location as Location,
    ProtocolError as ProtocolError,
    RequestError as RequestError,
    ResponseError as ResponseError,
    UnexpectedNullError as UnexpectedNullError,
)
from .injection import OMITTED as OMITTED
from .merge import merge as merge
from .operation import (
    Operation as Operation,
    OperationRequest as OperationRequest,
)
from .request import (
    PreparedRequest as PreparedRequest,
    Request as Request,
)

In particular:

1. Operation[Type, Variables, Data].__call__() takes the Variables and returns a Request[Type, Data];
2. Request[Type, Data].prepare() returns a PreparedRequest[Data];
3. PreparedRequest.__bytes__() returns the request body to send;
4. PreparedRequest[Data].parse() returns the response body as Data.

You compose these methods with your own transport, built on any HTTP client (synchronous, asynchronous, or callback-based) or anything else that carries bytes.

The transport takes only a few lines (and stays that short as you add operations). For example, the bookshop's asynchronous transport uses httpx2:

from bookshop.graphql.runtime.http import ACCEPT, CONTENT_TYPE, is_graphql_response

http: Final = httpx2.AsyncClient(
    base_url="https://bookshop.example",
    headers={"Accept": ACCEPT, "Content-Type": CONTENT_TYPE},
    # No timeout of its own: the caller's `asyncio.timeout()` is the deadline.
    timeout=None,
)


async def send[Data](request: Request[Literal["query", "mutation"], Data], /) -> Data:
    prepared = request.prepare()
    response = await http.post("/graphql", content=bytes(prepared))

    # GraphQL over HTTP sends a request error as a response with a 4xx or 5xx status.
    if not is_graphql_response(content_type=response.headers.get("Content-Type")):
        response.raise_for_status()

    return prepared.parse(response.content)

Data then reaches your transport's caller, so each call is typed by its operation:

async def title_of(isbn: ISBN, /) -> str:
    async with asyncio.timeout(5):
        data = await send(GetBook({"lookup": {"isbn": isbn}}))

    assert_type(data, GetBookData)
    return data["book"]["title"]

transport.py does the same with the standard library alone. If your transport treats queries and mutations differently, such as sending queries with GET, you can switch on Request.operation_type.

Subscriptions

A subscription's response is a stream of bodies, one per event, and PreparedRequest.parse() reads each as it comes. The bookshop's subscribe() reads Server-Sent Events; graphql-transport-ws or multipart HTTP would be read the same way:

async def subscribe[Data](
    request: Request[Literal["subscription"], Data], /
) -> AsyncGenerator[Data, None]:
    prepared = request.prepare()

    # Closing the generator leaves the block, closes the connection, and ends the subscription.
    async with http.sse("/graphql", method="POST", content=bytes(prepared)) as events:
        async for event in events:
            if event.event == "complete":
                return

            if event.event == "next":
                yield prepared.parse(event.data.encode())

Leaving the aclosing() block unsubscribes:

async def statuses_of(order_id: str, /) -> list[OrderStatus]:
    statuses: list[OrderStatus] = []
    events = subscribe(OnOrderStatusChanged({"orderId": order_id}))

    async with aclosing(events):
        async for event in events:
            statuses.append(event["orderStatusChanged"]["status"])

            if statuses[-1] == "DELIVERED":
                break

    return statuses

Merging

Sometimes you only know at runtime which operations go together, and writing every combination as its own operation in a document file is impractical. Some other codegen libraries let you build operations at runtime for this, giving up type safety. Here, you can use merge() to combine several queries, or several mutations, written ahead of time into one request, while each result keeps its exact type:

def book_and_similar(isbn: ISBN, text: str, /) -> tuple[str, list[str]]:
    # Two queries in a single transport request.
    book_data, search_data = send(
        merge(
            GetBook({"lookup": {"isbn": isbn}}),
            Search({"text": text}),
        )
    )
    assert_type(book_data, GetBookData)
    assert_type(search_data, SearchData)

A list built at runtime can be merged too, and its results are typed as the union of its operations' data types:

def look_up(
    isbns: Sequence[ISBN], publication_ids: Sequence[str], /
) -> tuple[GetBookData | GetPublicationData, ...]:
    # Any number of queries in a single transport request.
    queries: list[
        OperationRequest[Literal["query"], GetBookData | GetPublicationData]
    ] = [
        *(GetBook({"lookup": {"isbn": isbn}}) for isbn in isbns),
        *(GetPublication({"id": id_}) for id_ in publication_ids),
    ]
    return send(merge(queries))

merge() joins the requests into one operation (telling their root fields and variables apart by an index) and splits the response back the same way. The server runs mutations in order and can resolve queries in parallel, and since most servers batch and cache per request, one merged query can execute faster than several small concurrent ones.

See the batching recipe for automatic query merging.

Errors

On a response with errors, PreparedRequest.parse() raises:

• a RequestError in place of data, when the request failed before execution;
• an ExecutionError beside partial data, when some fields failed.

class ExecutionError(ResponseError, Generic[_Data_co]):
    """The server raised errors executing the request, but sent the rest of the data.

    A field that raised is `null`, as is its nearest nullable parent if it is non-null.
    """

    data: Final[Mapping[str, object] | None]

    def parse_data(self) -> _Data_co | None:
        """Return a fresh copy of the data, converted as in a response without errors.

When several operations are merged, an ExceptionGroup holds one for each operation that fails.

In an except ExecutionError as error: block, error.parse_data() returns a Mapping[str, object] | None, since Python types what a call returns, but not what it raises. A request can return its ExecutionError instead, so the partial data keeps the operation's type:

class OperationRequest[Type: OperationType, Data](Request[Type, Data]):
    # …
    def returning_error(
        self,
    ) -> OperationRequest[Type, Data | ExecutionError[Data]]:
        """Return a copy of this request that returns its `ExecutionError` instead of raising it.

        A `RequestError` is still raised: with no data, there is nothing to type.
        """

When merging, you choose for each request whether its error is returned or raised. In a subscription, an event with errors is yielded like any other, so your transport has nothing to catch.

def track(order_id: str, /) -> str:
    result = send(GetOrder({"id": order_id}).returning_error())

    if isinstance(result, ExecutionError):
        data = result.parse_data()
        assert_type(data, GetOrderData | None)
        return f"Partially loaded: {data} ({result!s})."

    tracked = result["order"]
    return (
        "No such order."
        if tracked is None
        else f"Order {tracked['id']} is {tracked['status']}."
    )


def cancel(order_ids: list[str], /) -> list[str]:
    results = send(
        merge(
            [
                CancelOrder({"input": {"order": order_id}}).returning_error()
                for order_id in order_ids
            ]
        )
    )
    # Each error has a note naming its operation and the variables sent.
    return [
        f"{error.__notes__[0]} {error!s}"
        for error in results
        if isinstance(error, ExecutionError)
    ]

Config

Each option below is a key of extensions.spartan, next to the required moduleRoot and package.

Custom scalars

Custom scalars travel as JSON values in a format decided by the server (such as an ISO 8601 string for a date). You can give each one a Python type, with a codec converting its values when they differ from their JSON form:

extensions:
  spartan:
    # …
    scalars:
      DateTime:
        type: datetime.datetime
        codec:
          decode: ..scalar.decode_datetime
          encode: ..scalar.encode_datetime
      ISBN:
        type: ..scalar.ISBN
      Money:
        type: decimal.Decimal
        codec:
          decode: decimal.Decimal
          encode: str
      UUID:
        type: uuid.UUID
        codec:
          decode: uuid.UUID
          encode: str

A scalar without a codec must be typed as what JSON decodes to: a builtin such as str, or a NewType or alias of one, like the bookshop's ISBN = NewType("ISBN", str).

Dotted names in the config resolve as follows:

• a bare name, such as str, is a builtin;
• an absolute dotted name, such as decimal.Decimal, is imported as is;
• a path starting with .., such as ..scalar.decode_datetime, is relative to the generated package's directory.

A dotted name must be a module's attribute, so a method such as datetime.fromisoformat needs to be wrapped in a function:

from datetime import datetime
# …


def decode_datetime(value: str, /) -> datetime:
    return datetime.fromisoformat(value)


def encode_datetime(value: datetime, /) -> str:
    return value.isoformat()

Request.prepare() then encodes each scalar's values, and PreparedRequest.parse() decodes them, so your code only ever handles their Python types:

def cheaper_than(limit: Decimal, /) -> list[str]:
    data = send(ListBooks({"filter": {"priceBelow": limit}}))
    labels: list[str] = []

    for book in data["books"]:
        price = book["price"]
        assert_type(price, Decimal)
        labels.append(f"{card(book)}: {price:.2f}")

    return labels

Non-null fields

Schemas often make fields nullable, such as a lookup that may find nothing. However, you may have more context than the server and expect a specific lookup to always succeed. Or you may want your code to fail fast on a null, without writing if value is None: raise … at every use. Either way, you can declare that through a client directive, adapted from the Client Controlled Nullability RFC. The generator strips the directive from the document, so the server never sees it:

extensions:
  spartan:
    # …
    nonNullDirectiveName: nonNull

With @nonNull on book, its type loses | None:

query GetBook(
  "An identifier or an ISBN."
  $lookup: BookLookup!
  # …
) {
  book(lookup: $lookup) @nonNull {
    ...BookCard
    # …
  }
}

fragment BookCard on Book {
  # …
  author {
    name
  }
}

def describe(isbn: ISBN, /) -> str:
    data = send(GetBook({"lookup": {"isbn": isbn}}))
    book = data["book"]
    # `@nonNull` took `| None` out of `book`'s type, so no check is needed before indexing it.
    author = book["author"]
    # `@nonNull` covers `book` alone, so `author` keeps its `| None`.
    by = "an anthology" if author is None else f"by {author['name']}"

If book still comes back null, PreparedRequest.parse() raises an UnexpectedNullError:

class UnexpectedNullError(Error):
    """A field marked `@nonNull` came back `null`."""

    @property
    def path(self) -> Sequence[str | int]:
        """The keys and list indices from the root of the data to the `null`, as in a GraphQL error's `path`."""

Structs

A selection set has a fixed depth, so data of unbounded depth, such as a tree, can only come back as an untyped custom scalar. The Struct RFC proposes a new struct keyword: selected without a selection set, a field of a struct type returns its value whole, like a scalar. Until the RFC is accepted and widely adopted, you can get the same result by following this library's convention in your schema:

• a designated interface has a single field (of any name) typed with any custom scalar;
• each type that implements this interface is named after an input followed by the interface's name;
• the field's payload in each such type has the shape of that input (even a recursive one).

The generated package then types the payload with that input, and never mentions the custom scalar.

The bookshop names that interface Struct:

extensions:
  spartan:
    # …
    structInterfaceName: Struct

type Query {
  # …
  "The filter of the saved search with this name, exactly as it was saved."
  savedSearch(name: String!): BookFilterStruct
}

"A JSON payload with the shape of the input type the implementation is named after."
interface Struct {
  value: JSON
}

type BookFilterStruct implements Struct {
  value: JSON
}

"A recursive filter on books."
input BookFilter @oneOf {
  genre: Genre
  author: ID
  priceBelow: Money
  and: [BookFilter!]
  or: [BookFilter!]
  not: BookFilter
}

One query receives a BookFilter as data:

query GetSavedSearch($name: String!) {
  savedSearch(name: $name) {
    value
  }
}

Another takes a BookFilter as a variable:

query ListBooks($filter: BookFilter, $first: Int) {
  books(filter: $filter, first: $first) {
    ...BookCard
    genre
  }
}

So the same BookFilter goes from one to the other as is, typed all the way:

def run_saved_search(name: str, /) -> list[str]:
    data = send(GetSavedSearch({"name": name}))
    search = data["savedSearch"]

    if search is None:
        return []

    book_filter = search["value"]
    assert_type(book_filter, BookFilter | None)
    # Sent back as is.
    books = send(ListBooks({"filter": book_filter}))
    return [book["title"] for book in books["books"]]

Injectors

Some values are better supplied in one place than at every call.

The customer an app acts for is one example: every order carries the customer's identifier, but the functions placing orders should not have to pass that identifier around.

Idempotency is another example. An idempotency key makes retrying a mutation safe:

1. the connection drops after the server placed an order;
2. the transport sends the order again;
3. the key, unique to the order, tells the server it already placed it, so it does not charge the customer twice.

However, GraphQL has no built-in idempotency, so the key ends up as an argument or an input field of each mutation needing one, and getting hold of a key becomes every caller's concern.

You can instead supply the customer ID or the idempotency key in one place, with an injector given to Request.prepare(). The injector's value goes to every variable or input field with the name given in the config:

extensions:
  spartan:
    # …
    injectorNames: [customer, idempotencyKey]

The customer is an argument of Query.orders and a field of PlaceOrderInput and CancelOrderInput. Both inputs also hold the idempotency key:

type Query {
  # …
  "The orders of the customer to act for, the authenticated account itself when omitted."
  orders(customer: ID): [Order!]!
  # …
}

input PlaceOrderInput {
  # …
  customer: ID
  "Makes placing the same order twice harmless: the client sends a new one per order."
  idempotencyKey: UUID!
  # …
}

input CancelOrderInput {
  # …
  customer: ID
  # …
  idempotencyKey: UUID!
  order: ID!
}

ListOrders declares $customer for the injector to fill:

query ListOrders($customer: ID) {
  orders(customer: $customer) {
    id
    status
  }
}

An injected name must have the same type (nullability aside) wherever it appears. The generated package's injection.py module types the function you supply for each name: that function must return the value as its scalar's Python type, or OMITTED to leave a nullable one out. injectors.py builds them once:

from contextvars import ContextVar
from typing import Final
from uuid import uuid4

from bookshop.graphql.injection import injectors
from bookshop.graphql.runtime import OMITTED

CURRENT_CUSTOMER: Final = ContextVar[str | None]("CURRENT_CUSTOMER", default=None)


def customer() -> str | OMITTED:
    current = CURRENT_CUSTOMER.get()
    if current is None:
        # The app authenticates with its own account and acts for the customer it serves.
        # With nobody to serve, the field is left out rather than null, so the server acts for that account.
        return OMITTED
    return current


INJECTORS: Final = injectors(
    {
        "customer": customer,
        "idempotencyKey": uuid4,
    }
)

No variables' type accepts the customer ID or the idempotency key, so no caller can pass one by mistake:

def order(book_id: str, address: Address, /) -> str:
    data = send(
        PlaceOrder(
            {"input": {"lines": [{"book": book_id}], "shippingAddress": address}}
        )
    )


def order_ids() -> list[str]:
    data = send(ListOrders({}))
    return [order["id"] for order in data["orders"]]

Request.prepare() calls the injectors once, so the idempotency key is in bytes(prepared). transport.py resends that body when the connection drops:

ATTEMPTS: Final = 3
# …
    prepared = request.prepare(injectors=INJECTORS)
    # Built once, so that a retry sends the same idempotency key.
    request_body = bytes(prepared)
    attempts_left = ATTEMPTS
    # …
    while True:
        response: HTTPResponse | HTTPError

        try:
            response = _post(
                "/graphql",
                request_body,
                accept=ACCEPT,
                timeout=timeout,
            )
        # …
        except ConnectionError:
            attempts_left -= 1

            if not attempts_left:
                raise

            continue

Colocation

Most other codegen libraries gather every operation in one generated module, often inside a single class. A feature's generated operations then live away from its documents and the code calling them. A project split into several packages cannot have each package own its operations either. Since this library turns each document into its own Python module, you can opt into generating that module next to its document:

extensions:
  spartan:
    # …
    documentSiblingModule: "{document}_graphql"

This pattern puts get_order.graphql's operations and fragments in get_order_graphql.py. Code calling these operations imports them from there:

from bookshop.get_order_graphql import GetOrder
from bookshop.graphql.schema import OrderStatus

Generator API

To run the generator programmatically, call generate():

class _Params(TypedDict, closed=True):
    document: DocumentNode
    schema: GraphQLSchema
    config: Config


def generate(**args: Unpack[_Params]) -> dict[PurePosixPath, bytes]:
    """Pure function returning the content of each file of the generated package by its path."""

Config and the types it is made of complete the public API:

from spartan_graphql.config import Config as Config
from spartan_graphql.document_sibling_module import (
    DocumentSiblingModule as DocumentSiblingModule,
)
from spartan_graphql.generate import generate as generate
from spartan_graphql.package_location import PackageLocation as PackageLocation
from spartan_graphql.scalar import Codec as Codec, Scalar as Scalar

Coming from another library

Most GraphQL code generators for Python give you a client class with one method per operation, with the variables as parameters and the response data as a model class. Here, the operation variables and response data are TypedDicts. The rest of the client is the transport you write. With no class gathering the operations, a document's operations can live in a module next to the document: see colocation.

Calling an operation:

-from bookshop.client import client
+from bookshop.transport import send

-data = await client.get_order(id="o1")
-order = data.order
+data = send(GetOrder({"id": "o1"}))
+order = data["order"]

Reading an enum as a Literal:

 def status_label(status: OrderStatus, /) -> str:
     match status:
-        case OrderStatus.PENDING:
+        case "PENDING":
             return "Being prepared"

Telling the members of a union or an interface apart by the presence of a key or by __typename rather than by class:

-    if isinstance(publication, PublicationBook | PublicationMagazine):
-        return f"{publication.title} has {publication.pages} pages."
+    if "pages" in publication:
+        return f"{publication['title']} has {publication['pages']} pages."

-    match publication:
-        case PublicationAudiobook():
-            return f"{publication.title} lasts {publication.duration} minutes."
+    match publication["__typename"]:
+        case "Audiobook":
+            return f"{publication['title']} lasts {publication['duration']} minutes."

Handling a partial response (returned rather than raised) with its data still typed:

-    try:
-        result = await client.get_order(id=order_id)
-    except GraphQLClientError as error:
-        assert_type(error.data, dict[str, Any] | None)
-        return f"Partially loaded: {error.data} ({error})."
+    result = send(GetOrder({"id": order_id}).returning_error())
+
+    if isinstance(result, ExecutionError):
+        data = result.parse_data()
+        assert_type(data, GetOrderData | None)
+        return f"Partially loaded: {data} ({result!s})."

Configuring in graphql-config rather than in pyproject.toml:

-[tool.graphql.codegen]
-schema_path = "schema.graphqls"
-queries_path = "."
-target_package_name = "client"
+schema: schema.graphqls
+documents: "*.graphql"
+# …
+extensions:
+  spartan:
+    # …
+    moduleRoot: ..
+    # …
+    package: bookshop.graphql

Recipes

Each recipe below commits to a choice Spartan GraphQL leaves to you, such as an observability stack, an async framework, an HTTP client, or a validation library. The library is designed to make each of them a few lines on your side, so copy them, adapt them, or combine them into the GraphQL client you need.

Tracing

With this send() as your transport, a slow page or a failing call points at the operation behind it, in your traces and in the server's, following OpenTelemetry's GraphQL conventions:

from opentelemetry.semconv._incubating.attributes import graphql_attributes  # noqa: PLC2701 (not stable yet)
# …
async def send[Data](request: Request[Literal["query", "mutation"], Data], /) -> Data:
    with tracer.start_as_current_span(
        f"{request.operation_type} {request.operation_name}",
        kind=SpanKind.CLIENT,
        attributes={
            graphql_attributes.GRAPHQL_OPERATION_NAME: request.operation_name,
            graphql_attributes.GRAPHQL_OPERATION_TYPE: request.operation_type,
        },
    ) as span:
        # In the span, so that it records variables failing to encode.
        prepared = request.prepare()
        # As headers, so that the server's spans join the trace.
        trace_context: dict[str, str] = {}
        inject(trace_context)

        response = await http.post(
            "/graphql", content=bytes(prepared), headers=trace_context
        )

        if not is_graphql_response(content_type=response.headers.get("Content-Type")):
            response.raise_for_status()

        data = prepared.parse(response.content)
        # A merge's data is a tuple, with each returned error in its request's place.
        results = data if isinstance(data, tuple) else (data,)
        # A raised error, `RequestError` included, marks the span on its own, but a returned one does not.
        if errors := [
            str(result) for result in results if isinstance(result, ExecutionError)
        ]:
            span.set_status(StatusCode.ERROR, "; ".join(errors))

        return data

Batching

With a QueryBatcher, your data-fetching functions stay independent, but their queries still get merged into a single request, letting the server optimize their execution:

def test_the_requests_of_the_same_tick_are_merged() -> None:
    # …
    batcher = QueryBatcher(send)

    async def order(order_id: str, /) -> GetOrderData:
        return await batcher.send(GetOrder({"id": order_id}))

    async def page() -> tuple[GetOrderData, GetOrderData]:
        # Two parts of a page, each fetching for itself, started in the same tick.
        header = asyncio.create_task(order("header"))
        footer = asyncio.create_task(order("footer"))
        return await header, await footer

    first, second = asyncio.run(page())

    assert_type(first, GetOrderData)
    # …
    # Both queries reached the transport as one merged request.
    (request,) = sent_requests
    assert request.operation_name == "BatchedQueries"

class QueryBatcher:
    """Sends the queries of one event-loop tick as one merged request, as a DataLoader does on a server."""
    # …
    async def send[Data](
        self, request: OperationRequest[Literal["query"], Data], /
    ) -> Data:
        if self._batch is None:
            requests: _Requests = []
            # Not `asyncio.create_task()`, which an eager task factory would run at once.
            # The task starts on the next tick, after the other coroutines of this tick added their request.
            task = asyncio.Task(self._run(requests))
            self._batch = _Batch(requests=requests, task=task)

        batch = self._batch
        index = len(batch.requests)
        batch.requests.append(request)
        batch.waiter_count += 1

        try:
            # A caller can give up, such as on its `asyncio.timeout()`.
            # The shield then leaves the others their data.
            results = await asyncio.shield(batch.task)
        finally:
            batch.waiter_count -= 1

            # No caller waits anymore, so neither does the transport.
            if not batch.waiter_count:
                batch.task.cancel()

        result = results[index]

        if isinstance(result, ExecutionError):
            match request.execution_error_handling:
                case "raise":
                    raise result
                case "return":
                    pass
                case _ as never:
                    assert_never(never)

        # `Data` includes the error of a request returning it.
        return cast(Data, result)

    async def _run(self, requests: _Requests, /) -> tuple[object, ...]:
        # The next tick starts another batch.
        self._batch = None
        merged = merge(
            [
                # So that a failed request leaves the others their data.
                request.returning_error()
                for request in requests
            ],
            operation_name="BatchedQueries",
        )
        return await self._send(merged)

File uploads

Uploading files through GraphQL is discouraged, but there is a community spec for it. Servers following the spec define a custom Upload scalar and expect clients to send the files in a multipart HTTP request. On the client side, you configure the Upload scalar with this Python type:

@dataclass(frozen=True, kw_only=True)
class Upload:
    content: BinaryIO
    filename: str
    content_type: str = "application/octet-stream"

def test_a_file_goes_in_the_variables_like_any_other_value(
    # …
) -> None:
    cover = Upload(content=BytesIO(b"cover"), filename="cover.txt")
    scan = Upload(content=BytesIO(b"scan"), filename="scan.txt")
    request = AttachFiles(
        {
            "isbn": "9780140449433",
            "cover": cover,
            "attachments": [{"file": scan, "caption": "p. 3"}, {"file": cover}],
        }
    )

    data = asyncio.run(send(request))

The scalar's encode() collects each Upload instance as the request is prepared:

_collected: ContextVar[tuple[Upload, ...]] = ContextVar("_collected")


def encode(value: Upload, /) -> str:
    uploads = _collected.get()

    # The same file in several places is sent once, mapped to each of them.
    index = next((i for i, upload in enumerate(uploads) if upload is value), None)

    if index is None:
        index = len(uploads)
        _collected.set((*uploads, value))

    return f"{_PREFIX}{index}"


def _collect_uploads[T](prepare: Callable[[], T], /) -> tuple[T, tuple[Upload, ...]]:
    context = copy_context()
    context.run(_collected.set, ())
    return context.run(prepare), context[_collected]
# …
async def send[Data](request: Request[Literal["query", "mutation"], Data], /) -> Data:
    prepared, uploads = _collect_uploads(request.prepare)
    post = partial(http.post, "/graphql")

    if uploads:
        response = await post(
            data=_multipart_fields(bytes(prepared)),
            files={
                str(index): (upload.filename, upload.content, upload.content_type)
                for index, upload in enumerate(uploads)
            },
        )
    else:
        response = await post(
            content=bytes(prepared), headers={"Content-Type": CONTENT_TYPE}
        )

Runtime validation

When the server deploys more often than you rerun the codegen, such as a third-party API, a breaking change can reach your code before your type checker sees it. validate() catches it as the response arrives rather than deeper in your code:

def test_a_field_made_nullable_after_codegen_is_caught() -> None:
    request = GetOrder({"id": "o1"})
    body = b'{"data": {"order": {"id": "o1", "status": null}}}'
    data = request.prepare().parse(body)

    with pytest.raises(
        ValidationError,
        match=(
            r"^1 validation error for GetOrderData\n"
            r"order\.status\n"
            r"  Input should be 'PENDING' or 'SHIPPED' "
        ),
    ):
        validate(data, request=request)

def validate[Data](data: Data, /, *, request: Request[Any, Data]) -> None:
    if isinstance(data, tuple):
        # A merge's data, in which each request may have returned its error.
        for item, data_type in zip(data, get_args(request.data_type), strict=True):
            _validate(item, data_type)
    else:
        _validate(data, request.data_type)


def _validate(data: object, data_type: type[object], /) -> None:
    match data:
        case ExecutionError():
            # An error can leave no data at all.
            if (partial_data := data.parse_data()) is not None:
                _validate(partial_data, data_type)
        case _:
            _adapter(
                # mypy does not see a type as hashable.
                data_type  # type: ignore[arg-type]
            ).validate_python(data, strict=True)

Call validate() in your transport, right after parse(), so that no response goes unchecked.

Fuzzing

data_strategy() implements a Hypothesis search strategy to fuzz your functions accepting GraphQL response data without any running GraphQL server:

from hypothesis import given, strategies as st
# …
def ean_13_barcode(data: GetOrderData, /) -> int | None:
    order = data["order"]

    if order is None:
        return None

    gs1_prefix = order["isbn"][:3]
    assert gs1_prefix in {"978", "979"}
    return int(order["isbn"])


@given(example=st.data())
def test_your_code_handles_any_order_the_server_can_send(
    example: st.DataObject,
) -> None:
    data = example.draw(data_strategy(GetOrder({"id": "o1"})))
    assert_type(data, GetOrderData)
    ean_13_barcode(data)

ISBN = NewType("ISBN", str)

# Without it, any `str` is drawn: the supertype of the `NewType`.
st.register_type_strategy(
    # Hypothesis supports a `NewType` here, but its annotation leaves it out.
    cast(Any, ISBN),
    st.from_regex(r"97[89][0-9]{10}", fullmatch=True),
)


def data_strategy[Data](request: Request[Any, Data], /) -> st.SearchStrategy[Data]:
    return st.from_type(request.data_type)

1. Before Python 3.15, the generated package needs typing-extensions to access the latest typing features. The generator declares that dependency, but the generated package is copied into your project rather than installed, so add typing-extensions to your own dependencies. ↩ ↩² ↩³

2. The generated code is tested against mypy, Pyright, Pyrefly, and ty. ↩

3. This library is partly bootstrapped: to fetch a schema from a URL, it uses an operation it generated itself from _introspection.graphql. ↩

4. A YAML config, like the quick start's, needs the yaml extra: install spartan-graphql[yaml] instead. JSON and TOML configs need nothing more. ↩