Part 4: GraphQL Client Development — TypeScript and Python

Consuming a GraphQL API from Two Worlds

Part 3 built the server. Now we write the clients.

This part covers two contexts I work with regularly:

1. TypeScript client — a Node.js service (or frontend) that queries the product service using Apollo Client and type-safe code generation

2. Python microservice client — a Python FastAPI service that calls the TypeScript GraphQL API as part of a larger microservices topology

Both scenarios are real patterns from my own projects. The Python service is an analytics or reporting service that needs product data but is not itself a GraphQL server (yet — federation in Part 6).

TypeScript: Apollo Client with Code Generation

Why Code Generation

Without code generation, you write queries as raw strings and get any back:

const result = await client.query({ query: gql`...` });
result.data.products.edges[0].node.name; // type: any — no IDE support, no compile errors

With GraphQL Code Generator, the query's return type is inferred exactly:

const result = await client.query<ProductsQuery, ProductsQueryVariables>({ query: ProductsDocument });
result.data?.products.edges[0].node.name; // type: string — fully typed

Setup

npm install @apollo/client graphql
npm install --save-dev @graphql-codegen/cli @graphql-codegen/typescript \
  @graphql-codegen/typescript-operations @graphql-codegen/typescript-react-apollo

Code Generator Config

# codegen.yml
overwrite: true
schema: "http://localhost:4001/graphql"   # or a local .graphql file
documents: "src/**/*.graphql"
generates:
  src/generated/graphql.ts:
    plugins:
      - typescript
      - typescript-operations
      - typescript-react-apollo
    config:
      withHooks: true
      withComponent: false
      withHOC: false

Add to package.json:

{
  "scripts": {
    "codegen": "graphql-codegen --config codegen.yml"
  }
}

Write Queries as .graphql Files

# src/queries/products.graphql
query Products(
  $filter: ProductFiltersInput
  $first: Int
  $after: String
  $sortBy: ProductSortField
  $sortOrder: SortOrder
) {
  products(
    filter: $filter
    first: $first
    after: $after
    sortBy: $sortBy
    sortOrder: $sortOrder
  ) {
    edges {
      cursor
      node {
        id
        name
        slug
        price
        currency
        inStock
        stockCount
        category {
          id
          name
        }
      }
    }
    pageInfo {
      endCursor
      hasNextPage
    }
    totalCount
  }
}

query Product($id: UUID!) {
  product(id: $id) {
    id
    name
    slug
    description
    price
    currency
    inStock
    stockCount
    category {
      id
      name
      slug
    }
    variants {
      id
      sku
      name
      price
      stockCount
    }
    createdAt
    updatedAt
  }
}

mutation CreateProduct($input: CreateProductInput!) {
  createProduct(input: $input) {
    id
    name
    slug
    price
    status
    createdAt
  }
}

mutation UpdateProduct($id: UUID!, $input: UpdateProductInput!) {
  updateProduct(id: $id, input: $input) {
    id
    name
    price
    status
    updatedAt
  }
}

subscription StockAlert($threshold: Int!) {
  stockAlert(threshold: $threshold) {
    id
    name
    stockCount
  }
}

Run npm run codegen to generate src/generated/graphql.ts with all types.

Apollo Client Instance

// src/apollo.ts
import { ApolloClient, InMemoryCache, createHttpLink, split } from '@apollo/client';
import { GraphQLWsLink } from '@apollo/client/link/subscriptions';
import { createClient } from 'graphql-ws';
import { getMainDefinition } from '@apollo/client/utilities';

const httpLink = createHttpLink({
  uri: process.env.PRODUCT_SERVICE_URL ?? 'http://localhost:4001/graphql',
  headers: {
    'Content-Type': 'application/json',
  },
});

const wsLink = new GraphQLWsLink(
  createClient({
    url: process.env.PRODUCT_SERVICE_WS_URL ?? 'ws://localhost:4001/graphql',
  }),
);

// Route subscriptions to WebSocket, queries/mutations to HTTP
const splitLink = split(
  ({ query }) => {
    const definition = getMainDefinition(query);
    return definition.kind === 'OperationDefinition' && definition.operation === 'subscription';
  },
  wsLink,
  httpLink,
);

export const client = new ApolloClient({
  link: splitLink,
  cache: new InMemoryCache({
    typePolicies: {
      ProductConnection: {
        fields: {
          edges: {
            keyArgs: ['filter', 'sortBy', 'sortOrder'],
            merge(existing = [], incoming: unknown[]) {
              return [...existing, ...incoming];
            },
          },
        },
      },
    },
  }),
});

The typePolicies section configures cursor pagination merging — when a query fetches the next page, Apollo Client appends the edges to the existing list.

Using Generated Hooks (React)

// src/components/ProductList.tsx
import { useProductsQuery } from '../generated/graphql';

function ProductList() {
  const { data, loading, error, fetchMore } = useProductsQuery({
    variables: { first: 20, sortBy: 'CREATED_AT', sortOrder: 'DESC' },
  });

  if (loading) return <div>Loading...</div>;
  if (error) return <div>Error: {error.message}</div>;

  const { edges, pageInfo, totalCount } = data!.products;

  return (
    <div>
      <p>{totalCount} products</p>
      <ul>
        {edges.map(({ node }) => (
          <li key={node.id}>
            {node.name} — {node.currency} {node.price}
            {!node.inStock && <span> (out of stock)</span>}
          </li>
        ))}
      </ul>
      {pageInfo.hasNextPage && (
        <button
          onClick={() =>
            fetchMore({
              variables: { after: pageInfo.endCursor },
            })
          }
        >
          Load more
        </button>
      )}
    </div>
  );
}

Node.js Client (Non-React)

For a microservice that uses the GraphQL API without React:

// src/service/productService.ts
import { client } from '../apollo';
import {
  ProductDocument,
  ProductQuery,
  ProductQueryVariables,
  ProductsDocument,
  ProductsQuery,
  ProductsQueryVariables,
} from '../generated/graphql';

export async function getProduct(id: string): Promise<ProductQuery['product']> {
  const { data } = await client.query<ProductQuery, ProductQueryVariables>({
    query: ProductDocument,
    variables: { id },
    fetchPolicy: 'network-only', // always fetch fresh data in a service context
  });
  return data.product ?? null;
}

export async function listProducts(
  filter?: ProductsQueryVariables['filter'],
  first = 20,
): Promise<ProductsQuery['products']['edges']> {
  const { data } = await client.query<ProductsQuery, ProductsQueryVariables>({
    query: ProductsDocument,
    variables: { filter, first },
    fetchPolicy: 'network-only',
  });
  return data.products.edges;
}

Python: gql Client

The Python gql library provides a GraphQL client that works with any GraphQL server. I use it in Python microservices that need to fetch data from a TypeScript GraphQL service.

Installation

pip install gql[aiohttp]        # async HTTP transport
pip install gql[websockets]     # WebSocket transport for subscriptions
pip install pydantic             # typed response parsing

Basic Query

# src/graphql_client.py
from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport

transport = AIOHTTPTransport(
    url="http://localhost:4001/graphql",
    headers={"Content-Type": "application/json"},
)

# Synchronous client for scripts
def get_client() -> Client:
    return Client(transport=transport, fetch_schema_from_transport=True)


PRODUCT_QUERY = gql("""
    query Product($id: UUID!) {
        product(id: $id) {
            id
            name
            slug
            price
            currency
            inStock
            stockCount
            category {
                id
                name
            }
        }
    }
""")

LIST_PRODUCTS_QUERY = gql("""
    query Products($filter: ProductFiltersInput, $first: Int, $after: String) {
        products(filter: $filter, first: $first, after: $after) {
            edges {
                cursor
                node {
                    id
                    name
                    price
                    currency
                    inStock
                    stockCount
                }
            }
            pageInfo {
                endCursor
                hasNextPage
            }
            totalCount
        }
    }
""")

Async Client with FastAPI

# src/product_gateway.py
from typing import Optional
from contextlib import asynccontextmanager
from gql import gql, Client
from gql.transport.aiohttp import AIOHTTPTransport
from pydantic import BaseModel

# ─── Response Models ─────────────────────────────────────────────────────────

class CategoryRef(BaseModel):
    id: str
    name: str


class ProductSummary(BaseModel):
    id: str
    name: str
    price: float
    currency: str
    inStock: bool
    stockCount: int


class ProductDetail(ProductSummary):
    slug: str
    description: Optional[str] = None
    category: CategoryRef


# ─── GraphQL Gateway ──────────────────────────────────────────────────────────

class ProductGraphQLGateway:
    """Wraps the TypeScript product GraphQL service."""

    def __init__(self, url: str):
        transport = AIOHTTPTransport(url=url)
        self._client = Client(transport=transport, fetch_schema_from_transport=False)

    async def get_product(self, product_id: str) -> Optional[ProductDetail]:
        query = gql("""
            query Product($id: UUID!) {
                product(id: $id) {
                    id
                    name
                    slug
                    description
                    price
                    currency
                    inStock
                    stockCount
                    category { id name }
                }
            }
        """)

        async with self._client as session:
            result = await session.execute(query, variable_values={"id": product_id})

        if result["product"] is None:
            return None

        return ProductDetail(**result["product"])

    async def list_products(
        self,
        category_id: Optional[str] = None,
        in_stock: Optional[bool] = None,
        first: int = 20,
        after: Optional[str] = None,
    ) -> tuple[list[ProductSummary], bool, Optional[str]]:
        query = gql("""
            query Products($filter: ProductFiltersInput, $first: Int, $after: String) {
                products(filter: $filter, first: $first, after: $after) {
                    edges { node { id name price currency inStock stockCount } cursor }
                    pageInfo { endCursor hasNextPage }
                    totalCount
                }
            }
        """)

        filter_input: dict = {}
        if category_id:
            filter_input["categoryId"] = category_id
        if in_stock is not None:
            filter_input["inStock"] = in_stock

        async with self._client as session:
            result = await session.execute(
                query,
                variable_values={
                    "filter": filter_input or None,
                    "first": first,
                    "after": after,
                },
            )

        products_data = result["products"]
        items = [ProductSummary(**edge["node"]) for edge in products_data["edges"]]
        has_next = products_data["pageInfo"]["hasNextPage"]
        end_cursor = products_data["pageInfo"]["endCursor"]

        return items, has_next, end_cursor

Wiring into FastAPI

# src/main.py
import os
from fastapi import FastAPI, HTTPException, Query
from typing import Optional
from .product_gateway import ProductGraphQLGateway, ProductDetail, ProductSummary

app = FastAPI(title="Analytics Service")

gateway = ProductGraphQLGateway(
    url=os.environ.get("PRODUCT_SERVICE_URL", "http://localhost:4001/graphql")
)


@app.get("/products/{product_id}", response_model=ProductDetail)
async def get_product(product_id: str):
    product = await gateway.get_product(product_id)
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    return product


@app.get("/products", response_model=list[ProductSummary])
async def list_products(
    category_id: Optional[str] = Query(None),
    in_stock: Optional[bool] = Query(None),
    page_size: int = Query(default=20, ge=1, le=100),
    after: Optional[str] = Query(None),
):
    items, has_next, end_cursor = await gateway.list_products(
        category_id=category_id,
        in_stock=in_stock,
        first=page_size,
        after=after,
    )
    return items

Python Subscriptions with WebSocket Transport

For a Python service that needs real-time updates:

# src/stock_monitor.py
import asyncio
from gql import gql, Client
from gql.transport.websockets import WebsocketsTransport

async def monitor_stock_alerts(threshold: int) -> None:
    transport = WebsocketsTransport(
        url="ws://localhost:4001/graphql",
        subprotocols=[WebsocketsTransport.GRAPHQLWS_SUBPROTOCOL],
    )

    async with Client(transport=transport) as session:
        subscription = gql("""
            subscription StockAlert($threshold: Int!) {
                stockAlert(threshold: $threshold) {
                    id
                    name
                    stockCount
                }
            }
        """)

        async for result in session.subscribe(subscription, variable_values={"threshold": threshold}):
            product = result["stockAlert"]
            if product:
                print(f"Low stock alert: {product['name']} — {product['stockCount']} remaining")

if __name__ == "__main__":
    asyncio.run(monitor_stock_alerts(threshold=10))

Error Handling

from gql.transport.exceptions import TransportQueryError
from gql.transport.aiohttp import AIOHTTPTransport

async def safe_get_product(gateway: ProductGraphQLGateway, product_id: str):
    try:
        return await gateway.get_product(product_id)
    except TransportQueryError as e:
        # GraphQL errors returned by the server (resolver threw an error)
        for error in e.errors or []:
            code = error.get("extensions", {}).get("code")
            if code == "NOT_FOUND":
                return None
            if code == "FORBIDDEN":
                raise PermissionError(error["message"])
        raise

Comparing Client Approaches

Aspect	TypeScript Apollo Client	Python gql
Type safety	Full — via code generation	Manual — use Pydantic models
Caching	Built-in normalized cache	No built-in cache
Subscriptions	via graphql-ws link	via WebSocket transport
Code generation	@graphql-codegen/cli	n/a
Best for	React apps, Node services	Python microservices, scripts
Auth headers	Apollo link chain	Transport headers

What's Next

You now know how to consume a GraphQL API from both TypeScript (with full type generation) and Python (using gql + Pydantic models).

In Part 5, we secure the API — adding JWT authentication to the Apollo context, implementing field-level authorization, and protecting subscriptions.