Authentication and Context Guide

This guide covers how authentication works in the GraphQL Starter Kit and how to extend it for production use.

Architecture Overview

Client Request
    │
    ├── Authorization: Bearer <JWT>
    │
    ▼
┌────────────────┐
│  HTTP Handler  │  ← extracts headers
└───────┬────────┘
        │
        ▼
┌────────────────┐
│  build_context │  ← parses JWT, builds ctx dict
└───────┬────────┘
        │
        ▼
┌────────────────┐
│   Executor     │  ← passes ctx to every resolver
└───────┬────────┘
        │
        ▼
┌────────────────┐
│   Resolver     │  ← checks ctx["user"], ctx["is_authenticated"]
└────────────────┘

Authentication in GraphQL happens in the context layer, not at the transport layer. Unlike REST where middleware can reject requests before routing, GraphQL has a single endpoint — so auth is checked per-field.

Token Flow

1. Login

mutation {
  login(email: "alice@example.com", password: "admin123") {
    token
    user { id displayName role }
  }
}

The server:

1. Finds the user by email.

2. Verifies the password (PBKDF2-HMAC-SHA256).

3. Creates a JWT with claims { user_id, role, iat, exp }.

4. Returns the token.

2. Authenticated Requests

curl -X POST http://127.0.0.1:4000/graphql \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer eyJhbGci..." \
  -d '{"query": "{ me { email displayName } }"}'

3. Token Verification

On each request, build_context():

1. Extracts the Authorization: Bearer header.

2. Verifies the HMAC-SHA256 signature.

3. Checks the exp claim against the current time.

4. Attaches the decoded payload to ctx["user"].

JWT Implementation Details

The included JWT implementation (src/auth.py) uses stdlib only:

import hmac, hashlib, json, base64, time

# Structure: header.payload.signature
# Header:    {"alg": "HS256", "typ": "JWT"}
# Payload:   {"user_id": "42", "role": "ADMIN", "iat": 1234, "exp": 5678}
# Signature: HMAC-SHA256(header_b64 + "." + payload_b64, secret)

For production, use PyJWT — this implementation teaches the structure but lacks:

• RS256 (asymmetric) signing
• JWK key rotation
• Audience/issuer validation
• Token refresh mechanism

Context Building

The context dict is created fresh for every request:

def build_context(headers, secret):
    user = extract_user_from_headers(headers, secret)
    return {
        "user": user,                    # Decoded JWT payload or None
        "is_authenticated": user is not None,
        "loaders": None,                 # Attached by server
    }

Resolvers access the context as their third argument:

def resolve_me(parent, args, ctx):
    if not ctx["is_authenticated"]:
        raise AuthError("Not authenticated")
    return get_user(ctx["user"]["user_id"])

Authorization Patterns

Pattern 1: Decorator-Based (Recommended)

from src.auth import require_auth, require_role

@require_auth
def resolve_my_profile(parent, args, ctx):
    return get_user(ctx["user"]["user_id"])

@require_role("ADMIN", "EDITOR")
def resolve_delete_post(parent, args, ctx):
    return delete_post(args["id"])

Pattern 2: Field-Level Ownership

@require_auth
def resolve_update_post(parent, args, ctx):
    post = get_post(args["id"])

    # The post author or an admin can update
    if post["authorId"] != ctx["user"]["user_id"]:
        if ctx["user"]["role"] != "ADMIN":
            raise AuthError("You can only edit your own posts.")

    return update_post(post, args["input"])

Pattern 3: Data Filtering

Instead of blocking access, filter data based on role:

def resolve_posts(parent, args, ctx):
    posts = get_all_posts()

    # Non-admin users only see published posts
    if not ctx.get("is_authenticated") or ctx["user"]["role"] != "ADMIN":
        posts = [p for p in posts if p["status"] == "PUBLISHED"]

    return paginate(posts, args)

Security Best Practices

1. Secret Management

# NEVER hardcode secrets in source
SECRET = "YOUR_SECRET_KEY_HERE"  # BAD

# Read from environment
import os
SECRET = os.environ["GRAPHQL_AUTH_SECRET"]  # GOOD

2. Token Expiration

# Short-lived access tokens + refresh tokens
ACCESS_TTL = 900       # 15 minutes
REFRESH_TTL = 604800   # 7 days

3. Rate Limiting

Protect the login and register mutations from brute-force attacks:

# In a production setup (e.g., with FastAPI):
from slowapi import Limiter

@app.post("/graphql")
@limiter.limit("10/minute", key_func=get_remote_address)
async def graphql_endpoint(request):
    ...

4. Input Validation

@resolvers.field("Mutation", "register")
def resolve_register(parent, args, ctx):
    email = args["email"]

    # Validate email format
    if "@" not in email or "." not in email.split("@")[1]:
        raise ValueError("Invalid email format.")

    # Validate password strength
    password = args["password"]
    if len(password) < 8:
        raise ValueError("Password must be at least 8 characters.")

    # Check uniqueness
    if user_exists(email):
        raise ValueError("Email already registered.")

5. Query Depth Limiting

Prevent deeply nested queries that could cause exponential data fetching:

MAX_DEPTH = 10

def check_depth(fields, current_depth=0):
    if current_depth > MAX_DEPTH:
        raise ValueError(f"Query exceeds maximum depth of {MAX_DEPTH}.")
    for field in fields:
        if field.sub_fields:
            check_depth(field.sub_fields, current_depth + 1)

Migrating to Production Auth

When you're ready to move beyond the built-in JWT:

Replace src/auth.py with your chosen auth provider (Auth0, Keycloak, Firebase Auth) while keeping the same build_context() → ctx["user"] interface.

Part of API Developer Pro. Support: support@datanest.dev

The N+1 Problem and DataLoader

The N+1 problem is the most common performance issue in GraphQL APIs. This guide explains what it is, why GraphQL is especially vulnerable, and how the DataLoader pattern solves it.

What Is the N+1 Problem?

Consider this query:

query {
  posts(first: 20) {
    edges {
      node {
        title
        author {
          displayName
        }
      }
    }
  }
}

A naive implementation:

1. 1 query to fetch 20 posts.

2. 20 queries to fetch each post's author (one per post).

Total: 21 queries — that's the "N+1" (1 for the list + N for each item's related data).

If posts also have category and tags:

1 (posts) + 20 (authors) + 20 (categories) + 20 (tags) = 61 queries

This gets worse with nesting. A query 3 levels deep can easily generate hundreds of database round-trips.

Why GraphQL Is Especially Vulnerable

In REST, the server controls what data is fetched — it can use JOINs and eager loading. In GraphQL, the client controls the query shape, meaning:

• The server can't predict which fields will be requested.
• Each field resolver runs independently.
• Resolvers don't know about sibling resolvers' needs.

This means two separate Post.author resolver calls for user-42 don't know about each other — each makes its own database call.

The DataLoader Solution

DataLoader solves N+1 by introducing two mechanisms:

1. Batching

Instead of loading one item at a time, DataLoader collects all load requests in a "tick" and dispatches them as a single batch.

Without DataLoader:
  resolve author for post-1 → SELECT * FROM users WHERE id = 'user-2'
  resolve author for post-2 → SELECT * FROM users WHERE id = 'user-2'
  resolve author for post-3 → SELECT * FROM users WHERE id = 'user-1'
  ...
  = 20 separate queries

With DataLoader:
  resolve author for post-1 → loader.load('user-2')  (queued)
  resolve author for post-2 → loader.load('user-2')  (queued, deduped)
  resolve author for post-3 → loader.load('user-1')  (queued)
  ...
  loader.dispatch()
  → SELECT * FROM users WHERE id IN ('user-1', 'user-2')
  = 1 query

2. Caching

Within a single request, DataLoader caches results by key. If user-2 is loaded once, subsequent loads return the cached value — zero additional queries.

Important: The cache is request-scoped. Each request gets a fresh DataLoader to prevent stale data and ensure request isolation.

How the Included DataLoader Works

The included DataLoader class (in src/dataloader.py) is synchronous, designed for the stdlib http.server.

Basic Usage

from src.dataloader import DataLoader

def batch_load_users(ids: list[str]) -> list[dict | None]:
    """Load multiple users in a single call.  Must return values in the same order as ids."""
    users = {u["id"]: u for u in db.find_users(ids)}
    return [users.get(uid) for uid in ids]

# Create a loader (one per request)
loader = DataLoader(batch_load_users)

# Phase 1: Queue loads
loader.load("user-1")  # Returns None (not loaded yet)
loader.load("user-2")
loader.load("user-1")  # Deduped — same key as above

# Phase 2: Dispatch the batch
results = loader.dispatch()
# results = {"user-1": {...}, "user-2": {...}}

# Phase 3: Retrieve cached values
user = loader.load("user-1")  # Returns cached dict instantly

The Two-Phase Pattern (Sync)

Since this implementation is synchronous (no async/await), loading happens in two phases:

1. Queue phase — Resolvers call loader.load(key), which queues the key and returns None.

2. Dispatch phase — After all resolvers at a depth level have run, call loader.dispatch() to execute the batch.

In the async version (used with aiohttp/Starlette), load() returns a Future and dispatch happens automatically via the event loop.

DataLoader Registry

Use the DataLoaderRegistry to manage multiple loaders:

from src.dataloader import DataLoaderRegistry

# In request setup:
loaders = DataLoaderRegistry()
loaders.register("users", batch_load_users)
loaders.register("categories", batch_load_categories)

# In resolver:
user = ctx["loaders"].get("users").load(user_id)

# After each depth level:
ctx["loaders"].dispatch_all()

Writing a Batch Function

The batch function is the bridge between DataLoader and your data source. It must follow these rules:

1. Accept a list of keys.

2. Return a list of values in the same order.

3. Return None for missing keys (not an empty list).

4. Be as efficient as possible — use WHERE id IN (...) or equivalent.

def batch_load_products(ids: list[str]) -> list[dict | None]:
    """
    CORRECT: Returns values in the same order as the input ids.
    Missing products get None, not omitted.
    """
    rows = db.execute("SELECT * FROM products WHERE id IN %s", (tuple(ids),))
    by_id = {row["id"]: row for row in rows}
    return [by_id.get(pid) for pid in ids]  # Same order, None for missing

Common mistakes:

# WRONG: Returns in arbitrary order
def bad_batch(ids):
    return db.find({"id": {"$in": ids}})  # Order not guaranteed!

# WRONG: Omits missing items
def bad_batch(ids):
    return [item for item in db.find(ids) if item is not None]
    # If 3 items requested and 1 is missing, returns 2 items instead of 3

Performance Comparison

Monitoring DataLoader Performance

The included DataLoader tracks statistics:

loader = DataLoader(batch_fn)
# ... use the loader ...

print(loader.stats)
# {"cache_size": 15, "queue_size": 0, "dispatches": 3}

Log these in production to identify:

• High dispatch counts — might indicate the loader isn't batching effectively.
• Large cache sizes — might indicate a large result set that could use pagination.
• Mismatched batch sizes — the batch function is returning wrong-sized results.

Gotchas

1. Request-Scoped Cache

Create a new DataLoader per request. A shared loader would return stale data and leak data between users.

2. Mutation Invalidation

After a mutation that changes data, call loader.clear(key) or loader.clear() to avoid returning stale cached data.

3. Circular References

If type A loads type B which loads type A, you can get infinite recursion. Set a maximum query depth to prevent this.

Part of API Developer Pro. Support: support@datanest.dev