Metadata-Version: 2.4
Name: iron-gql
Version: 0.13.4
Summary: iron_gql generates typed sync and async GraphQL clients and runtime helpers from schemas for Python services
Keywords: graphql,gql,client,httpx2,pydantic,codegen,async,sync
Author: Ilia Ablamonov
Author-email: Ilia Ablamonov <ilia@flamefork.ru>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Requires-Dist: httpx2[ws]>=2.7,<3
Requires-Dist: pydantic>=2.12.3,<3
Requires-Dist: graphql-core>=3.3.0,<3.4 ; extra == 'codegen'
Requires-Dist: uvicorn>=0.34 ; extra == 'testing'
Requires-Python: >=3.13
Project-URL: Homepage, https://github.com/Flamefork/iron_gql
Project-URL: Repository, https://github.com/Flamefork/iron_gql.git
Project-URL: Issues, https://github.com/Flamefork/iron_gql/issues
Provides-Extra: codegen
Provides-Extra: testing
Description-Content-Type: text/markdown

# iron_gql

[![License](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
[![main](https://github.com/Flamefork/iron_gql/actions/workflows/main.yml/badge.svg)](https://github.com/Flamefork/iron_gql/actions/workflows/main.yml)
[![PyPI - Version](https://img.shields.io/pypi/v/iron-gql)](https://pypi.org/project/iron-gql/)


`iron_gql` is a GraphQL code generator and runtime. It reads a schema SDL and your query documents, and it generates a typed Python client with Pydantic models. You can connect GraphQL APIs to services, CLIs, background jobs, and tests without hand-written boilerplate.

## Installation

```bash
pip install iron-gql            # runtime only (httpx2 + pydantic)
pip install iron-gql[codegen]   # + graphql-core for code generation
pip install iron-gql[testing]   # + uvicorn for the loopback test server
```

## Key Features
- **Query discovery.** `render_gql_module` scans your codebase for calls of the form `<module>_gql("""...""")`. It validates each statement and writes a module with typed helpers.
- **Typed inputs and results.** The generated Pydantic models match every selection set, enum, and input object that the discovered queries reference.
- **Sync or async runtime.** Pick the mode per module with `mode="sync"` or `mode="async"`. The generated module targets `runtime.GQLClient` or `runtime.AsyncGQLClient`, both of which send requests through `httpx2`. One project can hold modules of both kinds.
- **Subscriptions over SSE or WebSocket.** A client subscribes by [GraphQL over SSE](https://github.com/enisdenjo/graphql-sse/blob/master/PROTOCOL.md) (one stream per operation) or by [`graphql-transport-ws`](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md). You name the protocol when you create the client; see [Subscriptions](#subscriptions).
- **ASGI in-process calls.** `AsyncGQLClient` accepts an ASGI `target_app` and then calls the app in-process without using the network: queries, mutations, and `graphql-transport-ws` subscriptions. An SSE subscription needs a socket, because the ASGI transport of `httpx2` holds the response body until the app returns. The sync client has no in-process transport: the ASGI transport of `httpx2` is async-only, and WSGI cannot carry websockets. Test synchronous modules and SSE subscriptions against a real server on a loopback port, which [`iron_gql.testing`](#testing) starts for you.
- **Deterministic validation.** `graphql-core` (a codegen dependency) validates every statement against the schema. It rejects operations that share a name but have different bodies.

## Package Layout
- `runtime.py` contains `GQLClient` and `AsyncGQLClient`, the reusable `GQLOperation` base class, and the value serialization helpers.
- `codegen/generate.py` runs query discovery, validation, and module rendering.
- `codegen/parser.py` converts the GraphQL AST into typed helper structures for the renderer.
- `testing/` holds the test helpers: the client swap, the `graphql-transport-ws` server primitives, and the loopback servers.

## Getting Started
1. **Describe your schema.** Write the schema in an SDL file (`schema.graphql`). Include the root types that you use (query, mutation, subscription).
2. **Write queries where you use them.** Import the helper that the generator will create. Wrap each GraphQL statement in a call to this helper:
   ```python
   from myapp.gql.client import client_gql

   get_user = client_gql(
       """
       query GetUser($id: ID!) {
           user(id: $id) {
               id
               name
           }
       }
       """
   )
   ```
   The generator derives the helper name (`client_gql`) from the module path that you ask it to build.
3. **Generate the client module.**
   ```python
   from pathlib import Path

   from iron_gql.codegen import render_gql_module

   render_gql_module(
       mode="async",
       schema_path=Path("schema.graphql"),
       src_path=Path("."),
       module_full_name="myapp.gql.client",
       base_url_import="myapp.config:GRAPHQL_URL",
       subscription_protocol="graphql-sse",
       scalars={"ID": "builtins:str"},
       to_camel_fn_full_name="myapp.inflection:to_camel",
       to_snake_fn=my_project_to_snake,
       debug_path=Path("iron_gql/debug/myapp.gql.client"),
   ).write()
   ```
   `mode` is required: pass `"async"` or `"sync"` for every module. Each module is self-contained, so one project can generate both.

   `subscription_protocol` is required too: `"graphql-sse"` or `"graphql-transport-ws"`. The generated module creates its client with it, so the module's subscriptions speak the protocol of your server.

   `src_path` is the import root: the generator derives a module's name from the file's path relative to it, so the root must be what the project's absolute imports are written relative to (`Path(".")` for a flat layout, `Path("src")` for a src layout). Below this root the scan does not descend into a hidden directory (its name starts with a dot) or into the root of a virtual environment (a directory containing `pyvenv.cfg`). Everything else is walked whatever it is called: statements are collected even from a directory whose name cannot be part of a module name — a script is run, not imported.

   This writes `myapp/gql/client.py`. The module contains:
   - a client singleton for the chosen mode,
   - Pydantic result and input models,
   - a class per operation with typed `execute` methods (see [Generated Names](#generated-names)),
   - overloads for the helper function, so editors can infer return types.

   To check that a committed module is current without writing it, call `diff()` instead of `write()`. It returns a unified diff from the file to the rendered module, or an empty string if the file is current.
4. **Call your API.** An async module awaits `execute`:
   ```python
   async def fetch_user(user_id: str):
       query = get_user.with_headers({"Authorization": "Bearer token"})
       result = await query.execute(id=user_id)
       return result.user
   ```
   A sync module calls it directly:
   ```python
   def fetch_user(user_id: str):
       query = get_user.with_headers({"Authorization": "Bearer token"})
       result = query.execute(id=user_id)
       return result.user
   ```

## Generated Names

Classes named after what you wrote carry their kind, without exceptions. `query GetUser` gives the operation class `GetUserQuery` and its result model `GetUserQueryResult`; `query GetUserQuery` gives `GetUserQueryQuery`. Mutations and subscriptions end in `Mutation` and `Subscription`, a template adds the bound base `GetUserQueryBound`, and `fragment ImageUrl` gives `ImageUrlFragment` with the data model `ImageUrlData`. You get these classes from `<module>_gql("...")`, so you rarely write their names.

Result models are shared by every operation of the module that selects the same shape, and are named after their GraphQL type:

- `User` — the module holds one shape of `User`;
- `UserWithIdName` — the module holds one shape of `User` with exactly the fields `id` and `name`;
- otherwise `UserWithIdName_<tokens>`, which spells each field's type one level deep (`PostWithAuthor_Opt_UserWithName`), and a digest of the shape (`…_1a2b3c4d`) after it when two shapes still spell alike.

An interface selection that gives some of its possible types an `... on X` variant gives the rest one fallback model; its detailed name spells its `__typename` as `Except` and the types that do get a variant: `NodeWithTypename_ExceptPost`. A union selection has no fallback model: each of its possible types gets its own variant.

A union result, or an interface result with `... on X` variants, is a `type` alias, a result union alias: `Media` when it is the module's one shape of `Media`, `Media_<digest>` otherwise. An interface's fallback model and a result union alias over that interface are two shapes of one type, so neither is named after it.

A name belongs to one shape or to none. Adding a query never hands a name you import to another shape: it can take the name away, and then your import fails. A class you named never moves either — an operation, template or fragment class, their bases and results, an `On{Type}` base, every model of a template slot, and every model and result union alias of a fragment's data. If a model or a result union alias may claim its name, generation fails and asks you to rename the operation, template or fragment.

A model or a result union alias skips a name that would shadow one its module reads: a builtin the module writes (`str`, `list`, `LookupError`, …), and the name of a result field with a default, such as one under `@skip`/`@include`. A type `str` gives `strWithId`.

## Custom Scalars

The generator maps GraphQL scalars to Python types in two layers.

It maps **built-in scalars** automatically:

| GraphQL | Python |
|---------|--------|
| `String`, `Int`, `Float`, `Boolean` | `str`, `int`, `float`, `bool` |
| `Date` | `datetime.date` |
| `DateTime` | `datetime.datetime` |
| `JSON` | `object` |
| `Upload` | `iron_gql.FileVar` |

You configure **custom scalars** with the `scalars` parameter, in `"module:type"` format:

```python
render_gql_module(
    ...,
    scalars={
        "ID": "builtins:str",
        "Money": "decimal:Decimal",
        "ULID": "ulid:ULID",
    },
)
```

Custom scalar types must be Pydantic-compatible. That is, Pydantic must know how to parse the type from JSON (deserialization) and how to serialize it to JSON. Standard library types (`datetime`, `Decimal`, `UUID`, `Enum`) are compatible by default. Any type that implements `__get_pydantic_core_schema__` is also compatible. The generator maps unknown scalars to `object` and writes a warning to the log.

## Fragment Slots

Shared infrastructure code often owns a GraphQL operation, but it does not know in advance which fields each caller needs on some field of that operation. A fragment slot lets each caller supply its own fragment for that field. The generator resolves every slot at codegen time, so each caller ends up with its own fully static operation — no query text is assembled at call time.

Mark a field with `@slot` in a query, mutation, or subscription. Do not use `@slot` inside a fragment definition, and do not nest a slot inside another slot's own selection. Give the field a static selection that selects `__typename` at the top level of its own selection set. This `__typename` must be unaliased and must have no directives. It must not come through an inline fragment or a fragment spread. The slot field itself cannot have `@skip` or `@include`.

An operation that contains a `@slot` is a **template**. It has no `execute` of its own, only `bind`. Write it wherever you write any other statement — a module-level name is how you share one template between modules, not a condition of using it:

```python
get_post_attachment = api_gql("""
    query GetPostAttachment($id: ID!) {
        post(id: $id) {
            id
            attachment @slot { __typename }
        }
    }
""")
```

A statement with exactly one fragment definition returns a typed **fragment definition value** if the package contains a template. Each `api_gql()` call creates a new equivalent value of the same generated type. A package without a template has no binding context, so its fragment statements keep their old meaning. The generator creates an `On{Type}` base for each type condition in use. Plain definitions and factory applications inherit the base of their type condition; infrastructure code can accept any bindable fragment on that type through it. A statement with several definitions stays a bundle whose fragments can be spread by name.

```python
image_url = api_gql("""
    fragment ImageUrl on ImageAttachment {
        url
    }
""")

link_url = api_gql("""
    fragment LinkUrl on LinkAttachment {
        href
    }
""")
```

### Binding fragments to a template

Link a template to the fragments it should carry with `bind()`:

```python
get_post_attachment_image = get_post_attachment.bind(attachment=image_url)
get_post_attachment_link = get_post_attachment.bind(attachment=link_url)
```

The keyword argument is the snake_case form of the slot field's name, or of its alias if it has one.

`bind` is an ordinary method call, not a statement the generator has to read. Every combination of a template with a fragment compatible with its slot is generated from the schema, whether or not any call site writes it — so the fragment can arrive as a function parameter, and the template it fills need never be exported at all:

```python
async def attachment_of[TModel: pydantic.BaseModel, TReads](
    post_id: str, details: OnImageAttachment[TModel, TReads]
) -> TModel | None:
    bound = get_post_attachment.bind(attachment=details)
    result = await bound.execute(id=post_id)
    return details.read(result.post.attachment) if result.post else None
```

The caller passes its own fragment and gets back that fragment's own model. The helper never names the fragment, and its own module can keep the template private — nothing outside has to see it:

```python
caption = await attachment_of("p-1", image_caption)
```

This is what the `On{Type}` bases are for. A helper for any fragment on `ImageAttachment` accepts `OnImageAttachment[TModel, TReads]` and returns `TModel`; a helper that returns the whole result to its caller expresses the result type through both of the fragment's parameters. Either way the exact type is kept: a fragment's model is unique to its definition, so another fragment's data cannot be read from the slot by mistake.

A bind can also be written literally, with the template and fragments local to the function or written straight into the call:

```python
async def show_attachment(post_id: str) -> None:
    bound = get_post_attachment.bind(attachment=api_gql("""
        fragment ImageAlt on ImageAttachment {
            alt
        }
    """))
    result = await bound.execute(id=post_id)
```

A slot can also take a tuple of fragments. A tuple is the one form the generator still reads from your source: single fragments are enumerated from the schema, but every subset of the compatible fragments is not (that is exponential), so a multi-fragment combination exists only where a call site writes it literally. A tuple and not a list, because the overload written for it names a fixed length: a list of two fragments would also admit a list of one, and the phantom would then offer a fragment that shorter binding never bound. Pass a list and the generator says so, naming the call site. Their covered runtime types may overlap — each fragment reads its own slice of the payload independently. What the generator still rejects is a field-merge conflict in the expanded operation (the same response key selected with different arguments), reported by standard GraphQL validation. A fragment bound alone to a slot and the same fragment inside a tuple bind of that slot coexist:

```python
image_caption = api_gql("""
    fragment ImageCaption on ImageAttachment {
        caption
    }
""")

link_summary = api_gql("""
    fragment LinkSummary on LinkAttachment {
        href
    }
""")

get_post_attachment_any = get_post_attachment.bind(
    attachment=(image_caption, link_summary)
)
```

Slots you do not name in `bind()` stay unfilled: the operation still requests their static `__typename` selection, but no fragment data comes back for them. Omitting a slot and passing it an explicit empty sequence (`attachment=()`, or `attachment=[]`) mean the same thing — an empty one names no fragment, so nothing is written for it that a shorter call could land on. The same fragment can be bound into any number of templates, and a template can have any number of binds.

A binding **is** its combination: the template and the fragments in each slot. The generator derives empty and single-fragment variants from the schema. It also adds multi-fragment combinations that call sites specify as literals. Each template stores one binding spec for every canonical combination, so duplicate call sites share one spec.

The limit of 256 applies to the final canonical set for each template. This set includes schema-derived and literal multi-fragment combinations after deduplication. If the set exceeds the limit, the error reports the total and the schema and literal contributions. Split the template or narrow its slot types.

### Executing and reading

The generator writes one class per template, `{Name}{Kind}Bound[TResult]` (`GetPostAttachmentQueryBound` for `query GetPostAttachment`), with a single type parameter: the result this binding returns. `bind()` returns an instance of it carrying that combination's own operation text, and its `execute` answers with exactly the result the parameter names. Which fragments are readable in each slot is recorded by the result type itself, so `result.post.attachment` still knows what was offered — a slot left unfilled is statically unreadable. Code that works over any binding of one template takes the class with its parameter open and hands the result back to a caller who knows which binding it passed:

```python
async def fetch[TResult: pydantic.BaseModel](post_id: str, bound: GetPostAttachmentQueryBound[TResult]) -> TResult:
    return await bound.execute(id=post_id)

result = await fetch("1", get_post_attachment_image)
image = image_url.read(result.post.attachment) if result.post else None
```

The result models belong to the template, not to each binding. `GetPostAttachmentQueryResult` and its nested models have one generic type parameter per slot. A binding fills these parameters, for example `GetPostAttachmentQueryResult[ImageUrlFragment | LinkUrlFragment]`. Thus, one template writes one set of models for all its bindings, and a validation error names that shared model rather than any one combination. A helper can use `GetPostAttachmentQueryResult[Any]` to accept the result of any binding (see below).

Pickle support depends on which generic models the payload instantiates. A populated path to a slot creates nested Pydantic parametrizations without module-level pickle names. `pickle.dumps()` raises `PicklingError` for such a result. If `None` stops every slot path before a nested generic model is created, only the registered root specialization is instantiated. `{"post": null}` creates no nested model, so its result can cross a process boundary.

`execute` takes only the template's own variables. There is no slot keyword argument: the fragment selection already happened at `bind()` time.

Read a slot through a fragment definition value: `image_caption.read(result.post.attachment)` returns the fragment's own model, or `None` if the node is `None` or its runtime type is outside the fragment's coverage. Any fragment whose fields landed directly in the slot payload is read this way: a fragment passed to `bind()`, or a fragment that another bound fragment spreads next to its own fields. A new equivalent definition from a repeated `api_gql()` call reads the same result:

```python
result = await get_post_attachment_any.execute(id="p-1")
same_definition = api_gql("""
    fragment ImageCaption on ImageAttachment {
        caption
    }
""")
attachment = same_definition.read(result.post.attachment)
```

`fragment.read(node)` is checked statically: the node's type holds the definitions the binding offered for this slot. Reading a definition that was not offered — including any read of an unfilled slot — is a type error. On the type-erased path the same wiring mistake raises `ValueError` instead of returning `None`, so that a wiring bug does not look like an acceptable runtime type mismatch.

This check works as long as the exact type of the definition or application is kept. For a heterogeneous registry, give both type parameters explicitly: `GQLFragment[pydantic.BaseModel, Any]`. If the exact readable contract is already erased, use `{Name}{Kind}Result[Any]` as well: it accepts the result of any binding and supports the same `read`. Only the static check is lost; at runtime only a definition the binding offered for the slot is still accepted.

This way shared infrastructure can own an operation and return data instead of the raw result: the helper writes `Any` for the slot it does not know and reads through the fragment its caller passed.

```python
def title[TData: pydantic.BaseModel, TReads](
    result: GetPostAttachmentQueryResult[Any], fragment: GQLFragment[TData, TReads]
) -> TData | None:
    return fragment.read(result.post.attachment) if result.post else None
```

A fragment reached through a *narrower* type condition — an interface brick spread inside a per-type fragment — reads back as `None` for the types that condition excludes, exactly like any other type mismatch.

A fragment spread inside a bound fragment's *nested field* behaves differently: its fields arrive under that field, not in the slot payload, so the slot cannot read them directly. The data is available through the enclosing fragment's model (`image_parts.read(node).thumb.alt`); the nested definition is not in the node's offered set, so a direct `read` stays a static type error and raises `ValueError` at runtime.

Fragments are isolated from each other — each one reads exactly its own selection, never the fields another caller's fragment selected. The data is not part of the fields of the result model, so `model_dump()` does not include it; re-validating a dump without a fragments context fails immediately.

### Fragment variables

Fragments can spread other fragments and use variables of their own. For each variable the generator synthesizes a GraphQL declaration that is valid at every place of use. If no such declaration exists, generation fails. A fragment whose closure uses a variable becomes a **fragment factory**. A factory definition itself cannot be passed to `bind()`; the typed `with_args` creates a bindable **fragment application** carrying the variable values:

```python
image_thumbnail = api_gql("""
    fragment ImageThumbnail on ImageAttachment {
        thumbnail(width: $width)
    }
""")
```

Call `with_args` where the values are known and pass the application to `bind()`. The result can be read through the factory definition or through any application of the same factory; `read()` checks the definition, not where the argument values came from:

```python
async def show_thumbnail(post_id: str, width: int) -> None:
    bound = get_post_attachment.bind(
        attachment=image_thumbnail.with_args(width=width)
    )
    result = await bound.with_headers({"Authorization": "..."}).execute(id=post_id)
    if result.post is not None:
        image = image_thumbnail.read(result.post.attachment)
```

A required fragment variable maps to a required keyword in `with_args`. `None` always sends an explicit GraphQL `null`. If every usage position is nullable and each has a schema default, the keyword can be omitted: the variable is left out of the request and the schema default applies. A non-null position with a schema default still requires the keyword: graphql-core 3.3, like graphql-js 17, ignores a position's default for a variable inside a fragment, and a nullable declaration would be rejected by both the codegen and the server. Omission and `None` are different states. The exact `with_args` signature rejects missing required keywords and extra ones. An application's set of variables is immutable. The applied class stays private; create applications only through `with_args`.

Each `with_args` returns a new application. Applications of one factory can carry different arguments but read the same projection. `read()` fails only for a definition of another factory or another fragment.

Two applications of one factory can fill two slots of one template. They read their own slots, but they cannot supply two independent values for one GraphQL variable. If the applications disagree, `bind()` reports the variable and both sources without showing either value. Use the same value, or use two fragments with different names.

If a template already spreads a factory by name, the operation owns that factory's variables and `execute` supplies them. A second source exists only when the factory can also bind to a compatible slot of that template. The generator rejects this reachable conflict. A static spread is valid when the template has no compatible slot.

### Validation

The runtime validates every fragment readable at each slot at the response boundary — including a slot nobody read and a fragment reached only through another fragment's spread. For queries and mutations, validation happens inside `execute`. For subscriptions, it happens on every received message.

### What the generator rejects

The scanner reads a complete combination only when a `.bind()` call contains a non-empty literal sequence or a literal `**{...}` mapping. A multi-fragment tuple is the only combination that the schema cannot derive. The scanner also reads a non-empty list to report that it has no fixed length. Empty `()` and `[]` values contain no static fragment. The scanner checks their keyword names but does not resolve adjacent single values. A named `**opts` mapping is also unresolved because its contents are not statically known. The scanner ignores calls where every slot receives one value. This rule lets a helper bind a fragment that it receives as a parameter. Therefore, the following rules apply to tuple bindings.

Generation fails with `GraphQLGenerationError` ([Validation and Troubleshooting](#validation-and-troubleshooting)), naming the file and line, for:
- a tuple `.bind()` call the scan cannot read statically: a positional argument, a `**` spread (the resolver has no keyword name to thread through, whether the spread stands alone or beside another slot's tuple), a repeated slot keyword, a slot value that is neither a name nor an inline `api_gql("...")` call nor a tuple of those, or a template reached through an attribute chain into the scanned tree (`import infra` then `infra.template.bind(...)`) instead of a directly imported name. Once a call is read at all, every one of its slots has to resolve — the binding key is built from the whole call, so a resolvable tuple beside an unresolvable single value is an error rather than a partial read;
- a slot given a non-empty list rather than a tuple: no fixed-length overload can be written for a list, so one written for two fragments would accept one of them and offer a phantom that binding never bound. An empty list names no fragment and stays legal;
- a name such a bind reads that is bound more than once in the scope it resolves in — two assignments, two imports, or a mix; which of them it means is a question the scan will not guess at;
- a slot value of such a bind that does not resolve to a discovered fragment statement. The template it hangs off is judged differently, because `.bind()` is an ordinary method name that sockets and widgets carry too: a base that resolves to nothing at all leaves the call alone, and only a base whose name the scanned tree does hold as a statement is an error — reported with that statement's own address, since a template written where no resolution reaches it (a class body, a star import) is ours all the same. A name something else *does* bind where the call stands is answered by that binding and left alone: a same-named statement elsewhere in the tree is a coincidence, and two generator runs over one tree produce it routinely;
- a template whose final deduplicated set of schema-derived and literal combinations exceeds 256. The error reports the total and both contributions;
- a source file the scan cannot parse that mentions the gql function or `bind` — it may hold statements or binds, and dropping it would rewrite the module without them. A file that names neither owns nothing the module could lose and is skipped, unless some bind's name resolution has to travel through it;
- a fragment that is not spread-compatible with the slot — the same mismatch is also a type error at the `bind()` call site, so most callers see it in the editor before they even regenerate;
- a generated name claimed twice by an actual `On{Type}` base, a factory, a private Applied class, a fragment, a model, or an enum. A fragment name can also collide with a type parameter in another slot's `bind()` overload. Rename the fragment or add an alias to the field that creates the model;
- a class named after an operation, a template or a fragment — the operation, template or fragment class, its bound base, its result, an `On{Type}` base, a model of a slot, a model or result union alias of a fragment's data — whose name a result model or result union alias may claim, whichever of its names that shape ends up with ([Generated Names](#generated-names)). Rename the operation, template or fragment;
- an enum named like a builtin the module reads (`enum str`), and a field with a default — under `@skip`/`@include`, or an optional input field — named like anything its model's annotations read: a builtin, a scalar's module, an enum (`str: name @include(if: $full)`). Pydantic would resolve the annotations to the default. Alias the field;
- a fragment readable at a slot's root reached under `@skip`/`@include` — whether the directive sits on the spread or on an inline fragment around it — at any runtime type where that conditional path is the only way it reaches the root, since such a fragment is requested and validated on every response and so cannot be conditionally absent. Reaching it unconditionally as well — bound directly, or spread again without a directive — covers the types reached that way, and only the types left over are rejected. The same directive on a spread inside a nested field is fine;
- a bound fragment whose name a template also defines locally, with a different definition — one name is one definition in the expanded document;
- a factory that a template spreads by name and can also bind to a compatible slot in that template. `execute` and `with_args` would become two sources for one GraphQL variable;
- two slots of one template whose names collapse to the same Python name — a slot's name is its `bind()` keyword, and one keyword cannot mean two slots;
- `@slot` inside a fragment definition, or a slot nested inside another slot's own selection.

One slot name selected under two parents is not an error: both positions carry the same spliced fragments, so every fragment reachable from either position reads through its definition or application. A slot on a polymorphic parent works the same way — one node model per variant.

A combination with no text of its own raises `LookupError` where the `bind()` call runs: at import for a module-level bind, at call time for one inside a function. Single-fragment and empty combinations always exist now, so only tuple binds can land there, in three ways. A tuple written after the last regeneration: regenerating resolves it. A call the overload form admits without a combination behind it: one form covers each arity a slot has tuples for, fixing the length and the fragments a position may hold — every fragment any tuple of that arity wrote for the slot — but not which of them goes where, so a repeat like `(f, f)` or a pair drawn from two different tuples of one slot type-checks — as does a call mixing a single fragment in one slot with a literal tuple in another, which meets the two slots' separate forms while no combination pairs them. (A shorter tuple standing in for a longer one is not among them: the arity is fixed.) Each of those needs its combination written literally, if it is one at all. And a call the scan cannot read at all, whose template is an expression rather than a name: that one is written and still never generated, so the message points at `ignored_binds.json` instead, where the call is recorded with its reason.

Passing the same fragment to one slot twice is rejected at generation instead, naming the call site: a slot spreads each of its fragments once, so `bind(slot=(f, f))` asks for a combination that cannot exist.

`debug_path` writes the directories the walk did not descend into to `skipped_dirs.json` — the path and the reason — for the same reason `ignored_binds.json` exists: a tree left out on purpose and a tree the scan lost look the same, as statements missing from the package. The contents of pruned directories are not read, so the report has neither a file count nor what lies in them.

Every `.bind()` the scan leaves alone is recorded with the reason, and `debug_path` writes them to `ignored_binds.json` alongside the other debug artifacts. Most entries there are now ordinary: a call whose slots take single values needs nothing from the scan, and a third-party `.bind()` is none of the generator's business. The one worth reading the file for is a *tuple* bind written on an expression no scan can follow — `TEMPLATES["q"].bind(slot=(a, b))`, where the value under the key is a runtime question: its combination is never generated, so the call raises `LookupError`, and the reason recorded here is what tells that apart from a bind the generator lost.

## Customization Hooks
- **Naming conventions.** Supply `to_camel_fn_full_name` (a module:path string) and a `to_snake_fn` callable. These functions align the casing with your own `alias_generator`.
- **Endpoint configuration.** The generator writes `base_url_import` verbatim into the generated module. Set it to a global string, a configuration object, or a helper that returns the GraphQL endpoint.
- **Subscription protocol.** The generator writes `subscription_protocol` into the generated module, where it creates the module's client.
- **Model base.** `model_base_import` (a `module:attr` string, `"pydantic:BaseModel"` by default) names the Pydantic model that every model of the module derives from: `GQLModel`, the base of the generated models, inherits it. The generated module passes the same base to its client as `model_base`, and the client builds its own models of the protocol on it — the response body, a subscription's `next` payload, a graphql-transport-ws message and its `error` payload. A setting of your base therefore holds for everything the module parses. For example, `hide_input_in_errors=True` of your base holds for the client's own `Malformed response body` and `Malformed next payload` errors too:

  ```python
  # myapp/models.py
  import pydantic


  class Model(pydantic.BaseModel):
      model_config = pydantic.ConfigDict(hide_input_in_errors=True)
  ```

  ```python
  render_gql_module(..., model_base_import="myapp.models:Model")
  ```

  The `model_config` keys that the library sets win over the same keys of your base: the keys of `GQLModel` in the generated module, `extra="ignore"` of a model that validates inside a slot or a fragment subtree, and the keys of the protocol models — no alias generator and `extra="ignore"`. Every other key of your base applies.

## Runtime Highlights
- `AsyncGQLClient` accepts an ASGI `target_app`. You can use the same runtime for production HTTP calls and for in-process ASGI execution.
- `GQLClient.subscribe` returns a plain context manager over a blocking generator, so a synchronous consumer reads subscription messages with `for`.
- Both clients take `subscription_protocol` (required) and `subscription_idle_timeout`; see [Subscriptions](#subscriptions).
- Both clients take `model_base`, the base of the models that the client parses the protocol with; `pydantic.BaseModel` when you pass none. A generated module passes its own; see **Model base** in [Customization Hooks](#customization-hooks).
- `GQLOperation.with_headers` clones the operation object. The original does not change, so each call can have its own headers.
- `Upload` scalars map to `iron_gql.FileVar`. When variables contain `FileVar` instances, the client automatically sends a multipart upload (see the [GraphQL multipart request spec](https://github.com/jaydenseric/graphql-multipart-request-spec)).
- Variables are converted to JSON-compatible structures: dicts, lists, and nested input models are walked, and values of other types, such as custom scalars, are dumped with the Pydantic `TypeAdapter`.
- Pydantic secrets (`SecretStr`, `SecretBytes`, `Secret[T]`) go to the server as their values at any depth of the variables: a variable, a field of an input model, a list element, a mapping value. In `repr` and in the dumps of your models they stay masked. Map a secret scalar to one of them, e.g. `"Secret": "pydantic:SecretStr"`; for `Secret[T]`, give the parameterized type a name in your module and map the scalar to that name.

## Query Errors

A query or a mutation is sent with `Accept: application/graphql-response+json, application/json;q=0.9`, as [GraphQL over HTTP](https://graphql.github.io/graphql-over-http/draft/) asks of a client that does not know which of the two media types its server speaks. An `Accept` that you set in the headers of the client or of the call is sent as it is; the call's wins over the client's. The client reads the response as follows:

| The server answers | The client |
|---|---|
| 2xx with `data` | returns the result |
| 2xx with `errors`, with `data` or without | raises `GraphQLResponseError` with the errors |
| 2xx whose body is not a GraphQL response: not JSON, or JSON of another shape | raises `GraphQLResponseError`: "Malformed response body" |
| 3xx | raises `httpx2.HTTPStatusError` |
| 4xx or 5xx of the type `application/graphql-response+json`, with `errors` | raises `GraphQLResponseError` with the errors |
| any other 4xx or 5xx | raises `httpx2.HTTPStatusError`; `error.response` holds the status and the body |

The type is compared without its parameters, such as `charset`, and regardless of case. `GraphQLResponseError` means that the server explained the failure in GraphQL terms: a server of the current edition answers a request error with a 4xx or 5xx status and a GraphQL body, and a server of the earlier edition answers it with 200 and `application/json`. It also means that a response came with a 2xx status, of any type, but its body could not be read as a GraphQL response. `httpx2.HTTPStatusError` means that no GraphQL explanation came: a proxy or a gateway answered, the body has another type, or a GraphQL body carries no `errors`.

## Subscriptions

A generated subscription returns a context manager over a stream of typed results. Read the stream with `async for` (or `for` in a sync module). Leave the `with` block to close the connection, also before the stream ends.

```python
async with watch_events.execute(channel="news") as stream:
    async for event in stream:
        print(event.events.message)
```

Name the protocol when you create a client: `subscription_protocol="graphql-sse"` or `subscription_protocol="graphql-transport-ws"`. The protocols are not interchangeable: use the one that your server speaks.

With `"graphql-sse"`, each subscription is one `POST` request with `Accept: text/event-stream`, exactly: the reference graphql-sse server opens a stream for that value only. An `Accept` that you set is sent as it is, as for a query. The stream and the errors work as follows:

| The server | The client |
|---|---|
| sends `next` with data | yields a result |
| sends `next` with errors | raises `GraphQLResponseError` |
| sends `complete` | ends the stream |
| ends the response, or the connection breaks, before `complete` | raises `SubscriptionInterruptedError` |
| sends nothing for `subscription_idle_timeout` seconds | raises `SubscriptionInterruptedError` |
| refuses with a status that is not 2xx | raises as for a query: `GraphQLResponseError` for a 4xx or 5xx GraphQL response with `errors`, `httpx2.HTTPStatusError` otherwise |
| answers with a 2xx response that is not a stream | raises `GraphQLResponseError`: with the errors of the body, "Malformed response body" for a body that is not a GraphQL response, or, for one without errors, that no event stream came |

`SubscriptionInterruptedError` tells you that the server did not confirm the end of the operation. The client does not open the subscription again. Open it again yourself if your server lets a new subscription continue from its current state.

`subscription_idle_timeout` is the longest wait for the response headers and for the next data on the stream. The default `None` waits without a limit. A server that sends heartbeat comments keeps a healthy stream within the limit: set the timeout above the heartbeat interval of your server.

## Client Lifetime

A generated module creates its client at import and keeps it for the life of the process. Nobody has to close it: the client owns its connections.

- `GQLClient` keeps one connection pool. It closes the pool when the client is garbage-collected or when the interpreter exits.
- `AsyncGQLClient` opens separate connections in each event loop that it runs in, because an `httpx2` connection belongs to the loop that opened it. The first query or subscription in a loop opens them. They close when the loop shuts down its async generators, as `asyncio.run` and `asyncio.Runner` do on exit. One client therefore serves consecutive `asyncio.run` calls and loops in several threads at once. All loops of a client share its cookies, subscriptions included.

`close()` is final: after it, every query and subscription of the client raises `RuntimeError`, in any loop. Calling it again is safe. `AsyncGQLClient.close()` closes the connections of the running loop before it returns, also when it is called again or concurrently in that loop. The connections of other loops close when those loops end, or when `close()` is called in them.

Leave the `async with` block of a subscription before its loop ends.

Known limits:
- A loop closed without `shutdown_asyncgens`, such as `loop.close()` after `run_until_complete`, leaves the connections of `AsyncGQLClient` in it to the garbage collector, which emits `ResourceWarning`. The client still works in the next loop.
- `AsyncGQLClient` keeps the connections of a loop that was stopped but not closed, so such loops add up.
- After `os.fork()`, the child process inherits the open connections of the parent's clients.

## Example

The [`example/`](example/) directory reads as chapters. Each one adds what the one before it did not need:

| Chapter | What it adds |
|---|---|
| [`ch01_queries.py`](example/ch01_queries.py) | a query, a fragment, nested fields, `with_headers` |
| [`ch02_mutations.py`](example/ch02_mutations.py) | a generated input model and an enum |
| [`ch03_polymorphism.py`](example/ch03_polymorphism.py) | a union and an interface, matched by generated class |
| [`ch04_variables.py`](example/ch04_variables.py) | `@include`/`@skip`, and a `@oneOf` input |
| [`ch05_scalars.py`](example/ch05_scalars.py) | a `DateTime` that arrives as `datetime`, an `Upload` sent as a `FileVar` |
| [`ch06_slots.py`](example/ch06_slots.py) | a template, `bind`, `read`, and fragment variables through `with_args` |
| [`ch07_subscriptions.py`](example/ch07_subscriptions.py) | a subscription stream |
| [`ch08_sync.py`](example/ch08_sync.py) | the basics again, against a synchronous module |

A real project picks one mode; the example shows both, and [`generate.py`](example/generate.py) writes both modules from the one schema.

The chapters are written to be read and type-checked, not run: they point at a server that this repository does not ship. [`test_example.py`](example/test_example.py) is the exception. It runs chapters against [`fake_app.py`](example/fake_app.py) with the helpers from [Testing](#testing).

## Testing

`iron_gql.testing` holds the helpers that a service needs to test its own use of a generated module.

Which transport a test needs follows from the client:

| client | queries and mutations | `graphql-transport-ws` subscriptions | `graphql-sse` subscriptions |
|---|---|---|---|
| `AsyncGQLClient` | `target_app`, no socket | `target_app`, no socket | server in the test's event loop |
| `GQLClient` | server on a loopback port | server on a loopback port | server on a loopback port |

`AsyncGQLClient` takes an ASGI `target_app` and calls it in process, websockets included. An SSE subscription is the exception: the ASGI transport of `httpx2` holds the response body until the app returns, so `subscribe` refuses a `target_app` client with `ValueError`. Serve the app with `async_live_asgi_server` instead. `GQLClient` has no in-process transport: the ASGI transport of `httpx2` is async-only, and WSGI cannot carry websockets. A synchronous module is therefore tested against a real server, which `live_asgi_server` starts for you.

Only `iron_gql.testing.server` has a dependency of its own — `uvicorn`, from `pip install iron-gql[testing]`. Everything else here needs nothing beyond the runtime, so a project that only replaces clients installs plain `iron-gql`.

You supply the fake yourself. The library takes no position on how you answer a query: run your own schema library, return canned JSON, or serve the app you are testing.

### Replace the client

`use_client` binds your client into a generated module. Use it with `async with` for `AsyncGQLClient` and `with` for `GQLClient`. On exit it restores the previous client and closes the client that you passed in, so give each `use_client` a client instance of its own. A client that you create without `model_base` takes the module's base, so it parses responses the way the module's own client does; a client created with another base raises `ValueError`, and that client stays yours to close:

```python
from iron_gql.runtime import AsyncGQLClient
from iron_gql.testing import use_client
from myapp.gql import api

async def test_get_user():
    client = AsyncGQLClient(
        base_url="http://testserver",
        subscription_protocol="graphql-sse",
        target_app=my_asgi_app,
    )
    async with use_client(api, client):
        result = await get_user.execute(id="1")
        assert result.user.name == "Alice"
```

The swap holds for the whole process: a request made from a worker thread while the swap is active goes through your client too. Swaps of one module nest: an inner `use_client` overrides the outer one until it exits. A swap that exits while a swap of the same module entered after it is still active raises `RuntimeError` in the test that exits it. After that error, `use_client` restores the module's client as far as it can, but makes no guarantees.

Two limits apply. Requests of scenarios that run at the same time are not isolated from each other, since they all share the module's client. Entering and exiting swaps of one module from several threads at the same time is not supported.

### Serve an app on a loopback port

`live_asgi_server` serves an ASGI app with `uvicorn` on a port that the operating system picks, and it yields the URL of that server:

```python
from iron_gql.runtime import GQLClient
from iron_gql.testing import use_client
from iron_gql.testing.server import live_asgi_server

def test_get_user_sync():
    with (
        live_asgi_server(my_asgi_app) as base_url,
        use_client(
            api, GQLClient(base_url=base_url, subscription_protocol="graphql-sse")
        ),
    ):
        result = get_user.execute(id="1")
        assert result.user.name == "Alice"
```

The URL ends with `/graphql`. Pass `path="/other"` for a different one. The path is only a part of the URL: a fake app usually answers on every path, so the helper does not route.

`live_asgi_server` runs the server in a thread with an event loop of its own. To serve an app that uses what the test opened in its own loop — a database pool, an event the app waits for — use `async_live_asgi_server`. It runs the server as a task of the calling event loop and takes the same arguments:

```python
from iron_gql.runtime import AsyncGQLClient
from iron_gql.testing import use_client
from iron_gql.testing.server import async_live_asgi_server

async def test_watch_events():
    async with async_live_asgi_server(my_asgi_app) as base_url:
        client = AsyncGQLClient(base_url=base_url, subscription_protocol="graphql-sse")
        async with use_client(api, client), watch_events.execute(channel="news") as stream:
            assert [event.events.message async for event in stream] == ["hello"]
```

Both servers run the app without its lifespan: start what the app needs in the test.

### Script a subscription fake

A `graphql-transport-ws` fake has to hold up the server end of the protocol. `accept_graphql_ws` performs the handshake and hands you the connection; each step then asserts that the client kept to the protocol, and reports what arrived instead when it did not.

```python
from iron_gql.runtime import ASGIReceive
from iron_gql.runtime import ASGIScope
from iron_gql.runtime import ASGISend
from iron_gql.testing import accept_graphql_ws

async def events_app(scope: ASGIScope, receive: ASGIReceive, send: ASGISend) -> None:
    connection = await accept_graphql_ws(scope, receive, send)
    subscription = await connection.ack()
    assert subscription.payload["variables"] == {"channel": "test"}

    await subscription.next({"events": {"id": "1", "message": "hello"}})
    await subscription.complete()
    await connection.drain()
```

`accept_graphql_ws` accepts the socket, echoes the subprotocol when the client offered it, and consumes `connection_init`. On the connection you then call:

- `ack()` — sends `connection_ack` and waits for the client's `subscribe`. The subscription it returns exposes the received message as `payload` and stamps its `id` on everything it sends.
- `send_message(message)` and `expect_pong()` — for `ping` and other connection-level traffic.
- `close(code, reason)` — closes the socket, for testing how the client reacts.
- `drain()` — waits for the client to hang up. Return before that and you tear the connection down under it.

On the subscription: `next(data)`, `error(errors)`, `complete()`, and `send_message(message)` for anything else.

Control flow stays in your fake, so state lives in ordinary Python around these calls — a connection counter, a drop on the N-th connect, a failure of the first few messages.

A `graphql-sse` fake needs no helper: it answers the `POST` with `content-type: text/event-stream` and writes the events, for example `event: next\ndata: {"data": {...}}\n\n`, then `event: complete\ndata: \n\n`.

## Validation and Troubleshooting
- Every input the generator rejects raises `GraphQLGenerationError`, a `ValueError` whose `errors` holds one diagnosis per problem: a malformed gql function or `.bind()` call, a source file that does not parse, a missing schema file, a schema that does not parse or that graphql-core finds invalid, and a statement that does not parse or validate.
- A diagnosis names its place: the file and the line of the statement or call, or the schema file with the line and column. A statement's syntax error also gives the line and column inside the statement, counted after its common indentation and its leading and trailing whitespace are removed.
- A rejection caused by another exception keeps it as `__cause__`: the `SyntaxError` of a source file, including one a bind resolves through, the `GraphQLSyntaxError` of the schema or of a statement, and the `FileNotFoundError` of a missing schema file. Syntax errors of several statements come together in one error, its cause an `ExceptionGroup` of them, and are reported before anything else about the statements: a statement that does not parse would turn every use of its fragments into an error of its own.
- Other failures keep their own type: a file that cannot be read for another reason, such as permissions or an encoding that is not UTF-8, and writing the generated module. Around generation, `except TypeError`, `except SyntaxError`, `except GraphQLSyntaxError` and `except FileNotFoundError` no longer catch a rejected input; `except ValueError` still does.
- Operations that share a name must have identical bodies. To remove the conflict, rename the operations or merge them.
- The generated helper raises `LookupError` for a statement that it does not know. After you add or edit a statement, regenerate the module.
