APIs - REST, GraphQL, WebSockets, API Gateway, Authentication, Rate Limiting, and Versioning

APIs are the connective tissue of modern software. Every web application, every mobile app, every microservice, every third-party integration communicates through APIs. Understanding APIs, not just how to call them but how to design, secure, version, and operate them, is foundational to building production software.

This guide covers the full stack: what APIs are and why they exist, the three dominant API paradigms (REST, GraphQL, WebSockets), how to secure them, how to scale them, how to manage them as they evolve, and how AWS API Gateway fits into all of it.

---

Table of Contents

1. What an API Is - The Core Abstraction
2. HTTP Fundamentals - The Protocol APIs Speak
3. REST - Representational State Transfer
4. REST API Design - Principles for Production APIs
5. GraphQL - Query Language for APIs
6. WebSockets - Real-Time Bidirectional Communication
7. API Authentication and Authorisation
8. Rate Limiting and Throttling
9. API Versioning
10. AWS API Gateway - Managed API Infrastructure
11. API Gateway - REST APIs
12. API Gateway - HTTP APIs
13. API Gateway - WebSocket APIs
14. API Security Patterns on AWS
15. Observability - Monitoring APIs in Production
16. API Design Patterns and Anti-Patterns

---

What an API Is - The Core Abstraction

The Problem APIs Solve

Imagine you want to display the current weather in your application. You could build your own weather stations, collect sensor data, process it, and serve it. Or you could call a weather API, a service someone else built and operates, and get the data in milliseconds.

APIs are contracts. They define what a service can do and how to ask it to do things. The implementation behind the API is hidden. You do not need to know how the weather service processes sensor data. You need to know how to ask for the weather for a given location and what the response looks like.

This abstraction is why modern software is built the way it is:

• A mobile app calls the same API as the web app, the backend is shared
• Different internal services communicate through APIs, each team owns their service boundary
• Third-party functionality (payments, maps, identity) is consumed via API, you do not build what already exists

What an API Contract Defines

An API contract specifies:

• Endpoints - Where to send requests (URLs, addresses)
• Methods - What operations are available (read, create, update, delete)
• Request format - What to send (headers, parameters, body structure)
• Response format - What comes back (structure, data types, status codes)
• Authentication - How to prove who you are
• Error format - What failures look like and what they mean
• Rate limits - How often you can call the API
• Versioning - How changes to the API are communicated and managed

A well-designed API contract is stable, predictable, and honest about its limitations.

---

HTTP Fundamentals - The Protocol APIs Speak

Why HTTP

Most APIs, REST, GraphQL, and many others, communicate over HTTP (Hypertext Transfer Protocol). Understanding HTTP is understanding the foundation that APIs are built on.

Requests and Responses

Every HTTP interaction has a request (from client to server) and a response (from server to client).

A request contains:

• Method - The operation being requested
• URL - The resource being targeted
• Headers - Metadata about the request (content type, authentication, caching)
• Body - The data being sent (for POST, PUT, PATCH)

A response contains:

• Status code - A number indicating what happened
• Headers - Metadata about the response
• Body - The data being returned

HTTP Methods

Method	Purpose	Has Body	Idempotent	Safe
GET	Retrieve a resource	No	Yes	Yes
POST	Create a resource or trigger action	Yes	No	No
PUT	Replace a resource entirely	Yes	Yes	No
PATCH	Partially update a resource	Yes	No	No
DELETE	Remove a resource	No	Yes	No

Idempotent means making the same request multiple times produces the same result. GET /users/123 always returns user 123. DELETE /users/123, whether called once or ten times, results in user 123 not existing. POST /users called twice creates two users. Not idempotent.

Safe means the operation does not change state. GETs do not change anything.

HTTP Status Codes

Status codes communicate the outcome of the request. They are grouped by range:

Range	Category	Examples
2xx	Success	200 OK, 201 Created, 204 No Content
3xx	Redirection	301 Moved Permanently, 304 Not Modified
4xx	Client Error	400 Bad Request, 401 Unauthorized, 403 Forbidden, 404 Not Found, 429 Too Many Requests
5xx	Server Error	500 Internal Server Error, 502 Bad Gateway, 503 Service Unavailable

The status code communicates intent. A client that receives 429 knows to back off and retry later. A client that receives 404 knows the resource does not exist. A client that receives 500 knows the server failed, the request was valid.

Using status codes correctly is not optional. Returning 200 with an error body is one of the most common API design mistakes. It forces clients to inspect the body to know if the request succeeded, a subtle, expensive ambiguity.

Headers

Headers are key-value metadata attached to requests and responses.

Common request headers:

Content-Type: application/json          # Format of the request body
Accept: application/json                # Format the client can receive
Authorization: Bearer eyJhbGciOiJSUzI  # Authentication credential
X-Request-ID: f47ac10b-58cc-4372       # Unique ID for tracing
User-Agent: MyApp/1.0.0                 # Client identification

Common response headers:

Content-Type: application/json
Cache-Control: max-age=300, must-revalidate
X-Request-ID: f47ac10b-58cc-4372       # Echo back for tracing
X-RateLimit-Limit: 1000                # Rate limit ceiling
X-RateLimit-Remaining: 987             # Requests remaining in window
X-RateLimit-Reset: 1713312000          # When the window resets (Unix timestamp)

---

REST - Representational State Transfer

What REST Is

REST is an architectural style, not a protocol, not a standard, not a specification. It is a set of constraints that, when followed, produce APIs that are scalable, stateless, and easy to understand.

Roy Fielding defined REST in his doctoral dissertation in 2000. The six constraints are:

1. Client-Server - The client and server are separate. The client does not know how the server stores data. The server does not know what the client does with the data.
2. Stateless - Every request contains all information needed to process it. The server does not store client session state between requests. Authentication credentials are sent with every request.
3. Cacheable - Responses must indicate whether they can be cached. Caching reduces load and improves performance.
4. Uniform Interface - The API interface is consistent. Resources are identified by URLs. Resources are manipulated through representations (JSON, XML). Messages are self-descriptive.
5. Layered System - The client cannot tell whether it is connected directly to the server or to an intermediary (load balancer, cache, API gateway). Each layer only knows about the layer it interacts with.
6. Code on Demand (optional) - Servers can send executable code to clients (JavaScript in browsers).

Most APIs called "REST APIs" implement constraints 1-4 and call it REST. That is fine for practical purposes.

Resources and Representations

In REST, everything is a resource. A user is a resource. An order is a resource. A payment is a resource. Resources are identified by URLs.

A representation is the data format used to transfer the resource state, typically JSON.

Resource:       User
Identifier:     /users/88721
Representation: {"id": 88721, "name": "Kachi", "email": "[e-mail]"}

Operations on resources use HTTP methods:

GET    /users           # List all users
POST   /users           # Create a user
GET    /users/88721     # Get user 88721
PUT    /users/88721     # Replace user 88721 entirely
PATCH  /users/88721     # Partially update user 88721
DELETE /users/88721     # Delete user 88721

GET    /users/88721/orders        # Orders belonging to user 88721
POST   /users/88721/orders        # Create an order for user 88721
GET    /users/88721/orders/456    # Specific order 456 for user 88721

This is the core of REST: URLs are nouns (resources), methods are verbs (actions).

---

REST API Design - Principles for Production APIs

URL Design

Use nouns, not verbs:

Good:  GET /users
Bad:   GET /getUsers

Good:  POST /orders
Bad:   POST /createOrder

Use plural nouns consistently:

Good:  /users, /orders, /products
Bad:   /user, /order, /product  (inconsistent)

Use hierarchy for relationships:

/users/{userId}/orders              # Orders for a user
/orders/{orderId}/items             # Items in an order
/organisations/{orgId}/members      # Members of an organisation

Avoid deep nesting - maximum two levels:

Good:  GET /orders/{orderId}/items/{itemId}
Bad:   GET /users/{userId}/orders/{orderId}/items/{itemId}/reviews

Beyond two levels, the URL becomes hard to read and the API becomes tightly coupled to the data model. Provide shortcuts:

GET /items/{itemId}           # Direct access, not buried in the hierarchy
GET /items/{itemId}/reviews   # Related resource

Request and Response Design

Always return the full resource on create and update:

POST /users
Content-Type: application/json

{
  "name": "Kachi",
  "email": "[e-mail]"
}

HTTP/1.1 201 Created
Content-Type: application/json
Location: /users/88721

{
  "id": 88721,
  "name": "Kachi",
  "email": "[e-mail]",
  "createdAt": "2024-04-16T09:00:00Z",
  "updatedAt": "2024-04-16T09:00:00Z"
}

The client should not have to make a second GET request to see the resource it just created.

Use consistent error format:

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Request validation failed",
    "requestId": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
    "details": [
      {
        "field": "email",
        "code": "INVALID_FORMAT",
        "message": "email must be a valid email address"
      },
      {
        "field": "name",
        "code": "REQUIRED",
        "message": "name is required"
      }
    ]
  }
}

Every error response includes:

• A machine-readable error code (not just a human message)
• A human-readable message
• A request ID for support and debugging
• Field-level detail where validation fails

Pagination for collections:

GET /orders?limit=20&cursor=eyJpZCI6MTAwfQ

HTTP/1.1 200 OK
Content-Type: application/json

{
  "data": [
    {"id": 101, ...},
    {"id": 102, ...}
  ],
  "pagination": {
    "limit": 20,
    "nextCursor": "eyJpZCI6MTIwfQ",
    "hasMore": true,
    "total": 5432
  }
}

Cursor pagination (as above) is preferred over offset pagination for large datasets. Cursor-based pagination is stable, it does not skip or repeat items when data is inserted or deleted during pagination. Offset pagination (?page=3&limit=20) is simpler but breaks when the underlying data changes between page requests.

Filtering and sorting:

# Filter
GET /orders?status=pending&userId=88721

# Sorting
GET /orders?sort=createdAt:desc

# Field selection (reduce payload size)
GET /users?fields=id,name,email

# Search
GET /products?q=wireless+headphones&category=electronics

---

GraphQL - Query Language for APIs

The Problem REST Does Not Solve Well

You are building a mobile application. On the home screen, you need:

• User's name and avatar
• Their last 5 orders, with product names and totals
• Any active promotions

With REST, this requires three API calls: GET /users/me, GET /users/me/orders?limit=5, GET /promotions?active=true. Three round trips. Each returns more data than you need, the orders endpoint returns the full order including shipping address and line item details, but you only need product names and totals.

Or, you build a purpose-built endpoint: GET /home-screen-data. But now you have an endpoint that is tightly coupled to one specific screen. Every time the home screen changes, the API changes. Every new screen needs a new endpoint.

This is the n+1 request problem and over-fetching/under-fetching, the structural limitations of REST for complex, client-driven data requirements.

What GraphQL Is

GraphQL is a query language for APIs, developed by Facebook and open-sourced in 2015. Instead of fixed endpoints that return fixed data, GraphQL exposes a single endpoint and lets clients specify exactly what data they need.

The client writes a query describing the shape of the data it wants. The server returns exactly that data, nothing more, nothing less.

The Schema

A GraphQL API is defined by its schema, a strongly typed definition of all available data and operations.

# Schema definition
type User {
  id: ID!
  name: String!
  email: String!
  avatar: String
  orders(limit: Int = 10, status: OrderStatus): [Order!]!
  createdAt: DateTime!
}

type Order {
  id: ID!
  status: OrderStatus!
  total: Float!
  items: [OrderItem!]!
  shippingAddress: Address!
  createdAt: DateTime!
}

type OrderItem {
  id: ID!
  product: Product!
  quantity: Int!
  price: Float!
}

type Product {
  id: ID!
  name: String!
  price: Float!
  category: String!
}

type Promotion {
  id: ID!
  title: String!
  discountPercent: Float!
  expiresAt: DateTime!
}

enum OrderStatus {
  PENDING
  PROCESSING
  SHIPPED
  DELIVERED
  CANCELLED
}

# Operations
type Query {
  me: User
  user(id: ID!): User
  orders(userId: ID!, limit: Int, status: OrderStatus): [Order!]!
  promotions(active: Boolean): [Promotion!]!
}

type Mutation {
  createOrder(input: CreateOrderInput!): Order!
  updateOrderStatus(orderId: ID!, status: OrderStatus!): Order!
  cancelOrder(orderId: ID!): Order!
}

type Subscription {
  orderStatusChanged(orderId: ID!): Order!
}

Queries

# Get exactly what the home screen needs, one request, one response
query HomeScreenData {
  me {
    name
    avatar
  }
  myOrders: me {
    orders(limit: 5, status: PROCESSING) {
      id
      total
      status
      items {
        product {
          name
        }
        quantity
      }
    }
  }
  promotions(active: true) {
    id
    title
    discountPercent
    expiresAt
  }
}

Response:

{
  "data": {
    "me": {
      "name": "Kachi",
      "avatar": "https://cdn.example.com/avatars/88721.jpg"
    },
    "myOrders": {
      "orders": [
        {
          "id": "order-456",
          "total": 4250.00,
          "status": "PROCESSING",
          "items": [
            {
              "product": {"name": "Wireless Headphones"},
              "quantity": 2
            }
          ]
        }
      ]
    },
    "promotions": [
      {
        "id": "promo-789",
        "title": "10% off electronics",
        "discountPercent": 10,
        "expiresAt": "2024-04-30T23:59:59Z"
      }
    ]
  }
}

One request. Exactly the data needed. No over-fetching.

Mutations

mutation CreateOrder($input: CreateOrderInput!) {
  createOrder(input: $input) {
    id
    status
    total
    createdAt
  }
}

Variables sent with the mutation:

{
  "input": {
    "items": [
      {"productId": "prod-123", "quantity": 2}
    ],
    "shippingAddressId": "addr-456"
  }
}

Subscriptions

GraphQL subscriptions handle real-time data over WebSockets:

subscription TrackOrder($orderId: ID!) {
  orderStatusChanged(orderId: $orderId) {
    id
    status
    updatedAt
  }
}

When the order's status changes on the server, the subscription pushes the update to all connected clients watching that order.

GraphQL on AWS - AppSync

AWS AppSync is a managed GraphQL service. You define your schema, connect resolvers to data sources (DynamoDB, RDS, Lambda, OpenSearch, HTTP endpoints), and AppSync handles the GraphQL execution engine, WebSocket connections for subscriptions, and caching.

# AppSync resolver mapping template (VTL for DynamoDB)
# Request mapping
{
  "version": "2017-02-28",
  "operation": "GetItem",
  "key": {
    "id": $util.dynamodb.toDynamoDBJson($ctx.args.id)
  }
}

# Response mapping
$util.toJson($ctx.result)

REST vs GraphQL - When to Use Which

Consideration	REST	GraphQL
Multiple clients with different data needs	Poor fit	Excellent fit
Simple CRUD operations	Excellent fit	Overkill
Public API for third-party developers	Standard and well-understood	Steeper learning curve
Real-time data	Via WebSockets (separate)	Built-in subscriptions
Caching	Simple (HTTP cache headers)	More complex (per-field caching)
Error handling	HTTP status codes	Always 200, errors in response body
Documentation	OpenAPI/Swagger	Self-documenting (introspection)
File uploads	Native multipart	Requires workarounds

---

WebSockets - Real-Time Bidirectional Communication

The Problem HTTP Cannot Solve

HTTP is request-response. The client sends a request. The server responds. The server cannot push data to the client without the client requesting it.

For real-time use cases, live notifications, collaborative editing, live chat, real-time dashboards, multiplayer games, live market data, the client cannot poll often enough to feel real-time without enormous overhead. A client polling every second for a chat message that arrives once a minute makes 3,598 unnecessary requests per hour, per client.

WebSockets solve this by establishing a persistent, bidirectional connection. After the initial HTTP handshake, the connection upgrades to WebSocket protocol. Either the client or the server can send messages at any time. No polling. No repeated handshakes.

The WebSocket Handshake

# Client request to upgrade
GET /ws/chat HTTP/1.1
Host: api.myapp.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13

# Server response confirming upgrade
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

After the handshake, the TCP connection stays open. Both sides send frames, binary or text at any time.

WebSocket Server Implementation

# Python with FastAPI and WebSockets
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict, Set
import json
import asyncio

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        # Room -> set of WebSocket connections
        self.rooms: Dict[str, Set[WebSocket]] = {}

    async def connect(self, websocket: WebSocket, room_id: str):
        await websocket.accept()
        if room_id not in self.rooms:
            self.rooms[room_id] = set()
        self.rooms[room_id].add(websocket)

    def disconnect(self, websocket: WebSocket, room_id: str):
        if room_id in self.rooms:
            self.rooms[room_id].discard(websocket)
            if not self.rooms[room_id]:
                del self.rooms[room_id]

    async def broadcast_to_room(self, room_id: str, message: dict, sender: WebSocket = None):
        if room_id not in self.rooms:
            return

        message_str = json.dumps(message)
        disconnected = set()

        for connection in self.rooms[room_id]:
            if connection == sender:
                continue  # Don't echo back to sender
            try:
                await connection.send_text(message_str)
            except Exception:
                disconnected.add(connection)

        # Clean up disconnected clients
        for conn in disconnected:
            self.rooms[room_id].discard(conn)

manager = ConnectionManager()

@app.websocket("/ws/chat/{room_id}")
async def chat_endpoint(websocket: WebSocket, room_id: str, token: str = None):
    # Authenticate the connection
    user = await authenticate_token(token)
    if not user:
        await websocket.close(code=4001, reason="Unauthorised")
        return

    await manager.connect(websocket, room_id)

    # Notify room that user joined
    await manager.broadcast_to_room(room_id, {
        "type": "user_joined",
        "userId": user.id,
        "userName": user.name,
        "timestamp": datetime.utcnow().isoformat()
    })

    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)

            # Validate and process message
            if message.get("type") == "chat_message":
                await manager.broadcast_to_room(room_id, {
                    "type": "chat_message",
                    "userId": user.id,
                    "userName": user.name,
                    "content": message["content"][:1000],  # Truncate to prevent abuse
                    "timestamp": datetime.utcnow().isoformat()
                }, sender=websocket)

    except WebSocketDisconnect:
        manager.disconnect(websocket, room_id)
        await manager.broadcast_to_room(room_id, {
            "type": "user_left",
            "userId": user.id,
            "userName": user.name
        })

WebSocket Client

// Browser WebSocket client
const ws = new WebSocket('wss://api.myapp.com/ws/chat/room-456?token=eyJhbGci...');

ws.onopen = () => {
    console.log('Connected to chat room');
};

ws.onmessage = (event) => {
    const message = JSON.parse(event.data);

    switch(message.type) {
        case 'chat_message':
            displayMessage(message);
            break;
        case 'user_joined':
            showNotification(`${message.userName} joined the room`);
            break;
        case 'user_left':
            showNotification(`${message.userName} left the room`);
            break;
    }
};

ws.onerror = (error) => {
    console.error('WebSocket error:', error);
};

ws.onclose = (event) => {
    console.log('Connection closed:', event.code, event.reason);
    if (event.code !== 1000) {
        // Abnormal closure — attempt reconnect with exponential backoff
        reconnect();
    }
};

// Send a message
function sendMessage(content) {
    if (ws.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify({
            type: 'chat_message',
            content: content
        }));
    }
}

// Reconnection with exponential backoff
let reconnectDelay = 1000;
function reconnect() {
    setTimeout(() => {
        reconnectDelay = Math.min(reconnectDelay * 2, 30000); // Max 30 seconds
        // Re-establish connection
    }, reconnectDelay);
}

When to Use WebSockets vs REST vs GraphQL Subscriptions

Use Case	Solution
Live chat	WebSockets
Real-time collaborative editing	WebSockets
Live notifications (low frequency)	Server-Sent Events or polling
Real-time order tracking	WebSockets or GraphQL Subscriptions
Live market data	WebSockets
Multiplayer games	WebSockets
Standard CRUD operations	REST
Complex data fetching with multiple entities	GraphQL

---

API Authentication and Authorisation

Authentication vs Authorisation

These two concepts are consistently conflated. They are not the same.

Authentication - Proves identity. Who are you?
Authorisation - Checks permissions. What are you allowed to do?

Both are required. Authentication without authorisation means everyone who proves who they are can do everything. Authorisation without authentication means you make decisions about permissions without knowing who you are talking to.

API Keys

The simplest authentication mechanism. A long, random string that identifies the caller.

GET /v1/data
X-API-Key: ak_live_8f73h2kd83jf9s02mxn71qs

# Server-side validation
import secrets
import hashlib

def validate_api_key(provided_key: str) -> Optional[APIKey]:
    # Hash the provided key to compare against stored hash
    # Never store raw API keys — store hashes
    key_hash = hashlib.sha256(provided_key.encode()).hexdigest()

    api_key = db.query(APIKey).filter(
        APIKey.key_hash == key_hash,
        APIKey.is_active == True,
        APIKey.expires_at > datetime.utcnow()
    ).first()

    return api_key

# Generating a new API key
def generate_api_key(user_id: int, name: str) -> tuple[str, APIKey]:
    raw_key = f"ak_live_{secrets.token_urlsafe(32)}"
    key_hash = hashlib.sha256(raw_key.encode()).hexdigest()

    api_key = APIKey(
        user_id=user_id,
        name=name,
        key_hash=key_hash,  # Only store the hash
        prefix=raw_key[:12],  # Store prefix for identification
        created_at=datetime.utcnow(),
        expires_at=datetime.utcnow() + timedelta(days=365)
    )

    db.add(api_key)
    db.commit()

    return raw_key, api_key  # Return the raw key once - never again

API keys are simple and good for machine-to-machine authentication where the client is trusted. They are not good for user authentication, if a key is leaked, there is no way to cryptographically invalidate it without revoking and reissuing it.

JWT - JSON Web Tokens

JWT is a compact, self-contained token format used for authentication and information exchange. A JWT is signed by the server and carries claims, information about the user and their permissions, without requiring a database lookup on every request.

Structure: Three base64-encoded parts joined by dots.

header.payload.signature

eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiI4ODcyMSIsIm5hbWUiOiJLYWNoaSIsImlhdCI6MTcxMzMwMDAwMCwiZXhwIjoxNzEzMzg2NDAwLCJyb2xlcyI6WyJ1c2VyIl19.signature

Decoded payload:

{
  "sub": "88721",           // Subject (user ID)
  "name": "Kachi",
  "email": "[e-mail]",
  "roles": ["user"],
  "iat": 1713300000,        // Issued at (Unix timestamp)
  "exp": 1713386400,        // Expires at (Unix timestamp)
  "iss": "https://auth.myapp.com"  // Issuer
}

How JWT authentication works:

1. User logs in with username and password
2. Server validates credentials
3. Server creates a JWT, signs it with its private key
4. Server returns the JWT to the client
5. Client stores the JWT (memory, localStorage, httpOnly cookie)
6. Client sends the JWT with every subsequent request:
   Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
7. Server validates the JWT signature using its public key
8. Server reads the claims from the payload - no database lookup needed
9. Server authorises the request based on the claims

import jwt
from datetime import datetime, timedelta
from cryptography.hazmat.primitives import serialization

# Sign a JWT with RSA private key
def create_access_token(user: User) -> str:
    payload = {
        "sub": str(user.id),
        "name": user.name,
        "email": user.email,
        "roles": user.roles,
        "iat": datetime.utcnow(),
        "exp": datetime.utcnow() + timedelta(hours=1),
        "iss": "https://auth.myapp.com"
    }

    return jwt.encode(
        payload,
        PRIVATE_KEY,
        algorithm="RS256"
    )

# Validate a JWT
def validate_access_token(token: str) -> dict:
    try:
        payload = jwt.decode(
            token,
            PUBLIC_KEY,
            algorithms=["RS256"],
            options={"require": ["exp", "iat", "sub", "iss"]}
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise AuthError("Token has expired")
    except jwt.InvalidTokenError as e:
        raise AuthError(f"Invalid token: {e}")

# FastAPI middleware
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer

security = HTTPBearer()

async def get_current_user(credentials = Depends(security)) -> dict:
    try:
        payload = validate_access_token(credentials.credentials)
        return payload
    except AuthError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=str(e),
            headers={"WWW-Authenticate": "Bearer"}
        )

@app.get("/api/profile")
async def get_profile(user = Depends(get_current_user)):
    return {"userId": user["sub"], "name": user["name"]}

JWT security considerations:

• Never put sensitive data in the payload. The payload is base64-encoded, not encrypted. Anyone with the token can read the payload.
• Use short expiry times. 15-60 minutes for access tokens. JWTs cannot be individually revoked, you must wait for them to expire or maintain a revocation list.
• Use refresh tokens. A long-lived refresh token (stored securely) is used to obtain new short-lived access tokens without re-authentication.
• Sign with asymmetric keys (RS256, ES256), not symmetric (HS256), for multi-service architectures. Services can verify the signature using the public key without needing access to the private key.

OAuth 2.0 and OIDC

OAuth 2.0 is an authorisation framework for delegated access. OIDC (OpenID Connect) is an authentication layer on top of OAuth 2.0.

The pattern: instead of building your own identity system, delegate to an identity provider (Google, GitHub, Auth0, AWS Cognito). The user authenticates with the identity provider. The identity provider issues tokens. Your API validates those tokens.

# AWS Cognito with API Gateway - Cognito Authoriser
# The API Gateway validates the Cognito JWT before the request
# reaches your Lambda function. No auth code needed in the function.

# Your Lambda receives a pre-validated event with user identity
def handler(event, context):
    # User identity extracted from the validated JWT
    user_id = event['requestContext']['authorizer']['claims']['sub']
    user_email = event['requestContext']['authorizer']['claims']['email']
    user_groups = event['requestContext']['authorizer']['claims'].get('cognito:groups', [])

    # Authorise based on groups
    if 'admin' not in user_groups:
        return {
            'statusCode': 403,
            'body': json.dumps({'error': 'Forbidden'})
        }

    # Process the request
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Authorised'})
    }

mTLS - Mutual TLS

For machine-to-machine authentication where the highest security is required, between internal services, with financial partners, in regulated industries, mTLS (mutual TLS) provides cryptographic identity for both sides of the connection.

In standard TLS, the client verifies the server's certificate. In mTLS, both sides present certificates. The server verifies the client's certificate. If the client presents no certificate or an untrusted one, the connection is rejected before any data is exchanged.

API Gateway and AWS services support mTLS. It is configured at the certificate layer, not at the application layer.

---

Rate Limiting and Throttling

Why Rate Limiting Is Not Optional

Without rate limiting, a single misbehaving client, a buggy retry loop, an aggressive crawler, a deliberate denial-of-service attack, can consume all of your API's capacity and degrade service for everyone else.

Rate limiting protects availability. It also enables fair use enforcement, cost control, and tiered API products (free tier: 100 requests/day, paid tier: 10,000 requests/day).

Rate Limiting Algorithms

Fixed Window - Count requests in a fixed time window (e.g., the current minute). Reset the counter at the top of each window.

Problem: a burst at the end of one window and the start of the next can double the effective rate. 100 requests at 11:59 + 100 requests at 12:00 = 200 requests in 2 seconds.

Sliding Window - Count requests in the last N seconds, not the current fixed window. More accurate, prevents the boundary burst problem. Higher memory cost.

Token Bucket - A bucket with a fixed capacity fills with tokens at a constant rate. Each request consumes one token. If the bucket is empty, the request is rejected. Allows bursts up to the bucket capacity, but sustained rate is limited by the fill rate.

Leaky Bucket - Requests enter a queue (the bucket). The queue is processed at a constant rate. If the queue is full, new requests are dropped. Smooths out bursts. Good for protecting downstream systems from spikes.

Implementation

import time
import redis
from functools import wraps

# Sliding window rate limiting with Redis
class RateLimiter:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def is_allowed(
        self,
        key: str,
        limit: int,
        window_seconds: int
    ) -> tuple[bool, dict]:
        now = time.time()
        window_start = now - window_seconds

        pipe = self.redis.pipeline()

        # Remove requests outside the window
        pipe.zremrangebyscore(key, '-inf', window_start)

        # Count requests in window
        pipe.zcard(key)

        # Add current request
        pipe.zadd(key, {str(now): now})

        # Set expiry on the key
        pipe.expire(key, window_seconds)

        results = pipe.execute()
        request_count = results[1]

        # Reset time is when the oldest request in the window expires
        oldest = self.redis.zrange(key, 0, 0, withscores=True)
        reset_time = (oldest[0][1] + window_seconds) if oldest else now + window_seconds

        remaining = max(0, limit - request_count - 1)
        allowed = request_count < limit

        return allowed, {
            "limit": limit,
            "remaining": remaining,
            "reset": int(reset_time),
            "retry_after": max(0, int(reset_time - now)) if not allowed else None
        }


# FastAPI rate limiting middleware
from fastapi import Request, Response

rate_limiter = RateLimiter(redis.Redis(host='localhost', port=6379))

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # Identify the client
    api_key = request.headers.get("X-API-Key")
    client_ip = request.client.host

    identifier = f"api_key:{api_key}" if api_key else f"ip:{client_ip}"

    # Apply different limits based on tier
    if api_key:
        tier = await get_api_key_tier(api_key)
        limit = 10000 if tier == "premium" else 1000
    else:
        limit = 100  # Unauthenticated

    allowed, rate_info = rate_limiter.is_allowed(
        key=f"rate_limit:{identifier}",
        limit=limit,
        window_seconds=3600  # 1 hour window
    )

    if not allowed:
        return Response(
            content=json.dumps({
                "error": {
                    "code": "RATE_LIMIT_EXCEEDED",
                    "message": "Too many requests",
                    "retryAfter": rate_info["retry_after"]
                }
            }),
            status_code=429,
            headers={
                "X-RateLimit-Limit": str(rate_info["limit"]),
                "X-RateLimit-Remaining": "0",
                "X-RateLimit-Reset": str(rate_info["reset"]),
                "Retry-After": str(rate_info["retry_after"]),
                "Content-Type": "application/json"
            }
        )

    response = await call_next(request)

    # Add rate limit headers to all responses
    response.headers["X-RateLimit-Limit"] = str(rate_info["limit"])
    response.headers["X-RateLimit-Remaining"] = str(rate_info["remaining"])
    response.headers["X-RateLimit-Reset"] = str(rate_info["reset"])

    return response

Rate Limiting Headers

Always return rate limit information in response headers. Clients need to know their limits to implement backoff correctly:

X-RateLimit-Limit: 1000        # Maximum requests in the window
X-RateLimit-Remaining: 987     # Requests remaining in the current window
X-RateLimit-Reset: 1713312000  # Unix timestamp when the window resets
Retry-After: 3600              # Seconds to wait (only on 429 response)

A well-behaved client reads these headers and slows down before hitting the limit, rather than hammering the API until it gets a 429.

---

API Versioning

The Problem

Your API has customers. They have integrated with it. Their code works. You need to change something that breaks their code, a field is renamed, a response structure changes, a behaviour is modified. You cannot change the API without breaking existing integrations, and you cannot refuse to change the API forever.

Versioning is how you manage change without breaking the contract.

Versioning Strategies

URL path versioning - The version number is in the URL path.

https://api.myapp.com/v1/users
https://api.myapp.com/v2/users

Explicit, easy to route, easy to test, easy to document. Most common approach. The downside: URLs are supposed to identify resources, not versions of APIs. Purists object. Pragmatists use URL versioning anyway.

Query parameter versioning - The version is a query parameter.

https://api.myapp.com/users?version=2

Flexible. Easier to add optional versioning to existing routes. Harder to route at the infrastructure level (load balancers, API Gateway routing). Not commonly used.

Header versioning - The version is in a request header.

GET /users HTTP/1.1
Accept: application/vnd.myapp.v2+json

or

GET /users HTTP/1.1
API-Version: 2

Clean URLs. Follows the HTTP spec more precisely (content negotiation). Harder to test in a browser. Not visible in URLs shared in documentation or support tickets. Used by GitHub (via Accept header) and Stripe (via Stripe-Version header).

Versioning in Practice

Regardless of the strategy, versioning requires discipline:

Never break a published version. A published version is a promise. Adding new optional fields to a response is fine (backwards-compatible). Removing fields, changing field types, or changing behaviour is a breaking change, it requires a new version.

Run multiple versions simultaneously. v1 and v2 run in parallel. Clients migrate to v2 at their own pace. You deprecate v1 on a timeline, communicate the deprecation clearly, and eventually sunset it.

Add deprecation headers. When a version is deprecated:

Deprecation: Tue, 31 Dec 2024 23:59:59 GMT
Sunset: Tue, 31 Mar 2025 23:59:59 GMT
Link: <https://docs.myapp.com/api/migration-v1-v2>; rel="deprecation"

Version the data contract, not the implementation. v1 and v2 can share the same backend implementation. An adapter layer transforms v1 requests to internal format and v1 responses to v1 contract. v2 uses a different adapter. The business logic is shared.

# API versioning with FastAPI
from fastapi import APIRouter

v1_router = APIRouter(prefix="/v1")
v2_router = APIRouter(prefix="/v2")

@v1_router.get("/users/{user_id}")
async def get_user_v1(user_id: int):
    user = await get_user_from_db(user_id)
    # v1 contract: name is "full_name"
    return {
        "id": user.id,
        "full_name": user.name,        # Old field name
        "email": user.email
    }

@v2_router.get("/users/{user_id}")
async def get_user_v2(user_id: int):
    user = await get_user_from_db(user_id)
    # v2 contract: "full_name" is now "name", "displayName" is added
    return {
        "id": user.id,
        "name": user.name,             # New field name
        "displayName": user.display_name,  # New field in v2
        "email": user.email,
        "createdAt": user.created_at.isoformat()
    }

app.include_router(v1_router, deprecated=True)
app.include_router(v2_router)

---

AWS API Gateway - Managed API Infrastructure

What It Is

AWS API Gateway is a fully managed service for creating, publishing, maintaining, monitoring, and securing APIs. It handles the infrastructure layer — routing, throttling, authentication, caching, logging, so your backend only needs to handle business logic.

API Gateway sits between your clients and your backend. Every request goes through API Gateway, which validates, transforms, and routes it to the appropriate backend (Lambda, ECS, EC2, HTTP endpoint).

The Three API Types

API Gateway offers three distinct products:

	REST API	HTTP API	WebSocket API
Protocol	HTTP/HTTPS	HTTP/HTTPS	WebSocket
Latency	~6ms	~1ms	N/A
Cost	~$3.50/million requests	~$1.00/million requests	~$0.80/million messages
Features	Full (caching, request validation, usage plans)	Lightweight	Real-time bi-directional
Authorisers	Lambda, Cognito, IAM	JWT (Cognito, Auth0), Lambda	Lambda
Best for	Complex APIs needing full feature set	Microservices, Lambda proxies	Real-time applications

---

API Gateway - REST APIs

REST APIs in API Gateway provide the full feature set. Use them when you need request validation, response transformation, API keys and usage plans, caching, or request/response mapping templates.

import boto3
import json

apigateway = boto3.client('apigateway', region_name='us-east-1')

# Create a REST API
api = apigateway.create_rest_api(
    name='MyProductionAPI',
    description='Production API for MyApp',
    endpointConfiguration={'types': ['REGIONAL']}
)

API_ID = api['id']

# Get the root resource
root = apigateway.get_resources(restApiId=API_ID)
ROOT_ID = root['items'][0]['id']

# Create /users resource
users_resource = apigateway.create_resource(
    restApiId=API_ID,
    parentId=ROOT_ID,
    pathPart='users'
)

USERS_ID = users_resource['id']

# Create /users/{userId} resource
user_resource = apigateway.create_resource(
    restApiId=API_ID,
    parentId=USERS_ID,
    pathPart='{userId}'
)

USER_ID = user_resource['id']

# Create GET /users/{userId} method with Cognito authoriser
apigateway.put_method(
    restApiId=API_ID,
    resourceId=USER_ID,
    httpMethod='GET',
    authorizationType='COGNITO_USER_POOLS',
    authorizerId='abc123cognito',
    requestParameters={
        'method.request.path.userId': True  # Required path parameter
    }
)

# Integrate with Lambda
lambda_arn = 'arn:aws:lambda:us-east-1:123456789012:function:get-user'

apigateway.put_integration(
    restApiId=API_ID,
    resourceId=USER_ID,
    httpMethod='GET',
    type='AWS_PROXY',   # Lambda proxy — pass everything to Lambda
    integrationHttpMethod='POST',  # API Gateway always uses POST to invoke Lambda
    uri=f'arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/{lambda_arn}/invocations'
)

# Enable request validation
validator = apigateway.create_request_validator(
    restApiId=API_ID,
    name='validate-body-and-params',
    validateRequestBody=True,
    validateRequestParameters=True
)

# Add a usage plan with throttling and quotas
usage_plan = apigateway.create_usage_plan(
    name='standard-tier',
    description='Standard tier — 1000 req/hour, 10000 req/day',
    throttle={
        'burstLimit': 200,    # Tokens in the bucket at start
        'rateLimit': 100.0    # Requests per second sustained
    },
    quota={
        'limit': 10000,
        'period': 'DAY'
    }
)

# Deploy the API
deployment = apigateway.create_deployment(
    restApiId=API_ID,
    stageName='production',
    stageDescription='Production stage',
    description='Initial deployment'
)

# Create and configure stage
apigateway.update_stage(
    restApiId=API_ID,
    stageName='production',
    patchOperations=[
        {
            'op': 'replace',
            'path': '/defaultRouteSettings/loggingLevel',
            'value': 'INFO'
        },
        {
            'op': 'replace',
            'path': '/*/*/caching/enabled',
            'value': 'true'
        },
        {
            'op': 'replace',
            'path': '/*/*/caching/ttlInSeconds',
            'value': '300'
        }
    ]
)

Lambda Proxy Integration

In Lambda proxy integration, API Gateway passes the entire HTTP request to Lambda as an event and returns whatever Lambda returns as the HTTP response. Your Lambda function handles all request parsing and response formatting.

# Lambda function for GET /users/{userId}
import json
import boto3

def handler(event, context):
    # API Gateway provides the full request context
    user_id = event['pathParameters']['userId']
    user_identity = event['requestContext']['authorizer']['claims']['sub']

    # Ensure user can only access their own data
    if user_id != user_identity and 'admin' not in \
       event['requestContext']['authorizer']['claims'].get('cognito:groups', []):
        return {
            'statusCode': 403,
            'headers': {
                'Content-Type': 'application/json',
                'X-Request-ID': event['requestContext']['requestId']
            },
            'body': json.dumps({
                'error': {
                    'code': 'FORBIDDEN',
                    'message': 'You can only access your own profile'
                }
            })
        }

    # Fetch user from database
    user = get_user_from_db(user_id)

    if not user:
        return {
            'statusCode': 404,
            'headers': {'Content-Type': 'application/json'},
            'body': json.dumps({
                'error': {
                    'code': 'NOT_FOUND',
                    'message': f'User {user_id} not found'
                }
            })
        }

    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',
            'Cache-Control': 'private, max-age=60',
            'X-Request-ID': event['requestContext']['requestId']
        },
        'body': json.dumps({
            'id': user['id'],
            'name': user['name'],
            'email': user['email']
        })
    }

---

API Gateway - HTTP APIs

HTTP APIs are the lighter, faster, cheaper alternative to REST APIs. They sacrifice some features (no caching, no request transformation, no usage plans) for significantly lower latency and cost.

Use HTTP APIs when:

• You are building microservices with Lambda backends
• You do not need per-method caching
• JWT authorisation (Cognito, Auth0) is sufficient
• Low latency matters more than advanced API management features

# CloudFormation - HTTP API with JWT authoriser and Lambda integration
AWSTemplateFormatVersion: "2010-09-09"

Resources:
  MyHTTPAPI:
    Type: AWS::ApiGatewayV2::Api
    Properties:
      Name: my-http-api
      ProtocolType: HTTP
      CorsConfiguration:
        AllowOrigins:
          - "https://myapp.com"
          - "https://www.myapp.com"
        AllowMethods:
          - GET
          - POST
          - PUT
          - DELETE
          - OPTIONS
        AllowHeaders:
          - Content-Type
          - Authorization
          - X-Request-ID
        MaxAge: 300

  JWTAuthoriser:
    Type: AWS::ApiGatewayV2::Authorizer
    Properties:
      ApiId: !Ref MyHTTPAPI
      AuthorizerType: JWT
      Name: CognitoJWT
      IdentitySource:
        - "$request.header.Authorization"
      JwtConfiguration:
        Audience:
          - !Ref CognitoAppClientId
        Issuer: !Sub "https://cognito-idp.${AWS::Region}.amazonaws.com/${CognitoUserPoolId}"

  GetUsersRoute:
    Type: AWS::ApiGatewayV2::Route
    Properties:
      ApiId: !Ref MyHTTPAPI
      RouteKey: "GET /users/{userId}"
      AuthorizationType: JWT
      AuthorizerId: !Ref JWTAuthoriser
      Target: !Sub "integrations/${LambdaIntegration}"

  LambdaIntegration:
    Type: AWS::ApiGatewayV2::Integration
    Properties:
      ApiId: !Ref MyHTTPAPI
      IntegrationType: AWS_PROXY
      IntegrationUri: !GetAtt GetUserFunction.Arn
      PayloadFormatVersion: "2.0"

  DefaultStage:
    Type: AWS::ApiGatewayV2::Stage
    Properties:
      ApiId: !Ref MyHTTPAPI
      StageName: "$default"
      AutoDeploy: true
      DefaultRouteSettings:
        ThrottlingBurstLimit: 500
        ThrottlingRateLimit: 100
      AccessLogSettings:
        DestinationArn: !GetAtt APILogGroup.Arn
        Format: '{"requestId":"$context.requestId","ip":"$context.identity.sourceIp","requestTime":"$context.requestTime","httpMethod":"$context.httpMethod","routeKey":"$context.routeKey","status":"$context.status","protocol":"$context.protocol","responseLength":"$context.responseLength","integrationError":"$context.integrationErrorMessage"}'

---

API Gateway - WebSocket APIs

API Gateway WebSocket APIs maintain persistent connections and route messages to Lambda functions based on the message content.

# WebSocket API - CloudFormation
Resources:
  WebSocketAPI:
    Type: AWS::ApiGatewayV2::Api
    Properties:
      Name: real-time-api
      ProtocolType: WEBSOCKET
      RouteSelectionExpression: "$request.body.action"  # Route based on 'action' field

  ConnectRoute:
    Type: AWS::ApiGatewayV2::Route
    Properties:
      ApiId: !Ref WebSocketAPI
      RouteKey: $connect
      AuthorizationType: CUSTOM
      AuthorizerId: !Ref WebSocketAuthoriser
      Target: !Sub "integrations/${ConnectIntegration}"

  DisconnectRoute:
    Type: AWS::ApiGatewayV2::Route
    Properties:
      ApiId: !Ref WebSocketAPI
      RouteKey: $disconnect
      Target: !Sub "integrations/${DisconnectIntegration}"

  SendMessageRoute:
    Type: AWS::ApiGatewayV2::Route
    Properties:
      ApiId: !Ref WebSocketAPI
      RouteKey: sendMessage    # action: "sendMessage" routes here
      Target: !Sub "integrations/${SendMessageIntegration}"

# Lambda function for $connect route
import json
import boto3

dynamodb = boto3.resource('dynamodb')
connections_table = dynamodb.Table('WebSocketConnections')

def connect_handler(event, context):
    connection_id = event['requestContext']['connectionId']
    user_id = event['requestContext']['authorizer']['userId']
    room_id = event['queryStringParameters'].get('roomId')

    # Store the connection
    connections_table.put_item(Item={
        'connectionId': connection_id,
        'userId': user_id,
        'roomId': room_id,
        'ttl': int(time.time()) + 7200  # 2 hour TTL
    })

    return {'statusCode': 200}

# Lambda function for sendMessage route
def send_message_handler(event, context):
    connection_id = event['requestContext']['connectionId']
    body = json.loads(event['body'])

    # Get sender's room
    sender = connections_table.get_item(
        Key={'connectionId': connection_id}
    )['Item']

    room_id = sender['roomId']

    # Get all connections in the room
    room_connections = connections_table.query(
        IndexName='RoomIndex',
        KeyConditionExpression='roomId = :room',
        ExpressionAttributeValues={':room': room_id}
    )['Items']

    # Send message to all connections in the room
    api_client = boto3.client(
        'apigatewaymanagementapi',
        endpoint_url=f"https://{event['requestContext']['domainName']}/{event['requestContext']['stage']}"
    )

    message = json.dumps({
        'type': 'message',
        'content': body['content'],
        'senderId': sender['userId'],
        'timestamp': datetime.utcnow().isoformat()
    })

    stale_connections = []

    for conn in room_connections:
        if conn['connectionId'] == connection_id:
            continue  # Don't echo to sender

        try:
            api_client.post_to_connection(
                ConnectionId=conn['connectionId'],
                Data=message.encode()
            )
        except api_client.exceptions.GoneException:
            # Connection is stale — clean it up
            stale_connections.append(conn['connectionId'])

    # Remove stale connections
    for conn_id in stale_connections:
        connections_table.delete_item(Key={'connectionId': conn_id})

    return {'statusCode': 200}

---

API Security Patterns on AWS

Defence in Depth for APIs

Security is applied in layers. Each layer assumes the previous layer may be breached.

Client Request
    ↓
AWS WAF - Block known attack patterns, SQL injection, XSS, bad bots
    ↓
AWS Shield - DDoS protection (Standard free, Advanced managed)
    ↓
CloudFront - Rate limiting at edge, geo-blocking, TLS termination
    ↓
API Gateway - Throttling, API key validation, authoriser, request validation
    ↓
Lambda Authoriser - Custom business logic authorisation
    ↓
Application - Input validation, business rule enforcement
    ↓
Backend Services - Least privilege IAM, VPC isolation

AWS WAF on API Gateway

import boto3

wafv2 = boto3.client('wafv2', region_name='us-east-1')

# Create a WAF Web ACL
web_acl = wafv2.create_web_acl(
    Name='api-gateway-waf',
    Scope='REGIONAL',
    DefaultAction={'Allow': {}},
    Rules=[
        {
            'Name': 'AWSManagedRulesCommonRuleSet',
            'Priority': 1,
            'OverrideAction': {'None': {}},
            'Statement': {
                'ManagedRuleGroupStatement': {
                    'VendorName': 'AWS',
                    'Name': 'AWSManagedRulesCommonRuleSet',
                    'ExcludedRules': []
                }
            },
            'VisibilityConfig': {
                'SampledRequestsEnabled': True,
                'CloudWatchMetricsEnabled': True,
                'MetricName': 'CommonRuleSet'
            }
        },
        {
            'Name': 'RateLimitRule',
            'Priority': 2,
            'Action': {'Block': {}},
            'Statement': {
                'RateBasedStatement': {
                    'Limit': 2000,  # Max 2000 requests per 5 minutes per IP
                    'AggregateKeyType': 'IP'
                }
            },
            'VisibilityConfig': {
                'SampledRequestsEnabled': True,
                'CloudWatchMetricsEnabled': True,
                'MetricName': 'RateLimitRule'
            }
        },
        {
            'Name': 'SQLInjectionRule',
            'Priority': 3,
            'Action': {'Block': {}},
            'Statement': {
                'ManagedRuleGroupStatement': {
                    'VendorName': 'AWS',
                    'Name': 'AWSManagedRulesSQLiRuleSet'
                }
            },
            'VisibilityConfig': {
                'SampledRequestsEnabled': True,
                'CloudWatchMetricsEnabled': True,
                'MetricName': 'SQLInjectionRule'
            }
        }
    ],
    VisibilityConfig={
        'SampledRequestsEnabled': True,
        'CloudWatchMetricsEnabled': True,
        'MetricName': 'api-gateway-waf'
    }
)

Lambda Authoriser

A Lambda Authoriser is a Lambda function that API Gateway calls before invoking your backend. It receives the request's auth credentials and returns an IAM policy allowing or denying access.

# Lambda Authoriser function
import json
import boto3
import jwt

def handler(event, context):
    token = event.get('authorizationToken', '')
    method_arn = event.get('methodArn', '')

    try:
        # Remove 'Bearer ' prefix
        if token.startswith('Bearer '):
            token = token[7:]

        # Validate JWT
        payload = jwt.decode(
            token,
            PUBLIC_KEY,
            algorithms=['RS256']
        )

        user_id = payload['sub']
        roles = payload.get('roles', [])

        # Build policy
        policy = generate_policy(
            principal_id=user_id,
            effect='Allow',
            resource=method_arn,
            context={
                'userId': user_id,
                'roles': json.dumps(roles),
                'email': payload.get('email', '')
            }
        )

        return policy

    except jwt.ExpiredSignatureError:
        raise Exception('Unauthorized')  # Returns 401
    except jwt.InvalidTokenError:
        raise Exception('Unauthorized')

def generate_policy(principal_id, effect, resource, context=None):
    policy = {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': effect,
                'Resource': resource
            }]
        }
    }

    if context:
        policy['context'] = context

    return policy

The context values from the authoriser are available in the downstream Lambda via event['requestContext']['authorizer']. No need to re-validate the token in the backend.

---

Observability - Monitoring APIs in Production

The Three Pillars

Metrics - Quantitative measurements over time. Request count, error rate, latency percentiles (p50, p95, p99), throttle count.

Logs - Records of individual events. Request logs, error details, slow query logs.

Traces - End-to-end records of a request's journey through your system. Which services were called, how long each took, where errors occurred.

Key Metrics to Monitor

Availability: (successful_requests / total_requests) * 100
Error Rate:   (4xx + 5xx responses) / total_requests * 100
Latency p50:  The median response time (50% of requests faster than this)
Latency p99:  99% of requests faster than this (the slow tail)
Throttle Rate: Requests rejected by rate limiting
Cache Hit Rate: For APIs with caching enabled

CloudWatch Dashboard for API Gateway

import boto3

cw = boto3.client('cloudwatch', region_name='us-east-1')

# Custom metric for business-level tracking
def record_api_metric(api_name: str, endpoint: str, method: str, 
                      status_code: int, duration_ms: float):
    cw.put_metric_data(
        Namespace='MyApp/API',
        MetricData=[
            {
                'MetricName': 'RequestCount',
                'Dimensions': [
                    {'Name': 'APIName', 'Value': api_name},
                    {'Name': 'Endpoint', 'Value': endpoint},
                    {'Name': 'Method', 'Value': method}
                ],
                'Value': 1,
                'Unit': 'Count'
            },
            {
                'MetricName': 'Latency',
                'Dimensions': [
                    {'Name': 'APIName', 'Value': api_name},
                    {'Name': 'Endpoint', 'Value': endpoint}
                ],
                'Value': duration_ms,
                'Unit': 'Milliseconds'
            },
            {
                'MetricName': 'ErrorCount',
                'Dimensions': [
                    {'Name': 'APIName', 'Value': api_name},
                    {'Name': 'StatusCode', 'Value': str(status_code)}
                ],
                'Value': 1 if status_code >= 400 else 0,
                'Unit': 'Count'
            }
        ]
    )

# CloudWatch Alarm for error rate
cw.put_metric_alarm(
    AlarmName='api-error-rate-high',
    ComparisonOperator='GreaterThanThreshold',
    EvaluationPeriods=3,
    MetricName='5XXError',
    Namespace='AWS/ApiGateway',
    Period=60,
    Statistic='Sum',
    Threshold=10,  # More than 10 5xx errors per minute for 3 minutes
    AlarmDescription='API 5xx error rate is elevated',
    ActionsEnabled=True,
    AlarmActions=[
        'arn:aws:sns:us-east-1:123456789012:api-alerts'
    ],
    Dimensions=[
        {'Name': 'ApiName', 'Value': 'my-api'},
        {'Name': 'Stage', 'Value': 'production'}
    ],
    TreatMissingData='notBreaching'
)

Structured Logging

Every API log entry should be a structured JSON object — not a human-readable string. Structured logs are queryable in CloudWatch Insights, Datadog, or any log aggregation system.

import json
import time
import uuid
from functools import wraps

def api_logger(func):
    @wraps(func)
    async def wrapper(request, *args, **kwargs):
        request_id = request.headers.get('X-Request-ID', str(uuid.uuid4()))
        start_time = time.time()

        log_entry = {
            'requestId': request_id,
            'method': request.method,
            'path': request.url.path,
            'userAgent': request.headers.get('User-Agent'),
            'clientIp': request.client.host,
            'timestamp': datetime.utcnow().isoformat()
        }

        try:
            response = await func(request, *args, **kwargs)
            log_entry.update({
                'statusCode': response.status_code,
                'durationMs': round((time.time() - start_time) * 1000, 2),
                'level': 'INFO' if response.status_code < 400 else 'WARN'
            })
            return response
        except Exception as e:
            log_entry.update({
                'statusCode': 500,
                'error': str(e),
                'durationMs': round((time.time() - start_time) * 1000, 2),
                'level': 'ERROR'
            })
            raise
        finally:
            print(json.dumps(log_entry))

    return wrapper

---

API Design Patterns and Anti-Patterns

Patterns - What Good Looks Like

Consistent naming - Snake_case or camelCase, chosen once, applied everywhere. Never mix.

Pagination on all collection endpoints - Any endpoint that returns a list must support pagination. A collection that has 10 items today may have 10 million tomorrow.

Idempotency keys for mutations - Allow clients to safely retry POST requests without duplicate side effects:

POST /payments
Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000

{"amount": 4250, "currency": "NGN", "recipient": "acc-88721"}

The server stores the idempotency key and result. If the same key is received again, return the original result without re-processing. This makes retries safe.

HATEOAS for discoverability - Include links to related resources in responses:

{
  "id": "order-456",
  "status": "PROCESSING",
  "_links": {
    "self": {"href": "/orders/456"},
    "cancel": {"href": "/orders/456/cancel", "method": "POST"},
    "user": {"href": "/users/88721"},
    "items": {"href": "/orders/456/items"}
  }
}

Clients can discover what actions are available without hardcoding URLs.

Anti-Patterns - What Breaks in Production

Returning 200 for errors:

// Wrong
HTTP/1.1 200 OK
{"success": false, "error": "User not found"}

// Right
HTTP/1.1 404 Not Found
{"error": {"code": "NOT_FOUND", "message": "User not found"}}

Exposing internal implementation details:

// Exposes your database schema
GET /api?query=SELECT * FROM users WHERE id=88721

// Exposes your internal error
{"error": "ERROR 1064 (42000): You have an error in your SQL syntax"}

Inconsistent response structures:

// GET /users/88721 returns this:
{"id": 88721, "name": "Kachi"}

// GET /orders/456 returns this:
{"order_id": "456", "customer_name": "Kachi"}

No request IDs: Without a unique request ID in every response, debugging production issues requires guessing which log entries correspond to which request.

Breaking changes without versioning: Adding a required field to an existing request body, removing a field from a response, changing a field's data type, any of these without a version bump breaks existing clients silently.

Undocumented rate limits: If you rate limit but don't tell clients the limits, they cannot implement correct backoff. They will hammer your API until they get 429s, which is exactly what you don't want.

---

Written by Onyedikachi Obidiegwu | Cloud Security Engineer