“Think about describing the data, not the view. Model your data as objects in the graph — the API should expose data semantics, not feature-specific payloads.”
— Lee Byron (co-creator of GraphQL)
Summary
This note defines GraphQL as a schema-driven data-access layer that lets clients ask for exactly the fields they need, then shows how SDL, resolvers, pagination, authorization, federation, and Python implementations turn that flexibility into a workable interface for analytical and operational data products.
GraphQL model and schema design
Explains what GraphQL is, when data engineers use it, and how the schema definition language expresses types, inputs, enums, interfaces, unions, and directives.
Treats the schema as the central contract that governs query shape, field meaning, and consumer expectations across one endpoint.
Queries, mutations, subscriptions, and resolvers
Covers read, write, and subscription flows, then explains resolver execution, SQL-backed resolution, and Python server implementations with Strawberry and Ariadne.
Connects GraphQL’s expressive query model to the operational reality that every requested field must still be resolved efficiently and securely.
Performance, auth, and scaling patterns
Explains the N+1 problem, DataLoader batching, cursor pagination, auth and authorization in resolvers, introspection, federation, and real GitHub GraphQL automation examples.
Uses these sections to show where GraphQL shines for flexible consumer-facing access and where it becomes operationally demanding.
Operations and safety
Warnings: careless resolvers create N+1 explosions, field-level auth is mandatory, and schema evolution needs deliberate deprecation rather than endpoint versioning.
Recommendations: design the schema around stable domain semantics, batch resolver access with DataLoader, keep auth in resolver context, and use federation only when domain boundaries are real and maintained.
Glossary
GraphQL
A query language and execution runtime for APIs where clients request exactly the fields they need from a typed schema.
It matters here because the note treats GraphQL as a flexible data-access layer for serving multiple consumers from one contract.
Client-shaped responses
GraphQL is valuable when different consumers need different shapes from the same underlying domain without proliferating bespoke endpoints.
Schema Definition Language / SDL
The type-definition syntax used to declare GraphQL object types, inputs, enums, interfaces, unions, and fields.
It matters here because the SDL is the source of truth that makes the API contract introspectable and strongly typed.
Contract before resolver code
The schema is what consumers explore and depend on. Resolver code should implement that contract, not invent it ad hoc.
Resolver
The server-side function that supplies the value for a requested GraphQL field.
It matters here because the flexibility of GraphQL is only as good as the efficiency and correctness of its resolver layer.
Field freedom has a cost
A single GraphQL request can fan out into many backend calls. Resolver design determines whether that power becomes elegant data access or a performance incident.
Query
A read operation in GraphQL that requests data shaped by the client.
It matters here because read flexibility is the main reason many consumers adopt GraphQL over fixed-shape REST endpoints.
One endpoint, many shapes
Queries are powerful because they let clients ask for nested structures without requiring the server to publish one endpoint per response variation.
Mutation
A write operation in GraphQL used to create, update, or otherwise change server-side state.
It matters here because write behavior still needs explicit contracts, auth, and side-effect discipline even in a query-centric API model.
Flexibility does not remove write risk
Mutations can be expressive, but they still need the same care around validation, authorization, and idempotency as other state-changing APIs.
Subscription
A GraphQL operation that pushes updates over time to subscribed clients rather than returning one static response.
It matters here because some data products need live update behavior without abandoning the GraphQL schema model.
Schema-based real-time channel
Subscriptions let teams keep one conceptual contract across both snapshot queries and live updates, though operational support is more complex than plain queries.
Fragment
A reusable named selection set that lets clients avoid repeating the same field group in multiple queries.
It matters here because large GraphQL consumers often rely on fragments to keep complex request documents maintainable.
Reuse on the client side
Fragments improve consistency by making repeated field groups explicit instead of copy-pasted across many consumer queries.
Introspection
The ability of a GraphQL service to expose its own schema structure for discovery and tooling.
It matters here because introspection is a major reason GraphQL APIs feel self-documenting to consumers and tools.
Powerful but sensitive
Introspection improves developer experience, but some production environments restrict it because schema visibility can also aid attackers or leak internal structure.
DataLoader
A batching and caching pattern used to combine many resolver-level lookups into fewer backend queries.
It matters here because it is the standard mitigation for the N+1 query problem in GraphQL servers.
Batch by access pattern
DataLoader works well because GraphQL resolver execution naturally creates repeated similar lookups that can often be coalesced per request.
Cursor pagination
A pagination style that uses opaque cursors to move through ordered results safely rather than relying on simple numeric offsets.
It matters here because GraphQL APIs frequently need stable pagination under concurrent inserts and large result sets.
Offset is often too weak
Offset pagination becomes unreliable when data changes between requests. Cursor designs preserve order and continuation more robustly.
Federation
A GraphQL architecture where multiple domain-owned subgraphs combine into one composed schema.
It matters here because federation is often the scaling pattern teams reach for once one GraphQL schema spans many independently owned services.
Works only with real domain ownership
Federation adds coordination overhead. It pays off only when service and schema boundaries genuinely match organizational ownership.
What GraphQL Is
GraphQL is a query language for APIs, a type system for describing data, and a runtime for executing queries — developed by Facebook in 2012 and open-sourced in 2015. Unlike REST, the server exposes a schema and the client declares exactly what it needs; the server resolves each requested field independently against whatever data source backs it.
GraphQL is three things simultaneously:
1. A query language — clients write structured queries describing the data shape they need.
2. A type system — the schema is the contract between client and server, defining every type, field, and relationship.
3. A runtime — the server executes queries by calling resolver functions for each requested field.
flowchart LR
C["Client<br/>query { index(code:SPX)<br/>{ name constituents<br/>performance } }"]
GQL["GraphQL Server<br/>/graphql<br/>One HTTP POST"]
IR["IndexResolver"]
CR["ConstituentResolver"]
PR["PerformanceResolver"]
PG[("PostgreSQL")]
BQ[("BigQuery")]
TS[("TimescaleDB")]
C -->|POST /graphql| GQL
GQL --> IR
GQL --> CR
GQL --> PR
IR --> PG
CR --> BQ
PR --> TS
The single query above fetches data from three different storage systems in one round trip. The client specifies exactly which fields it needs — no more, no less.
GraphQL Wire Format
GraphQL runs over HTTP. Queries are typically sent as POST /graphql with a JSON body: {"query": "...", "variables": {...}}. Responses are JSON: {"data": {...}, "errors": [...]}. Unlike gRPC, there is no binary encoding by default. See serialization-formats for encoding trade-offs.
When Data Engineers Use GraphQL
1. Flexible data access layers
A financial analytics platform serves both a trading dashboard (needs real-time prices, risk metrics) and a regulatory reporting pipeline (needs positions, trades, reference data). Rather than building separate REST endpoints for each consumer, one GraphQL API serves both — each client requests only what it needs.
2. Data mesh API layers
In a data mesh, each domain exposes its data as a product. GraphQL is well-suited as the product interface because it is self-documenting, introspectable, and flexible enough to serve any consumer without versioning.
3. Serving multiple consumers from one endpoint
Mobile apps, web dashboards, Jupyter notebooks, and pipeline scripts all have different data needs. REST APIs accumulate bespoke endpoints over time. A GraphQL API stays clean — clients compose their own queries.
4. GitHub API (data engineers use it daily)
The GitHub GraphQL API is the canonical example of GraphQL at scale. Data engineers use it to automate pipeline deployments, track PR states, monitor CI runs, and extract repository metadata for reporting. It replaced the GitHub REST v3 API for most complex queries.
5. Replacing multiple REST calls with one GraphQL query
A pipeline that needs to fetch a company’s profile, its recent filings, and the filing attachments from a REST API makes 3 round trips. The equivalent GraphQL query makes 1.
Real Pipeline Use Case
An index rebalancing pipeline uses the GitHub GraphQL API to find the latest tagged release of a factor model repository, download the asset list CSV, and open a pull request with the new constituent weights — all in one script with three GraphQL mutations and queries.
Schema Definition Language (SDL)
The SDL is the heart of a GraphQL API. It defines every type the API exposes.
Scalar Types
GraphQL has five built-in scalars: String (UTF-8), Int (32-bit signed), Float (64-bit), Boolean, and ID (unique identifier, serialized as string). Custom scalars let you define domain-specific types like Date, DateTime, Decimal, and JSON — the implementation is server-side.
Object types define the data nodes in the graph. The ! suffix marks a field as non-null; omitting it means the field is nullable. Fields can take arguments for filtering, pagination, and parameterization — these become part of the schema contract.
type Index { code: String! name: String! description: String assetClass: AssetClass! constituents( sector: String minWeight: Float limit: Int = 100 ): [Constituent!]! performance( from: Date! to: Date! frequency: Frequency = DAILY ): [DailyPerformance!]! lastRebalanced: Date totalReturn(from: Date!, to: Date!): Float}type Constituent { symbol: String! name: String! weight: Float! sector: String! industry: String country: String! marketCap: Float price: Price}type Price { current: Float! open: Float high: Float low: Float previousClose: Float change: Float changePct: Float asOf: DateTime!}type DailyPerformance { date: Date! returnPct: Float! # daily return as percentage cumulativePct: Float! # cumulative return from query start date indexLevel: Float! volume: Int}enum AssetClass { EQUITY FIXED_INCOME COMMODITY FX CRYPTO MULTI_ASSET}enum Frequency { DAILY WEEKLY MONTHLY QUARTERLY}
Interfaces and Unions
Interfaces define shared fields that multiple types must implement — use them when different types have a common shape but differ in additional fields. Unions are looser: a union field can return one of several types that share no fields, requiring inline fragments to select type-specific data.
Directives annotate fields and types with additional behavior. The built-in @include and @skip conditionally include fields at query time. Custom directives declared in the schema (@auth, @cached, @rateLimit) are implemented server-side and enforce cross-cutting concerns without coupling them to resolver logic.
query GetIndex($includePerformance: Boolean = false) { index(code: "SPX") { name constituents { symbol weight } performance(from: "2025-01-01", to: "2025-03-22") @include(if: $includePerformance) { date returnPct } }}directive @deprecated(reason: String) on FIELD_DEFINITIONdirective @auth(roles: [String!]!) on FIELD_DEFINITION | OBJECTdirective @rateLimit(max: Int!, window: String!) on FIELD_DEFINITIONdirective @cached(ttl: Int!) on FIELD_DEFINITIONtype Query { index(code: String!): Index @cached(ttl: 60) adminStats: AdminStats @auth(roles: ["ADMIN"]) legacyPrice(symbol: String!): Float @deprecated(reason: "Use index(code).constituents.price instead")}
Queries
Queries are read-only operations. A client declares the exact fields it needs, and the server resolves each field independently — no more, no less data is returned.
Basic Query
A named query with no variables, selecting specific fields from a specific index.
query GetSPXConstituents { index(code: "SPX") { name constituents(sector: "Technology", limit: 10) { symbol name weight sector } }}
Variables are passed as a separate JSON object alongside the query string, allowing the same named query to be reused with different inputs. This is the standard pattern for parameterized queries in pipeline scripts.
query GetIndexPerformance($code: String!, $range: DateRange!) { index(code: $code) { name performance(from: $range.from, to: $range.to) { date returnPct cumulativePct indexLevel } }}
Aliases let you call the same field multiple times in one request with different arguments. Each alias becomes a key in the response object. This replaces three separate REST calls with a single round trip.
query CompareIndices { spx: index(code: "SPX") { name totalReturn(from: "2025-01-01", to: "2025-03-22") } ndx: index(code: "NDX") { name totalReturn(from: "2025-01-01", to: "2025-03-22") } ftse: index(code: "FTSE100") { name totalReturn(from: "2025-01-01", to: "2025-03-22") }}
Fragments (Reusable Field Sets)
Fragments define a named set of fields that can be spread into multiple queries with ...FragmentName, reducing duplication when the same field selection appears in many places.
fragment ConstituentFields on Constituent { symbol name weight sector price { current changePct asOf }}query TechHeavyIndices { spx: index(code: "SPX") { constituents(sector: "Technology") { ...ConstituentFields } } ndx: index(code: "NDX") { constituents(sector: "Technology") { ...ConstituentFields } }}
Inline Fragments for Unions
When a field returns a union or interface, inline fragments select type-specific fields. The __typename meta-field lets the client determine which concrete type was returned.
query SearchSecurities { search(query: "Apple", assetClasses: [EQUITY]) { __typename ... on Equity { symbol name sector pe dividendYield } ... on Bond { symbol name coupon maturity yieldToMaturity } ... on Index { code name assetClass } }}
Mutations
Mutations are write operations — creating, updating, or deleting data. They return the modified object, so the client can refresh its state in the same round trip without a follow-up query.
GraphQL subscriptions are WebSocket-based and not appropriate for high-throughput data (thousands of events per second). For real-time market data feeds, use gRPC server streaming or a message broker from streaming-architecture. Use GraphQL subscriptions for user-facing real-time updates at human-readable frequencies.
Right Tool for Each Frequency
Use GraphQL subscriptions for dashboard-level updates (portfolio value refreshing every few seconds, pipeline status notifications) where human-readable frequency is sufficient. For tick-level market data or high-throughput pipeline events (>100 events/sec), route through gRPC server streaming or a Pub/Sub topic and expose a separate WebSocket or SSE endpoint — keeping the GraphQL API clean for query-oriented use cases.
Resolvers
Resolvers are the functions that execute when a field is requested. Each field in the schema maps to a resolver.
Resolver Execution Model
The execution engine calls the root resolver first, then calls child resolvers for each requested field — passing the parent object’s result as the first argument. Execution is depth-first; each level is resolved before moving deeper.
flowchart TD
Q["query { index(code: SPX) }"]
IR["indexResolver(code)"]
N["field: name"]
CR["constituentsResolver(index)"]
SYM["field: symbol"]
WGT["field: weight"]
PR["priceResolver(constituent)"]
CUR["field: current"]
Q --> IR
IR --> N
IR --> CR
CR --> SYM
CR --> WGT
CR --> PR
PR --> CUR
### SQL Resolver Example
Root resolvers receive `_` (parent, which is `None` for root fields), `info` (execution context carrying the request, auth user, and injected clients), and any field arguments. Child resolvers receive the parent object as the first positional argument.
```python
import asyncpg
from dataclasses import dataclass
from typing import Optional
@dataclass
class IndexRow:
code: str
name: str
description: Optional[str]
asset_class: str
last_rebalanced: Optional[str]
async def resolve_index(_, info, code: str) -> Optional[IndexRow]:
pool: asyncpg.Pool = info.context["db_pool"]
row = await pool.fetchrow(
"""
SELECT code, name, description, asset_class, last_rebalanced
FROM indices
WHERE code = $1 AND is_active = TRUE
""",
code,
)
if row is None:
return None
return IndexRow(**row)
async def resolve_constituents(
index: IndexRow,
info,
sector: Optional[str] = None,
min_weight: Optional[float] = None,
limit: int = 100,
) -> list[dict]:
pool: asyncpg.Pool = info.context["db_pool"]
query = """
SELECT
c.symbol,
s.name,
ic.weight,
s.sector,
s.industry,
s.country,
s.market_cap
FROM index_constituents ic
JOIN constituents c ON c.id = ic.constituent_id
JOIN securities s ON s.symbol = c.symbol
WHERE ic.index_code = $1
AND ic.is_current = TRUE
"""
params = [index.code]
idx = 2
if sector:
query += f" AND s.sector = ${idx}"
params.append(sector)
idx += 1
if min_weight is not None:
query += f" AND ic.weight >= ${idx}"
params.append(min_weight)
idx += 1
query += f" ORDER BY ic.weight DESC LIMIT ${idx}"
params.append(limit)
rows = await pool.fetch(query, *params)
return [dict(r) for r in rows]
async def resolve_performance(
index: IndexRow,
info,
from_: str,
to: str,
frequency: str = "DAILY",
) -> list[dict]:
bq_client = info.context["bq_client"]
query = f"""
SELECT
date,
daily_return_pct AS return_pct,
cumulative_pct,
index_level,
total_volume AS volume
FROM `project.finance.index_performance`
WHERE index_code = @index_code
AND date BETWEEN @from_date AND @to_date
ORDER BY date
"""
job_config = bq_client.QueryJobConfig(
query_parameters=[
bq_client.ScalarQueryParameter("index_code", "STRING", index.code),
bq_client.ScalarQueryParameter("from_date", "DATE", from_),
bq_client.ScalarQueryParameter("to_date", "DATE", to),
]
)
result = bq_client.query(query, job_config=job_config).result()
return [dict(r) for r in result]
Python GraphQL Server with Strawberry
Strawberry is a code-first GraphQL library for Python. You define types as Python dataclasses decorated with @strawberry.type.
Ariadne is a schema-first (SDL-first) alternative. You write the SDL, then bind resolvers.
# ariadne_server.pyfrom ariadne import QueryType, MutationType, ObjectType, make_executable_schemafrom ariadne.asgi import GraphQL# Load schema from filewith open("schema.graphql") as f: type_defs = f.read()query = QueryType()mutation = MutationType()index_type = ObjectType("Index")constituent_type = ObjectType("Constituent")@query.field("index")async def resolve_query_index(_, info, code: str): pool = info.context["db_pool"] row = await pool.fetchrow( "SELECT code, name, description, asset_class FROM indices WHERE code = $1", code, ) return dict(row) if row else None@index_type.field("constituents")async def resolve_index_constituents(index, info, sector=None, limit=100): pool = info.context["db_pool"] rows = await pool.fetch( """ SELECT c.symbol, s.name, ic.weight, s.sector FROM index_constituents ic JOIN constituents c ON c.id = ic.constituent_id JOIN securities s ON s.symbol = c.symbol WHERE ic.index_code = $1 AND ic.is_current = TRUE ORDER BY ic.weight DESC LIMIT $2 """, index["code"], limit, ) return [dict(r) for r in rows]@constituent_type.field("price")async def resolve_constituent_price(constituent, info): loader = info.context["price_loader"] return await loader.load(constituent["symbol"])schema = make_executable_schema(type_defs, query, mutation, index_type, constituent_type)app = GraphQL(schema, debug=True)
N+1 Query Problem and DataLoader
The N+1 problem is the most important performance issue in GraphQL. It occurs when resolving a list of N items each triggers an individual database query, resulting in N+1 total queries.
The Problem
When resolving a list of N items, each item’s child resolver fires independently — one database query per item. For 500 constituents each requesting a price, that is 502 total queries.
query { index(code: "SPX") { constituents(limit: 500) { symbol price { current changePct } } }}
N+1 Without DataLoader in Production
An unguarded GraphQL API serving 500 constituents will fire 502 database queries per request. At any meaningful load, this collapses the database. Unlike REST endpoints where the developer controls exactly what the query fetches, GraphQL resolvers compose dynamically — the N+1 explosion is invisible until it hits production. Always attach DataLoaders before exposing any list field that has a child resolver.
Fix
DataLoader batches all individual loads that occur in the same async “tick” into a single WHERE symbol = ANY($1) query. 500 price lookups become 1 query.
DataLoader Pattern
DataLoader batches all individual loads that occur in the same tick of the event loop into a single batched query.
# dataloader.pyfrom __future__ import annotationsimport asynciofrom collections import defaultdictfrom typing import Anyimport asyncpgclass PriceLoader: def __init__(self, pool: asyncpg.Pool) -> None: self._pool = pool self._batch: dict[str, asyncio.Future] = {} self._scheduled = False async def load(self, symbol: str) -> dict | None: if symbol not in self._batch: loop = asyncio.get_event_loop() future: asyncio.Future = loop.create_future() self._batch[symbol] = future if not self._scheduled: self._scheduled = True loop.call_soon(self._dispatch) return await self._batch[symbol] def _dispatch(self) -> None: batch = self._batch self._batch = {} self._scheduled = False asyncio.create_task(self._fetch_batch(batch)) async def _fetch_batch(self, batch: dict[str, asyncio.Future]) -> None: symbols = list(batch.keys()) try: rows = await self._pool.fetch( """ SELECT symbol, last_price AS current, previous_close, last_price - previous_close AS change, ROUND( (last_price - previous_close) / previous_close * 100, 4 ) AS change_pct, as_of FROM live_prices WHERE symbol = ANY($1::text[]) """, symbols, ) results = {r["symbol"]: dict(r) for r in rows} except Exception as exc: for fut in batch.values(): if not fut.done(): fut.set_exception(exc) return for symbol, future in batch.items(): if not future.done(): future.set_result(results.get(symbol))def create_price_loader(pool: asyncpg.Pool) -> PriceLoader: return PriceLoader(pool)
strawberry-django and DataLoader
Strawberry integrates with strawberry-django and strawberry-graphql-django which auto-generate DataLoaders for Django ORM relationships. For raw SQL or BigQuery, write your own as above.
BigQuery DataLoader
The same batching pattern applied to BigQuery. Because the BigQuery client is synchronous, the batch fetch runs in a thread executor to avoid blocking the async event loop.
class BigQueryPriceLoader: """Batch historical price lookups against BigQuery.""" def __init__(self, bq_client, as_of_date: str) -> None: self._bq = bq_client self._as_of = as_of_date self._batch: dict[str, asyncio.Future] = {} self._scheduled = False async def load(self, symbol: str) -> dict | None: if symbol not in self._batch: loop = asyncio.get_event_loop() self._batch[symbol] = loop.create_future() if not self._scheduled: self._scheduled = True loop.call_soon(self._dispatch) return await self._batch[symbol] def _dispatch(self) -> None: batch = self._batch self._batch = {} self._scheduled = False asyncio.create_task(self._fetch(batch)) async def _fetch(self, batch: dict[str, asyncio.Future]) -> None: symbols = list(batch.keys()) query = """ SELECT symbol, close_price, open_price, high_price, low_price, volume FROM `project.finance.daily_prices` WHERE symbol IN UNNEST(@symbols) AND price_date = @as_of """ job_config = self._bq.QueryJobConfig( query_parameters=[ self._bq.ArrayQueryParameter("symbols", "STRING", symbols), self._bq.ScalarQueryParameter("as_of", "DATE", self._as_of), ] ) def _run(): return { r["symbol"]: dict(r) for r in self._bq.query(query, job_config=job_config).result() } loop = asyncio.get_event_loop() try: results = await loop.run_in_executor(None, _run) except Exception as exc: for fut in batch.values(): if not fut.done(): fut.set_exception(exc) return for symbol, fut in batch.items(): if not fut.done(): fut.set_result(results.get(symbol))
Pagination: Relay-Style Cursor Connections
Relay-style cursor pagination is the GraphQL standard. It handles arbitrary sort orders safely, unlike offset pagination.
Schema
The Relay schema wraps each item in an Edge that carries both the data node and an opaque cursor. PageInfo exposes navigation state. Clients use first/after for forward pagination and last/before for backward.
type ConstituentConnection { edges: [ConstituentEdge!]! pageInfo: PageInfo! totalCount: Int!}type ConstituentEdge { node: Constituent! cursor: String! # opaque base64-encoded cursor}type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String}type Query { # Forward pagination: after + first # Backward pagination: before + last constituents( indexCode: String! first: Int after: String last: Int before: String sector: String ): ConstituentConnection!}
Resolver Implementation
Cursors encode the sort key (weight + symbol for tie-breaking) as base64 JSON. The resolver fetches page_size + 1 rows to detect whether a next page exists, then trims back to page_size before building edges.
import base64import jsondef encode_cursor(symbol: str, weight: float) -> str: """Encode a stable cursor from the sort key.""" payload = json.dumps({"symbol": symbol, "weight": weight}) return base64.b64encode(payload.encode()).decode()def decode_cursor(cursor: str) -> dict: payload = base64.b64decode(cursor.encode()).decode() return json.loads(payload)async def resolve_constituents_connection( _, info, index_code: str, first: int | None = None, after: str | None = None, last: int | None = None, before: str | None = None, sector: str | None = None,) -> dict: pool = info.context["db_pool"] # Determine pagination direction page_size = first or last or 20 is_forward = first is not None or (first is None and last is None) query = """ SELECT c.symbol, s.name, ic.weight, s.sector, s.country FROM index_constituents ic JOIN constituents c ON c.id = ic.constituent_id JOIN securities s ON s.symbol = c.symbol WHERE ic.index_code = $1 AND ic.is_current = TRUE """ params = [index_code] idx = 2 if sector: query += f" AND s.sector = ${idx}" params.append(sector) idx += 1 if after: cursor = decode_cursor(after) query += f" AND (ic.weight < ${idx} OR (ic.weight = ${idx} AND c.symbol > ${idx+1}))" params.extend([cursor["weight"], cursor["weight"], cursor["symbol"]]) idx += 2 query += " ORDER BY ic.weight DESC, c.symbol ASC" query += f" LIMIT ${idx}" params.append(page_size + 1) # fetch one extra to detect hasNextPage rows = await pool.fetch(query, *params) has_next = len(rows) > page_size rows = rows[:page_size] total = await pool.fetchval( "SELECT COUNT(*) FROM index_constituents WHERE index_code = $1 AND is_current = TRUE", index_code, ) edges = [ { "node": dict(r), "cursor": encode_cursor(r["symbol"], r["weight"]), } for r in rows ] return { "edges": edges, "totalCount": total, "pageInfo": { "hasNextPage": has_next, "hasPreviousPage": after is not None, "startCursor": edges[0]["cursor"] if edges else None, "endCursor": edges[-1]["cursor"] if edges else None, }, }
Client Pagination Loop
A pipeline script follows cursors until hasNextPage is false, collecting all pages into a flat list.
import httpximport jsonQUERY = """query GetConstituents($indexCode: String!, $after: String) { constituents(indexCode: $indexCode, first: 100, after: $after) { edges { node { symbol name weight sector } cursor } pageInfo { hasNextPage endCursor } totalCount }}"""async def fetch_all_constituents(base_url: str, index_code: str) -> list[dict]: all_nodes = [] cursor = None async with httpx.AsyncClient() as client: while True: response = await client.post( f"{base_url}/graphql", json={"query": QUERY, "variables": {"indexCode": index_code, "after": cursor}}, ) data = response.json()["data"]["constituents"] all_nodes.extend(edge["node"] for edge in data["edges"]) print(f" Fetched {len(all_nodes)}/{data['totalCount']}") if not data["pageInfo"]["hasNextPage"]: break cursor = data["pageInfo"]["endCursor"] return all_nodes
Authentication and Authorization in Resolvers
GraphQL has no built-in auth layer. Authentication is handled in the context function that runs before any resolver; authorization is enforced inside individual resolvers or via declarative permission classes. Both happen at the application layer, not the transport layer.
GraphQL Authentication — Context-Based Auth
The context function decodes the JWT from the Authorization header and attaches the user object. Every resolver then reads from info.context["user"] — no middleware required.
Strawberry’s BasePermission classes let you declare access requirements directly on any field via permission_classes=[...]. The framework calls has_permission before the resolver; returning False adds a GraphQL error without a stack trace.
Some access rules are data-specific: a user can only see their own portfolios. This filtering belongs in the resolver query, not in middleware — the resolver has the user context and can scope the SQL WHERE clause accordingly.
async def resolve_portfolios(_, info) -> list[dict]: user = info.context["user"] pool = info.context["db_pool"] if user is None: raise PermissionError("Authentication required") if "admin" in user["roles"]: rows = await pool.fetch("SELECT * FROM portfolios ORDER BY created_at DESC") else: rows = await pool.fetch( "SELECT * FROM portfolios WHERE owner_id = $1 ORDER BY created_at DESC", user["sub"], ) return [dict(r) for r in rows]
GraphQL Introspection
GraphQL schemas are self-documenting. Clients can query the schema itself.
# Discover all types in the schemaquery IntrospectSchema { __schema { types { name kind description fields { name type { name kind ofType { name kind } } description args { name type { name } defaultValue } } } }}# Discover a specific typequery IntrospectIndex { __type(name: "Index") { name description fields { name description type { name kind ofType { name kind } } } }}
# Generate documentation from introspection in a pipeline scriptimport httpximport jsonasync def get_schema_types(base_url: str) -> list[dict]: query = """ { __schema { types { name kind description fields { name description type { name kind ofType { name } } } } } } """ async with httpx.AsyncClient() as client: response = await client.post( f"{base_url}/graphql", json={"query": query}, ) schema = response.json()["data"]["__schema"] # Filter out intrinsic types (starting with __) return [t for t in schema["types"] if not t["name"].startswith("__")]
Introspection in Production
Introspection reveals every type, field, argument, and resolver in your schema — a complete map for attackers to enumerate attack surface. Disable it in production after documenting your API:
Use a schema registry (Apollo Studio, GraphQL Inspector) to give internal teams schema documentation without exposing introspection to the public endpoint.
GraphQL vs REST Comparison
Both GraphQL and REST run over HTTP and return JSON. The choice depends on consumer diversity and response shape stability — not on performance or security.
GraphQL or REST?
Multiple consumers needing different projections (dashboard vs pipeline vs notebook) → GraphQL: each composes its own query, no endpoint accumulation
Public API with stable, well-defined response shapes → REST: simpler, natively cacheable, lower learning curve
Aggressive HTTP caching required (CDN, ETags) → REST: GraphQL POST queries bypass standard HTTP caches
File uploads or binary payloads → REST: GraphQL multipart is awkward
High-throughput streaming (>100 events/sec) → neither; use gRPC or Pub/Sub
Small team, CRUD operations only → REST: GraphQL overhead (SDL, resolvers, DataLoader) is not justified
Dimension
GraphQL
REST
Endpoint count
One (/graphql)
Many (/indices, /prices, /portfolios)
Overfetching
Never (client specifies fields)
Common (fixed response shape)
Underfetching
Never (nested queries, one round trip)
Common (multiple requests needed)
Type system
Built-in (SDL)
Optional (OpenAPI/Swagger)
Versioning
Schema evolution with deprecations
URL versioning (/v1, /v2)
Caching
Difficult (POST, dynamic queries)
Built-in HTTP caching
File uploads
Awkward (multipart spec)
Native multipart
Streaming
Subscriptions (WebSocket)
SSE, WebSocket, chunked transfer
Error handling
errors array alongside data
HTTP status codes
Tooling
GraphiQL, Apollo Studio, Rover
Postman, curl, SwaggerUI
Learning curve
Higher (SDL, resolvers, DataLoader)
Lower
Browser caching
Not via HTTP Cache
ETags, Cache-Control
N+1 risk
High without DataLoader
Controlled at endpoint level
Performance
JSON over HTTP (same as REST)
JSON over HTTP
Real-time
Subscriptions
Webhooks, SSE
Self-documenting
Yes (introspection)
With OpenAPI
Decision Rule
Use GraphQL when different consumers need different shapes of the same data, or when you want to aggregate multiple data sources into one query. Use REST when responses are stable, caching is important, or the API is public-facing with simple operations. See rest-api-design-and-consumption for REST patterns.
GraphQL Federation
Federation lets you compose a unified GraphQL schema from multiple independent subgraph services — the foundation of a data mesh API layer. Each domain team owns and deploys its own subgraph; a gateway composes them into a single supergraph schema that clients query as one API.
Single Server or Federation?
Single team, one data domain → single Strawberry/Ariadne server; federation adds deployment complexity without benefit
Multiple domain teams, each owning a slice of the schema → federation: teams evolve their subgraph independently, types can span services via entity references
Data mesh product interfaces → federation maps naturally to the “data as a product” principle — each domain publishes a typed subgraph
Early-stage product → start with a monolith and migrate to federation later; premature federation is over-engineering
The index subgraph declares Index and Constituent as entities with @key — the field(s) that uniquely identify them. Other subgraphs reference these entities by key to extend them with additional fields.
The pricing subgraph extends Constituent with price-related fields. It references symbol as the entity key (@external) without owning the type definition — the gateway merges both subgraph schemas at query time.
extend schema @link(url: "https://specs.apollo.dev/federation/v2.0")# Extend Constituent type defined in index subgraphtype Constituent @key(fields: "symbol") { symbol: String! @external price: Price history(from: Date!, to: Date!): [DailyPrice!]!}type Price { current: Float! changePct: Float! asOf: DateTime!}
GraphQL Federation — Python Subgraph with Strawberry
@strawberry.federation.type(keys=["symbol"]) marks the type as a federated entity. resolve_reference is called by the gateway when it needs to hydrate an entity from its key fields — the pricing subgraph receives a Constituent stub with only symbol populated and loads the rest.
import strawberryfrom strawberry.federation import Schema@strawberry.federation.type(keys=["symbol"])class Constituent: symbol: strawberry.ID @classmethod def resolve_reference(cls, symbol: strawberry.ID) -> "Constituent": return cls(symbol=symbol) @strawberry.field async def price(self, info) -> Optional[Price]: loader = info.context["price_loader"] return await loader.load(str(self.symbol)) @strawberry.field async def history( self, info, from_: str, to: str ) -> list[DailyPrice]: pool = info.context["db_pool"] rows = await pool.fetch( "SELECT date, close_price FROM daily_prices WHERE symbol = $1 AND date BETWEEN $2 AND $3", str(self.symbol), from_, to, ) return [DailyPrice(date=str(r["date"]), close=r["close_price"]) for r in rows]schema = Schema(query=Query, types=[Constituent])
Real-World: GitHub GraphQL API for Pipeline Automation
Data engineers use the GitHub GraphQL API daily. Examples below show common automation patterns.
GitHub GraphQL API — Setup and Authentication
A minimal async client wrapping httpx. All GitHub GraphQL requests authenticate with a Bearer token from the environment and raise on both HTTP errors and GraphQL-level errors.
import httpximport osGITHUB_TOKEN = os.environ["GITHUB_TOKEN"]GITHUB_API = "https://api.github.com/graphql"async def github_query(query: str, variables: dict = None) -> dict: async with httpx.AsyncClient() as client: response = await client.post( GITHUB_API, headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}, json={"query": query, "variables": variables or {}}, ) response.raise_for_status() result = response.json() if "errors" in result: raise RuntimeError(f"GraphQL errors: {result['errors']}") return result["data"]
GitHub GraphQL API — Find Latest Release of a Factor Model
Fetches the latest published release of a repository, returning the tag name, publish date, and download URLs for all attached release assets — useful for pulling versioned model files in a pipeline.
GitHub GraphQL API — Find Open PRs with a Specific Label
Lists open pull requests carrying a specific label, ordered by creation date. Used to audit pipeline-update PRs or trigger automation on labeled branches.
GitHub GraphQL API — Create a Pull Request Programmatically
Opens a PR from a given head branch to a base branch. The repositoryId is a node ID retrieved from a separate repository query. The mutation returns the PR number and URL for logging and notification.
Polls the check run rollup for a branch until the CI state resolves to SUCCESS, FAILURE, or ERROR. Used to gate downstream pipeline steps on CI passing.
query GetCIStatus($owner: String!, $repo: String!, $branch: String!) { repository(owner: $owner, name: $repo) { ref(qualifiedName: $branch) { target { ... on Commit { statusCheckRollup { state contexts(first: 20) { nodes { ... on CheckRun { name status conclusion startedAt completedAt } } } } } } } }}
async def wait_for_ci(owner: str, repo: str, branch: str, timeout: int = 600) -> str: import asyncio, time QUERY = """... (above query) ...""" start = time.monotonic() while time.monotonic() - start < timeout: data = await github_query(QUERY, {"owner": owner, "repo": repo, "branch": f"refs/heads/{branch}"}) ref = data["repository"]["ref"] if ref and ref["target"].get("statusCheckRollup"): state = ref["target"]["statusCheckRollup"]["state"] if state in ("SUCCESS", "FAILURE", "ERROR"): return state print(f" CI state: {state} — waiting...") await asyncio.sleep(30) raise TimeoutError(f"CI did not complete within {timeout}s")
When NOT to Use GraphQL
Avoid GraphQL in These Scenarios
Simple CRUD APIs: If every endpoint returns the same shape every time (list users, get user by ID, update user), REST is simpler. GraphQL’s flexibility adds overhead (resolver setup, DataLoader, schema design) that is not justified.
High-throughput data streaming: GraphQL subscriptions are WebSocket-based and JSON-encoded. For thousands of events per second, use gRPC server streaming or Apache Kafka from streaming-architecture. GraphQL subscriptions are for human-scale real-time updates.
File uploads: The GraphQL multipart request spec is awkward. Use signed cloud storage URLs (GCS, S3) or a dedicated REST endpoint for uploads.
Teams unfamiliar with the N+1 problem: An unoptimized GraphQL API can be dramatically slower than REST because of accidental N+1 queries. The DataLoader pattern must be applied diligently. REST endpoints are easier to profile.
Public APIs requiring aggressive HTTP caching: Because GraphQL queries are typically POST requests with unique query strings, standard HTTP caching (ETags, CDNs) does not apply without specialized tooling like persisted queries. REST APIs on GET endpoints cache naturally.
When GraphQL Is the Right Choice
GraphQL excels when multiple consumer teams need different projections of the same data — a trading dashboard, a regulatory pipeline, and a Jupyter notebook each compose their own query without requiring new REST endpoints. Pair it with DataLoaders from day one, enable introspection in development and disable in production, and use persisted queries if aggressive HTTP caching is required. For file uploads, use signed GCS URLs and call the GraphQL mutation with the URL only, never the file bytes.