Response Transformation at the API Gateway
In this tutorial, you'll learn about Response Transformation. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Response transformation modifies backend responses before returning them to clients, enabling consistent formatting, header management, and error standardization.
What You'll Learn
By the end of this lesson, you will implement response header modification, body rewriting, error standardization, and response aggregation patterns.
Why It Matters
Different backend services return different response formats. Response transformation ensures clients receive consistent, predictable responses regardless of which service handled the request.
Real-World Use
A gateway transforms all error responses to a standard RFC 7807 Problem Details format, adds CORS headers, wraps response bodies in a consistent envelope, and removes sensitive internal data.
Response Transformation Flow
flowchart LR
Backend[Backend] -->|Raw Response| GW[Gateway]
GW -->|Add CORS Headers| GW
GW -->|Standardize Error| GW
GW -->|Wrap Body| GW
GW -->|Transformed| Client[Client]
style GW fill:#22c55e,color:#fff
Response Header Transformation
# response_headers.py
from typing import Dict, List, Optional
class ResponseHeaderTransformer:
def __init__(self):
self.add_headers: Dict[str, str] = {}
self.remove_headers: set = set()
self.security_headers: Dict[str, str] = {
"X-Content-Type-Options": "nosniff",
"X-Frame-Options": "DENY",
"X-XSS-Protection": "1; mode=block",
"Strict-Transport-Security": "max-age=31536000",
}
def add(self, name: str, value: str):
self.add_headers[name] = value
def remove(self, name: str):
self.remove_headers.add(name.lower())
def apply_security_headers(self):
self.add_headers.update(self.security_headers)
def transform(self, response_headers: Dict[str, str]) -> Dict[str, str]:
result = {}
for key, value in response_headers.items():
if key.lower() not in self.remove_headers:
result[key] = value
result.update(self.add_headers)
return result
transformer = ResponseHeaderTransformer()
transformer.apply_security_headers()
transformer.remove("x-internal-trace")
transformer.add("X-Cache-Status", "MISS")
incoming = {
"Content-Type": "application/json",
"X-Internal-Trace": "trace-abc-123",
"Cache-Control": "private",
}
transformed = transformer.transform(incoming)
for key, value in sorted(transformed.items()):
print(f"{key}: {value}")
Expected output:
Cache-Control: private
Content-Type: application/json
Strict-Transport-Security: max-age=31536000
X-Cache-Status: MISS
X-Content-Type-Options: nosniff
X-Frame-Options: DENY
X-XSS-Protection: 1; mode=block
Error Standardization
# error_standardization.py
import json
from typing import Any, Dict, Optional
class ErrorStandardizer:
def standardize(self, status_code: int, body: Any) -> Dict:
if isinstance(body, dict) and "error" in body:
message = body["error"]
elif isinstance(body, str):
message = body
else:
message = "An error occurred"
return {
"type": "https://httpwg.org/specs/rfc7807.html",
"title": self._get_title(status_code),
"status": status_code,
"detail": message,
"instance": "/errors/request",
}
def _get_title(self, status_code: int) -> str:
titles = {
400: "Bad Request",
401: "Unauthorized",
403: "Forbidden",
404: "Not Found",
422: "Unprocessable Entity",
429: "Too Many Requests",
500: "Internal Server Error",
502: "Bad Gateway",
503: "Service Unavailable",
}
return titles.get(status_code, "Unknown Error")
def is_error(self, status_code: int) -> bool:
return status_code >= 400
standardizer = ErrorStandardizer()
errors = [
(400, {"error": "Invalid email format"}),
(404, "Resource not found"),
(500, {"message": "Database connection failed"}),
(429, {"error": "Rate limit exceeded", "retry_after": 60}),
]
for code, body in errors:
if standardizer.is_error(code):
standardized = standardizer.standardize(code, body)
print(f"HTTP {code}:")
print(json.dumps(standardized, indent=2))
print()
Expected output shows each error standardized to RFC 7807 format with appropriate title and detail.
Response Body Transformation
# response_body.py
import json
from typing import Any, Dict, Optional
class ResponseBodyTransformer:
def wrap_envelope(self, body: Any, status: int = 200) -> Dict:
return {
"success": status < 400,
"status": status,
"data": body,
"meta": {
"version": "1.0",
"timestamp": "2026-06-28T00:00:00Z",
},
}
def remove_fields(self, body: Dict, fields: list) -> Dict:
for field in fields:
body.pop(field, None)
return body
def paginate(self, body: list, page: int, limit: int, total: int) -> Dict:
return {
"data": body,
"pagination": {
"page": page,
"limit": limit,
"total": total,
"pages": (total + limit - 1) // limit,
},
}
transformer = ResponseBodyTransformer()
data = {"id": 1, "name": "Alice", "email": "alice@example.com", "ssn": "123-45-6789"}
data = transformer.remove_fields(data, ["ssn"])
enveloped = transformer.wrap_envelope(data, 200)
print(json.dumps(enveloped, indent=2))
Expected output:
{
"success": true,
"status": 200,
"data": {
"id": 1,
"name": "Alice",
"email": "alice@example.com"
},
"meta": {
"version": "1.0",
"timestamp": "2026-06-28T00:00:00Z"
}
}
Aggregation of Multiple Responses
# response_aggregation.py
import json
from typing import Any, Dict, List
class ResponseAggregator:
def aggregate(self, responses: List[Dict]) -> Dict:
combined = {}
errors = []
for response in responses:
service = response.get("service", "unknown")
if response.get("status", 500) >= 400:
errors.append({service: response.get("body", "Error")})
else:
combined[service] = response.get("body")
result = {"data": combined}
if errors:
result["errors"] = errors
return result
aggregator = ResponseAggregator()
responses = [
{"service": "users", "status": 200, "body": {"id": 1, "name": "Alice"}},
{"service": "posts", "status": 200, "body": [{"title": "Hello"}]},
{"service": "analytics", "status": 500, "body": "Service down"},
]
result = aggregator.aggregate(responses)
print(json.dumps(result, indent=2))
Expected output:
{
"data": {
"users": {"id": 1, "name": "Alice"},
"posts": [{"title": "Hello"}]
},
"errors": [
{"analytics": "Service down"}
]
}
Common Mistakes
1. Not Updating Content-Length
Transforming the response body changes its size. Always recalculate Content-Length before sending.
2. Overwriting Cache Headers
Adding Cache-Control headers that conflict with backend settings causes inconsistent Caching behavior.
3. Returning Internal Details in Errors
Raw error messages from backends may include stack traces or database queries. Always sanitize errors.
4. Ignoring CORS Headers
Clients cannot read responses if CORS headers are missing. Add CORS headers at the gateway for all cross-origin requests.
5. Breaking Streaming Responses
Response transformation that buffers the entire body breaks streaming responses. Use streaming-friendly transformations.
Practice Questions
1. Why should you standardize error responses at the gateway?
Clients can use a single error parser instead of handling different formats from each service, reducing client complexity.
2. What is a response envelope?
A consistent wrapper structure (usually success/data/meta) that every response follows, making client Parsing predictable.
3. How does the gateway remove sensitive fields from responses?
It intercepts the backend response, parses the body, removes specified fields (like passwords, SSNs), then forwards to the client.
4. Why add security headers at the gateway?
Security headers protect all clients regardless of which backend service handles the request, ensuring consistent security posture.
Challenge
Build a response transformation pipeline that adds security headers, wraps responses in a standard envelope, removes internal fields, and standardizes errors to RFC 7807 format.
FAQ
Mini Project: Response Pipeline
# response_pipeline.py
import json
from typing import Any, Dict, Optional
class ResponsePipeline:
def __init__(self):
self.transformers = []
def add_transformer(self, name: str, transform_fn):
self.transformers.append({"name": name, "fn": transform_fn})
def process(self, status: int, headers: Dict, body: Any) -> Dict:
for t in self.transformers:
if t["name"] == "headers":
headers = t["fn"](headers)
elif t["name"] == "body":
body = t["fn"](body, status)
elif t["name"] == "error":
if status >= 400:
body = t["fn"](status, body)
return {"status": status, "headers": headers, "body": body}
pipeline = ResponsePipeline()
def add_cors(headers):
headers["Access-Control-Allow-Origin"] = "*"
return headers
def wrap_body(body, status):
return {"success": status < 400, "status": status, "data": body}
def standardize_error(status, body):
return {"error": body if isinstance(body, str) else body.get("error", "Error"), "status": status}
pipeline.add_transformer("headers", add_cors)
pipeline.add_transformer("body", wrap_body)
pipeline.add_transformer("error", standardize_error)
result_ok = pipeline.process(200, {"Content-Type": "application/json"}, {"id": 1})
result_err = pipeline.process(404, {"Content-Type": "application/json"}, {"error": "Not found"})
print("Success response:", json.dumps(result_ok["body"], indent=2))
print("Error response:", json.dumps(result_err["body"], indent=2))
Expected output shows wrapped success body and standardized error body with CORS headers.
What's Next
You understand response transformation. Next, learn about rate limiting at the gateway, then explore authentication at the gateway.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro