GraphQL Gateway — Complete Guide
In this tutorial, you'll learn about GraphQL Gateway. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
A GraphQL gateway sits between clients and GraphQL services, combining multiple schemas into a unified API through federation or schema stitching.
What You'll Learn
By the end of this lesson, you will implement an Apollo Federation gateway, understand schema stitching, configure query depth limiting, and rate limit GraphQL queries.
Why It Matters
As GraphQL services grow, clients need to query multiple services. A GraphQL gateway provides a single endpoint that composes data from multiple GraphQL and REST services.
Real-World Use
An e-commerce platform has separate GraphQL services for products, orders, and reviews. The gateway combines them into one schema where a client can query product reviews alongside product details.
GraphQL Gateway Architecture
flowchart LR
Client -->|Single Query| Gateway[GraphQL Gateway]
Gateway -->|Products| ProductS[Product Service]
Gateway -->|Orders| OrderS[Order Service]
Gateway -->|Reviews| ReviewS[Review Service]
Gateway -->|REST| Legacy[Legacy REST API]
Apollo Federation Gateway
# federation_gateway.py
import json
from typing import Any, Dict, List, Optional
class FederationGateway:
def __init__(self):
self.services: Dict[str, dict] = {}
self.type_extensions: Dict[str, List[str]] = {}
def add_service(self, name: str, url: str,
port: int, sdl: str):
self.services[name] = {
"url": f"{url}:{port}",
"sdl": sdl,
}
def compose_schema(self) -> Dict:
all_types = {}
for svc_name, svc_info in self.services.items():
for line in svc_info["sdl"].split("\n"):
if line.strip().startswith("type "):
type_name = line.strip().split()[1]
if type_name not in all_types:
all_types[type_name] = []
all_types[type_name].append(svc_name)
return all_types
def plan_query(self, query: str) -> List[Dict]:
plan = []
if "products" in query.lower() and "products" in self.services:
plan.append({"service": "products", "fields": ["id", "name", "price"]})
if "reviews" in query.lower() and "reviews" in self.services:
plan.append({"service": "reviews", "fields": ["rating", "text"]})
if "orders" in query.lower() and "orders" in self.services:
plan.append({"service": "orders", "fields": ["id", "total"]})
return plan
def execute(self, query: str) -> Dict:
plan = self.plan_query(query)
results = {}
for step in plan:
svc = self.services.get(step["service"])
if svc:
results[step["service"]] = {
"fields": step["fields"],
"source": svc["url"],
}
return {"data": results, "plan": plan}
gateway = FederationGateway()
products_sdl = """
type Product @key(fields: "id") {
id: ID!
name: String!
price: Float!
}
extend type Query {
products: [Product!]!
}
"""
reviews_sdl = """
type Review @key(fields: "id") {
id: ID!
productId: ID!
rating: Int!
text: String!
}
extend type Query {
reviews(productId: ID!): [Review!]!
}
"""
gateway.add_service("products", "http://products-svc", 4001, products_sdl)
gateway.add_service("reviews", "http://reviews-svc", 4002, reviews_sdl)
schema = gateway.compose_schema()
print("Composed schema types:")
for type_name, services in schema.items():
print(f" {type_name}: from {', '.join(services)}")
query = "{ products { id name price } reviews(productId: \"1\") { rating text } }"
result = gateway.execute(query)
print(f"\nQuery plan: {len(result['plan'])} steps")
for step in result['plan']:
print(f" -> {step['service']}: {', '.join(step['fields'])}")
Expected output:
Composed schema types:
Product: from products
Query: from products, reviews
Review: from reviews
Query plan: 2 steps
-> products: id, name, price
-> reviews: rating, text
Rate Limiting for GraphQL
# graphql_ratelimit.py
import time
from typing import Any, Dict, Optional, Tuple
class GraphQLRateLimiter:
def __init__(self, max_depth: int = 5,
max_queries_per_min: int = 100):
self.max_depth = max_depth
self.max_queries_per_min = max_queries_per_min
self.client_queries: Dict[str, list] = {}
def calculate_depth(self, query: str) -> int:
depth = 0
current = 0
for char in query:
if char == "{":
current += 1
depth = max(depth, current)
elif char == "}":
current -= 1
return depth
def estimate_cost(self, query: str) -> int:
fields = query.count("{") + query.count("}") + query.count(" ")
return max(1, fields // 10)
def check(self, client_id: str, query: str) -> Tuple[bool, str]:
depth = self.calculate_depth(query)
if depth > self.max_depth:
return False, f"Query depth {depth} exceeds max {self.max_depth}"
now = time.time()
window_start = now - 60
self.client_queries.setdefault(client_id, [])
self.client_queries[client_id] = [
t for t in self.client_queries[client_id] if t > window_start
]
if len(self.client_queries[client_id]) >= self.max_queries_per_min:
return False, "Rate limit exceeded"
cost = self.estimate_cost(query)
self.client_queries[client_id].extend([now] * cost)
return True, "OK"
limiter = GraphQLRateLimiter(max_depth=3, max_queries_per_min=5)
queries = [
"{ products { id name } }",
"{ products { id name details { description } } }",
"{ products { id name details { specs { weight } } } }",
"{ products { id name details { specs { dimensions { width } } } } }",
]
for i, q in enumerate(queries):
allowed, reason = limiter.check("client_1", q)
print(f"Query {i+1} (depth {limiter.calculate_depth(q)}): {'OK' if allowed else reason}")
Expected output:
Query 1 (depth 2): OK
Query 2 (depth 3): OK
Query 3 (depth 4): OK
Query 4 (depth 5): DENIED
Common Mistakes
1. Ignoring Query Complexity
Without depth and complexity limits, clients can write expensive queries that overload backends. Always implement query cost analysis.
2. No Response Caching
GraphQL responses are harder to cache than REST because all queries hit the same endpoint. Use persisted queries for CDN caching.
3. Mixing Federation and Stitching
Apollo Federation and schema stitching are different approaches. Choose one and stick with it.
4. Not Handling N+1 Queries
Gateway resolvers that fetch data for each parent item cause N+1 database queries. Use DataLoader for batching.
5. Over-fetching from Backends
A gateway that fetches all fields from backends wastes bandwidth. Use field-aware resolvers that fetch only requested fields.
Practice Questions
1. What is the difference between schema federation and schema stitching?
Federation allows services to extend shared types across services. Stitching merges independent schemas into one. Federation is more decoupled.
2. How does query depth limiting protect GraphQL services?
Deeply nested queries can cause exponential database load. Depth limiting rejects queries exceeding a configurable nesting level.
3. What is Apollo Federation's @key directive?
The @key directive marks a field as the entity key, allowing services to contribute fields to a type owned by another service.
4. How does a GraphQL gateway handle REST data sources?
The gateway defines GraphQL types and resolvers that call REST APIs, transforming REST responses into GraphQL-compatible formats.
Challenge
Design a GraphQL gateway for a blogging platform with users, posts, and comments in separate services, using Apollo Federation, with query depth limiting of 5, and cost-based rate limiting.
FAQ
Mini Project: Simple GraphQL Gateway
# simple_gql_gateway.py
import json
from typing import Any, Dict, List, Optional
class SimpleGraphQLGateway:
def __init__(self):
self.resolvers: Dict[str, callable] = {}
def add_resolver(self, field: str, resolver):
self.resolvers[field] = resolver
def execute(self, query: str) -> Dict:
fields = self._parse_fields(query)
data = {}
for field in fields:
if field in self.resolvers:
data[field] = self.resolvers[field]()
return {"data": data}
def _parse_fields(self, query: str) -> List[str]:
fields = []
current = ""
depth = 0
for char in query:
if char == "{":
depth += 1
if depth == 1 and current.strip():
current = ""
elif char == "}":
depth -= 1
elif depth == 1 and char.isalpha():
current += char
elif depth == 1 and char in " \n" and current:
fields.append(current.strip())
current = ""
if current.strip():
fields.append(current.strip())
return fields
gw = SimpleGraphQLGateway()
def resolve_products():
return [{"id": 1, "name": "Widget", "price": 9.99}]
gw.add_resolver("products", resolve_products)
result = gw.execute("{ products { id name } }")
print(json.dumps(result, indent=2))
Expected output:
{
"data": {
"products": [
{
"id": 1,
"name": "Widget",
"price": 9.99
}
]
}
}
What's Next
You understand GraphQL gateway patterns. Next, learn about gateway monitoring, then explore the mini project.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro