datalayer-graphql

Async Python client for a GraphQL supergraph running on Apollo Router. It provides one connection layer for authentication, Apollo client headers, and operation execution.

Built on gql with an httpx transport. It adds bearer-token authentication, Apollo headers, and federation-aware error handling.

Features

• Async SupergraphClient built on gql + httpx
• Typed SupergraphConfig via Pydantic
• Bearer token auth from environment variables
• Standard Apollo client headers (apollographql-client-name, apollographql-client-version)
• Operation registry — call execute(operation_name=...) without passing the query each time
• .graphql file loading via load_operations()
• Structured exception hierarchy with federation error support
• Response headers captured for router inspection

Requirements

• Python 3.13+
• uv (recommended) or pip
• Network access to your GraphQL endpoint
• A valid bearer token accepted by that endpoint

Installation

From the monorepo root:

uv sync --package datalayer-graphql

As a workspace dependency (already wired in datalayer-mcp):

dependencies = ["datalayer-graphql"]

[tool.uv.sources]
datalayer-graphql = { workspace = true }

Quick start

import asyncio

from datalayer_graphql import BearerTokenAuth, SupergraphClient, SupergraphConfig


async def main() -> None:
    client = SupergraphClient(
        SupergraphConfig(
            url="https://graphql.example.com",
            auth=BearerTokenAuth(token_env="SUPERGRAPH_TOKEN"),
            client_name="my-graphql-client",
            operations={
                "Cves": """
                    query Cves($first: Int!) {
                      cves(first: $first) {
                        totalCount
                        edges { node { title url } }
                      }
                    }
                """,
            },
        )
    )

    async with client:
        result = await client.execute(
            operation_name="Cves",
            variables={"first": 10},
        )
        print(result.data)


asyncio.run(main())

Set the token before running:

export SUPERGRAPH_TOKEN="your-bearer-token"

See examples/graphql/ for more complete examples including operations registries, variable passing, and .graphql file loading.

Loading operations from .graphql files

from pathlib import Path
from datalayer_graphql import load_operations, SupergraphConfig, BearerTokenAuth

operations = load_operations(Path("operations/"))

config = SupergraphConfig(
    url="https://graphql.example.com",
    auth=BearerTokenAuth(token_env="SUPERGRAPH_TOKEN"),
    operations=operations,
)

Each .graphql file must contain exactly one named operation. The operation name becomes the key in the returned dict.

Configuration

SupergraphConfig

Field	Type	Default	Description
url	HttpUrl	—	Supergraph endpoint URL
auth	BearerTokenAuth	—	Authentication configuration
client_name	str	Package default	Value for apollographql-client-name header
client_version	str	"latest"	Value for apollographql-client-version header
timeout	float	30.0	HTTP request timeout in seconds
verify_ssl	bool	True	TLS certificate verification. Keep enabled unless your endpoint requires a custom trust setup
operations	dict[str, str]	{}	Maps operation name -> GraphQL query string
extra_headers	dict[str, str]	{}	Additional HTTP headers merged into every request

BearerTokenAuth

Reads a bearer token from an environment variable at request time (not stored in config):

BearerTokenAuth(token_env="SUPERGRAPH_TOKEN")

• Looks up os.environ["SUPERGRAPH_TOKEN"]
• Strips whitespace and a leading Bearer prefix if present
• Sets Authorization: Bearer <token> on every HTTP request via httpx auth hooks

Operation registry

GraphQL POST bodies require a query string. To support an API that only passes operation_name, register queries on config:

SupergraphConfig(
    url="...",
    auth=BearerTokenAuth(token_env="SUPERGRAPH_TOKEN"),
    operations={
        "GetCustomerById": "query GetCustomerById($id: ID!) { customer(id: $id) { id name } }",
        "Cves": "query Cves($first: Int!) { cves(first: $first) { totalCount } }",
    },
)

You can also pass query= directly to execute() to bypass the registry:

await client.execute(
    operation_name="AdHoc",
    query="query AdHoc { __typename }",
)

API reference

SupergraphClient

client = SupergraphClient(config: SupergraphConfig)

Creates a gql.Client with an HTTPXAsyncTransport, configured with default headers, auth, and connection pooling.

execute

result = await client.execute(
    operation_name: str,
    variables: dict | None = None,
    query: str | None = None,
) -> ExecuteResult

Sends a GraphQL POST request via the gql transport.

Returns: ExecuteResult(data=..., extensions=..., response_headers=...)

Context manager

Always close the client when done (or use async with):

async with SupergraphClient(config) as client:
    result = await client.execute(operation_name="Cves", variables={"first": 10})

ExecuteResult

@dataclass(frozen=True)
class ExecuteResult:
    data: dict[str, Any] | None
    extensions: dict[str, Any] | None = None
    response_headers: Mapping[str, str] | None = None

load_operations

from pathlib import Path

operations: dict[str, str] = load_operations(Path("operations/"))

Scans a directory for .graphql files. Each file must contain exactly one named operation. Returns {operation_name: query_string}.

Error handling

Exception hierarchy

SupergraphError                         # Base for all supergraph errors
├── SupergraphConnectionError           # Network / connection failures
├── SupergraphHTTPError                 # HTTP 4xx/5xx (status_code attribute)
└── GraphQLError                        # GraphQL-level errors in response body
    ├── errors: list[GraphQLErrorDetail] # Parsed error details
    ├── data: dict | None                # Partial data (federation)
    └── extensions: dict | None

GraphQLErrorDetail

@dataclass(frozen=True)
class GraphQLErrorDetail:
    message: str
    path: list[str | int] | None
    locations: list[dict[str, int]] | None
    extensions: dict[str, Any] | None

    @property
    def code(self) -> str | None: ...  # e.g. "SUBREQUEST_HTTP_ERROR"
    @property
    def service(self) -> str | None: ...  # e.g. "inventory-service"

Error mapping

Situation	Exception
Missing env var for token	ValueError from BearerTokenAuth.resolve_token()
Missing query for operation	ValueError from execute()
HTTP 401/403/503	SupergraphHTTPError (.status_code attribute)
GraphQL errors in body	GraphQLError with parsed GraphQLErrorDetail list
Network / connection failure	SupergraphConnectionError
Malformed response	SupergraphError
Timeout	TimeoutError (standard Python)

Federation error example:

try:
    result = await client.execute(operation_name="GetDocs")
except GraphQLError as e:
    for detail in e.errors:
        print(detail.code)  # "SUBREQUEST_HTTP_ERROR"
        print(detail.service)  # "inventory-service"
    if e.data:
        print("Partial data:", e.data)

Package layout

src/datalayer_graphql/
├── __init__.py        # Public exports
├── auth/
│   └── bearer.py      # BearerTokenAuth + httpx auth flow
├── client/
│   └── supergraph.py  # SupergraphClient (wraps gql.Client)
├── config/
│   └── supergraph.py  # SupergraphConfig + header builder
├── exceptions.py      # Exception hierarchy
├── models.py          # ExecuteResult
└── operations.py      # load_operations()

Tests

Unit tests

uv run pytest packages/datalayer-graphql/tests/test_client.py tests/test_operations.py -v

Integration smoke tests

Live smoke tests use the endpoint configured in .env. They are skipped automatically when .env is missing or incomplete.

Setup

cd packages/datalayer-graphql
cp .env.example .env

Edit .env:

SUPERGRAPH_URL=https://graphql.example.com
SUPERGRAPH_TOKEN=your-bearer-token-here
SUPERGRAPH_CLIENT_NAME=my-graphql-client
SUPERGRAPH_VERIFY_SSL=true
SUPERGRAPH_SMOKE_OPERATION=Cves
SUPERGRAPH_SMOKE_VARIABLES={"first": 10}

Run

uv run pytest packages/datalayer-graphql/tests/test_smoke_supergraph.py -v -s

Obtaining a token

Use the included token script to fetch a client-credentials token from your OAuth 2.0 provider and write it to your .env file:

uv run scripts/fetch_token.py \
  --token-url https://auth.example.com/oauth2/token \
  --client-id my-client-id \
  --client-secret YOUR_SECRET \
  --scope my-api-scope \
  --env-file packages/datalayer-graphql/.env

Or fetch manually with curl:

curl -X POST \
  "https://auth.example.com/oauth2/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=my-client-id" \
  -d "client_secret=YOUR_SECRET" \
  -d "scope=my-api-scope"

Copy access_token into SUPERGRAPH_TOKEN in .env.

Development

Install dev dependencies

uv sync --all-packages

Lint and type check

uv run ruff check .
uv run mypy packages/datalayer-graphql/src/

Roadmap

• ClientCredentialsAuth for automatic token fetching and refresh
• Router response header validation
• Federation-aware error types (GraphQLErrorDetail with code, service)
• Unit tests with mock transport
• .graphql file loading utility

License

See LICENSE.txt.