rath

Rath is a transport-agnostic GraphQL client for Python focused on composability. Built on Pydantic v2 and Koil, it works seamlessly in both async and sync code.

Inspired by Apollo Client, Rath composes request logic out of links — small, chainable middleware objects that transform or forward an operation until a terminating link dispatches it over the network. Need auth headers, retries, schema validation, persisted queries, or a websocket transport for subscriptions? Each is just another link you drop into the chain.

Installation

pip install rath

Or with uv:

uv add rath

Quickstart

from rath import Rath
from rath.links.aiohttp import AIOHttpLink

link = AIOHttpLink(endpoint_url="https://countries.trevorblades.com/")

with Rath(link=link) as rath:
    result = rath.query(
        """
        query {
            countries {
                native
                capital
            }
        }
        """
    )
    print(result.data)  # GraphQLResult — the payload lives on .data

Features

• Pluggable transports — aiohttp, httpx and websockets out of the box.
• Composable links — build your request pipeline from small, reusable pieces.
• Async-first, sync-friendly — call aquery/asubscribe from async code, or query/subscribe from sync code (powered by Koil).
• Auth with automatic refresh — re-fetch the token and retry on a 401/403.
• Schema validation — validate operations locally before they hit the wire.
• Automatic Persisted Queries (APQ) — send a hash instead of the full document.
• Typed operations — pairs beautifully with turms generated Pydantic models.
• First-class testing — mock terminating links so unit tests never touch the network.

Table of contents

• Installation
• The link concept
• Quickstart
• Async usage
• Variables and headers
• Handling errors
• Authentication with refresh
• Subscriptions and transport splitting
• Retries
• Timeouts
• Schema validation
• Automatic Persisted Queries
• Testing without a server
• Typed operations with turms
• Included links

The link concept

A Rath client is configured with a single terminating link — the link that actually sends the operation over the network (e.g. AIOHttpLink, HttpxLink, GraphQLWSLink).

To add behaviour, you wrap the terminating link with continuation links using compose(). Operations flow left → right through the chain; results flow back right → left:

operation ─▶ AuthTokenLink ─▶ RetryLink ─▶ AIOHttpLink ─▶ server
result    ◀────────────────────────────────────────────◀

from rath import Rath
from rath.links import compose
from rath.links.auth import ComposedAuthLink
from rath.links.retry import RetryLink
from rath.links.aiohttp import AIOHttpLink

link = compose(
    ComposedAuthLink(token_loader=...),   # add the Bearer header
    RetryLink(maximum_retry_attempts=3),  # retry on failure
    AIOHttpLink(endpoint_url="..."),      # terminating link
)

rath = Rath(link=link)

Rath(link=...) also accepts a plain list — it is composed for you, and the last element must be a terminating link:

rath = Rath(link=[auth, retry, AIOHttpLink(endpoint_url="...")])

Async usage

Rath is built for async; the sync API above is just a thin wrapper. In async code use aquery, aquery_operation and asubscribe:

import asyncio
from rath import Rath
from rath.links.aiohttp import AIOHttpLink

link = AIOHttpLink(endpoint_url="https://countries.trevorblades.com/")


async def main():
    async with Rath(link=link) as rath:
        result = await rath.aquery(
            """
            query {
                countries {
                    native
                    capital
                }
            }
            """
        )
        print(result.data)


asyncio.run(main())

Variables and headers

Pass variables and per-operation headers directly to query/aquery:

with Rath(link=link) as rath:
    result = rath.query(
        """
        query Country($code: ID!) {
            country(code: $code) {
                name
                capital
            }
        }
        """,
        variables={"code": "DE"},
        headers={"X-Tenant": "acme"},
    )
    print(result.data["country"])

Handling errors

When the server returns GraphQL errors, Rath raises a GraphQLException carrying the messages and the offending operation:

from rath.operation import GraphQLException

with Rath(link=link) as rath:
    try:
        result = rath.query("query { doesNotExist }")
    except GraphQLException as exc:
        print("GraphQL error:", exc.message)
        print("Operation:", exc.operation.operation_name)

Authentication with refresh

ComposedAuthLink injects a Bearer token (returned by token_loader) into every request. If the terminating link reports an authentication error (e.g. a 403), the token_refresher is called and the operation is retried — up to maximum_refresh_attempts times.

from rath import Rath
from rath.links import compose
from rath.links.auth import ComposedAuthLink
from rath.links.aiohttp import AIOHttpLink


async def aload_token() -> str:
    return "SERVER_TOKEN"


async def arefresh_token() -> str:
    # e.g. exchange a refresh token for a fresh access token
    return "REFRESHED_TOKEN"


auth = ComposedAuthLink(token_loader=aload_token, token_refresher=arefresh_token)
http = AIOHttpLink(endpoint_url="https://countries.trevorblades.com/")

with Rath(link=compose(auth, http)) as rath:
    result = rath.query("query { countries { capital } }")
    print(result.data)

Subscriptions and transport splitting

Different operation types often want different transports: websockets for subscriptions, plain HTTP for queries and mutations (so they stay cacheable). split routes each operation to one of two links based on a predicate.

from graphql import OperationType
from rath import Rath
from rath.links import compose, split
from rath.links.auth import ComposedAuthLink
from rath.links.aiohttp import AIOHttpLink
from rath.links.graphql_ws import GraphQLWSLink


async def aload_token() -> str:
    return "SERVER_TOKEN"


auth = ComposedAuthLink(token_loader=aload_token)
http = AIOHttpLink(endpoint_url="https://api.example.com/graphql")
ws = GraphQLWSLink(ws_endpoint_url="wss://api.example.com/graphql")

# Predicate True  -> first link (http)
# Predicate False -> second link (ws)
transport = split(http, ws, lambda op: op.node.operation != OperationType.SUBSCRIPTION)

with Rath(link=compose(auth, transport)) as rath:
    # queries and mutations go over HTTP
    print(rath.query("query { countries { capital } }").data)

    # subscriptions go over the websocket link
    for event in rath.subscribe("subscription { newCountry { capital } }"):
        print(event.data)

Note: op.node.operation is a graphql-core OperationType enum — compare against OperationType.SUBSCRIPTION, not the string "subscription".

Retries

RetryLink retries an operation when the terminating link fails (handy for flaky connections and dropped subscriptions):

from rath.links import compose
from rath.links.retry import RetryLink
from rath.links.aiohttp import AIOHttpLink

link = compose(
    RetryLink(maximum_retry_attempts=5, sleep_interval=1),  # wait 1s between attempts
    AIOHttpLink(endpoint_url="https://countries.trevorblades.com/"),
)

Timeouts

TimeoutLink aborts an operation that takes longer than timeout seconds:

from rath.links import compose
from rath.links.timeout import TimeoutLink
from rath.links.aiohttp import AIOHttpLink

link = compose(
    TimeoutLink(timeout=10),  # seconds
    AIOHttpLink(endpoint_url="https://countries.trevorblades.com/"),
)

Schema validation

ValidatingLink validates each operation against a GraphQL schema before it is sent, so malformed queries fail fast with a clear error. Provide the schema as a DSL string, a glob of .graphql files, or let it introspect the remote schema.

from rath.links import compose
from rath.links.validate import ValidatingLink
from rath.links.aiohttp import AIOHttpLink

link = compose(
    ValidatingLink(schema_glob="schemas/**/*.graphql"),
    AIOHttpLink(endpoint_url="https://countries.trevorblades.com/"),
)

Automatic Persisted Queries (APQ)

ApqLink sends a SHA-256 hash of the query instead of the full document. If the server hasn't seen the query yet it transparently retries with the full text. This shrinks request payloads and plays well with CDN caching.

from rath.links import compose
from rath.links.apq import ApqLink
from rath.links.aiohttp import AIOHttpLink

link = compose(
    ApqLink(),
    AIOHttpLink(endpoint_url="https://countries.trevorblades.com/"),
)

Testing without a server

The rath.links.testing package provides terminating links that resolve operations locally, so unit tests never hit the network. AsyncMockLink takes a dict of resolvers keyed by field name; each resolver receives the Operation and returns the value for that field:

from rath import Rath
from rath.links.testing.mock import AsyncMockLink
from rath.operation import Operation


async def resolve_countries(operation: Operation):
    return [{"capital": "Berlin", "native": "Deutschland"}]


link = AsyncMockLink(query_resolver={"countries": resolve_countries})

with Rath(link=link) as rath:
    result = rath.query("query { countries { native capital } }")
    assert result.data == {"countries": [{"capital": "Berlin", "native": "Deutschland"}]}

Use AsyncStatefulMockLink when you also need to mock subscriptions, and AssertLink together with compose() to assert that upstream links transformed an operation as expected.

Typed operations with turms

Looking for fully typed operations for your GraphQL API? Rath pairs with turms, a code generator that turns your .graphql documents into typed, Pydantic-powered functions:

import asyncio
from examples.api.schema import aget_capsules  # turms-generated
from rath import Rath
from rath.links import compose
from rath.links.auth import ComposedAuthLink
from rath.links.aiohttp import AIOHttpLink


async def aload_token() -> str:
    return ""


rath = Rath(
    link=compose(
        ComposedAuthLink(token_loader=aload_token),
        AIOHttpLink(endpoint_url="https://api.spacex.land/graphql/"),
    ),
)


async def main():
    # Entering the client registers it as the current Rath, so turms-generated
    # functions can find it without passing the client explicitly.
    async with rath:
        capsules = await aget_capsules()  # fully typed result
        print(capsules)


asyncio.run(main())

This repository contains an example turms-generated client (examples/) querying the public SpaceX API, plus a sample of the generated code.

Included links

License

See LICENSE.