Metadata-Version: 2.5
Name: redhat-datalayer-graphql
Version: 0.1.1
Summary: Async Python client for Red Hat's GraphQL supergraph (Apollo Router)
Author: Red Hat DataLayer Team
Maintainer-email: Mayur Deshmukh <mdeshmuk@redhat.com>, Pranav Advani <padvani@redhat.com>
License-Expression: Apache-2.0
Requires-Python: >=3.13
Requires-Dist: gql[httpx]>=3.5.0
Requires-Dist: graphql-core>=3.2.0
Requires-Dist: httpx>=0.28.1
Requires-Dist: pydantic>=2.13.4
Description-Content-Type: text/markdown

# 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](https://gql.readthedocs.io/) with an [httpx](https://www.python-httpx.org/) transport. It adds bearer-token authentication, Apollo headers, and federation-aware error handling.

## Features

- Async `SupergraphClient` built on `gql` + `httpx`
- Typed `SupergraphConfig` via [Pydantic](https://docs.pydantic.dev/)
- 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](https://docs.astral.sh/uv/) (recommended) or pip
- Network access to your GraphQL endpoint
- A valid bearer token accepted by that endpoint

## Installation

From the monorepo root:

```bash
uv sync --package datalayer-graphql
```

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

```toml
dependencies = ["datalayer-graphql"]

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

## Quick start

```python
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:

```bash
export SUPERGRAPH_TOKEN="your-bearer-token"
```

See [`examples/graphql/`](../../examples/graphql/) for more complete examples including
operations registries, variable passing, and `.graphql` file loading.

### Loading operations from `.graphql` files

```python
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):

```python
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:

```python
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:

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

## API reference

### `SupergraphClient`

```python
client = SupergraphClient(config: SupergraphConfig)
```

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

#### `execute`

```python
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`):

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

### `ExecuteResult`

```python
@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`

```python
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`

```python
@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:

```python
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

```bash
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

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

Edit `.env`:

```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

```bash
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:

```bash
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:

```bash
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

```bash
uv sync --all-packages
```

### Lint and type check

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

### Roadmap

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

## License

See [`LICENSE.txt`](../../LICENSE.txt).
