Router API

Router Class

Routers allow you to organize related endpoints together and apply common configuration like prefixes, tags, and middleware.

Basic Router (Organize Your Endpoints)

from zenith import Router

# WHY ROUTERS?
# Problem: Putting all endpoints in main.py becomes messy after 10+ routes
# Solution: Routers group related endpoints together (like users, posts, etc.)

# Create a router - like a mini-app for related endpoints
router = Router()  # No configuration needed to start

# Now define routes on the router (not the main app)
@router.get("/")
async def list_items():
    """GET /items/ - List all items.

    Note: The path "/" is relative to where this router is mounted.
    If mounted at /items, this becomes GET /items/
    """
    return {"items": []}  # Empty list for now

@router.post("/")
async def create_item(item: dict):
    """POST /items/ - Create a new item.

    The 'item' parameter is automatically parsed from JSON body.
    """
    # In real app: save to database
    return {"created": item}  # Echo back what was created

@router.get("/{item_id}")
async def get_item(item_id: int):
    """GET /items/{item_id} - Get specific item.

    Path parameter {item_id} is extracted and converted to int.
    Example: GET /items/123 → item_id = 123
    """
    return {"item_id": item_id}

# To use this router, include it in your main app:
# app.include_router(router, prefix="/items")
# Now all these routes are available under /items/

Router with Configuration (Production-Ready Setup)

from zenith import Router
from pydantic import BaseModel

# Define your models first
class UserCreate(BaseModel):
    name: str
    email: str

# Configure router with common settings for all its routes
router = Router(
    # PREFIX - Prepended to all routes in this router
    prefix="/api/v1/users",
    # So "/" becomes "/api/v1/users/"
    # "/{id}" becomes "/api/v1/users/{id}"

    # TAGS - Group endpoints in API documentation
    tags=["Users"],  # All routes appear under "Users" in /docs

    # COMMON RESPONSES - Apply to all routes in router
    responses={
        # These errors might happen on any endpoint
        404: {"description": "User not found"},
        422: {"description": "Validation error"}
        # Each route inherits these + can add its own
    }
)

@router.get("/")
async def list_users():
    """List all users.

    Full path: GET /api/v1/users/
    Tagged as: "Users" in documentation
    Responses: 200 (success), 404, 422 (from router config)
    """
    # Your business logic here
    users = await fetch_users_from_database()
    return {"users": users}

@router.post("/")
async def create_user(user: UserCreate):
    """Create a new user.

    Full path: POST /api/v1/users/
    Body: JSON matching UserCreate model
    Automatic: Validation, serialization, documentation
    """
    # user is already validated by Pydantic!
    new_user = await save_to_database(user)
    return {"user": new_user}, 201  # 201 Created status

# BENEFITS of router configuration:
# 1. Don't repeat prefix on every route
# 2. Automatic API documentation grouping
# 3. Common error responses defined once
# 4. Easy to move/rename groups of endpoints

Router Parameters

prefix

Add a common prefix to all routes:

# Router with prefix
api_router = Router(prefix="/api/v1")
user_router = Router(prefix="/users")
post_router = Router(prefix="/posts")

@user_router.get("/")  # Becomes /users/
async def list_users():
    return {"users": []}

@post_router.get("/")  # Becomes /posts/
async def list_posts():
    return {"posts": []}

# Include in main router
api_router.include_router(user_router)
api_router.include_router(post_router)

# Include in app (final paths: /api/v1/users/, /api/v1/posts/)
app.include_router(api_router)

tags

Group endpoints in OpenAPI documentation:

# Separate routers by feature
user_router = Router(prefix="/users", tags=["Users"])
auth_router = Router(prefix="/auth", tags=["Authentication"])
admin_router = Router(prefix="/admin", tags=["Administration"])

@user_router.get("/")
async def list_users():
    return {"users": []}

@auth_router.post("/login")
async def login(credentials: LoginRequest):
    return {"token": "..."}

dependencies (Apply Auth/Logic to All Routes)

from zenith import Router, Inject, Depends
from zenith.auth import get_current_user

# PROBLEM: You have 20 endpoints that all need authentication
# BAD: Adding auth dependency to every single route
# GOOD: Apply auth once at router level!

# Create router where ALL routes require authentication
protected_router = Router(
    prefix="/protected",

    # DEPENDENCIES run before every route in this router
    dependencies=[
        Inject(get_current_user)  # Validates JWT token
    ]
    # Can have multiple dependencies:
    # dependencies=[
    #     Inject(get_current_user),     # Auth check
    #     Inject(check_subscription),    # Payment check
    #     Inject(log_request)           # Audit logging
    # ]
)

@protected_router.get("/profile")
async def get_profile(
    # Note: No auth dependency here!
    # Router already handled authentication
    current_user = Depends(get_current_user)  # Get the authenticated user
):
    """Get user profile.

    Authentication flow:
    1. Request comes to /protected/profile
    2. Router runs get_current_user dependency
    3. If auth fails: Returns 401 Unauthorized
    4. If auth succeeds: Runs this function
    5. current_user is available here
    """
    return {
        "profile": {
            "id": current_user.id,
            "email": current_user.email
        }
    }

@protected_router.post("/settings")
async def update_settings(
    settings: UserSettings,
    current_user = Depends(get_current_user)
):
    """Update user settings.

    This is also protected by router-level auth.
    No need to repeat auth logic!
    """
    # Update settings for authenticated user
    await update_user_settings(current_user.id, settings)
    return {"updated": True}

# UNPROTECTED router for comparison
public_router = Router(prefix="/public")
# No dependencies = no authentication required

@public_router.get("/info")
async def public_info():
    """This endpoint is publicly accessible."""
    return {"message": "This is public information"}

responses

Define common response schemas:

common_responses = {
    400: {"description": "Bad Request", "model": ErrorResponse},
    401: {"description": "Unauthorized", "model": ErrorResponse},
    404: {"description": "Not Found", "model": ErrorResponse},
    500: {"description": "Internal Server Error", "model": ErrorResponse}
}

router = Router(
    prefix="/users",
    tags=["Users"],
    responses=common_responses
)

@router.get("/{user_id}", responses={
    200: {"model": UserResponse}
})
async def get_user(user_id: int):
    # Inherits common responses + adds 200 response
    return {"user": "..."}

Route Methods

HTTP Methods

router = Router(prefix="/items")

@router.get("/")  # GET /items/
async def list_items():
    return {"items": []}

@router.post("/")  # POST /items/
async def create_item(item: ItemCreate):
    return {"created": item}

@router.get("/{item_id}")  # GET /items/{item_id}
async def get_item(item_id: int):
    return {"item_id": item_id}

@router.put("/{item_id}")  # PUT /items/{item_id}
async def update_item(item_id: int, item: ItemUpdate):
    return {"updated": item_id}

@router.patch("/{item_id}")  # PATCH /items/{item_id}
async def partial_update(item_id: int, item: ItemPartial):
    return {"patched": item_id}

@router.delete("/{item_id}")  # DELETE /items/{item_id}
async def delete_item(item_id: int):
    return {"deleted": item_id}

@router.head("/{item_id}")  # HEAD /items/{item_id}
async def check_item_exists(item_id: int):
    return  # Just headers, no body

@router.options("/")  # OPTIONS /items/
async def item_options():
    return {"methods": ["GET", "POST"]}

Route Decorators with Parameters

@router.get(
    "/{item_id}",
    response_model=ItemResponse,
    status_code=200,
    summary="Get item by ID",
    description="Retrieve a single item by its unique identifier",
    response_description="The requested item",
    tags=["Items", "Retrieval"],
    deprecated=False,
    operation_id="getItemById",
    responses={
        404: {"description": "Item not found"}
    }
)
async def get_item(item_id: int = Path(..., description="Item ID")):
    return {"item_id": item_id}

Including Routers

Basic Inclusion (Connecting Routers to Your App)

from zenith import Zenith, Router

# Step 1: Create your main application
app = Zenith()

# Step 2: Create routers for different features
# Think of routers like folders organizing your code

# User-related endpoints
users_router = Router(prefix="/users")
# Routes: /users/, /users/{id}, /users/me, etc.

# Item-related endpoints
items_router = Router(prefix="/items")
# Routes: /items/, /items/{id}, /items/search, etc.

# Auth endpoints
auth_router = Router(prefix="/auth")
# Routes: /auth/login, /auth/register, /auth/logout

# Step 3: Define routes on each router (usually in separate files)
@users_router.get("/")
async def list_users():
    return {"users": []}

@items_router.get("/")
async def list_items():
    return {"items": []}

@auth_router.post("/login")
async def login(credentials: LoginRequest):
    return {"token": "..."}

# Step 4: Connect routers to main app
app.include_router(users_router)  # Adds all /users/* routes
app.include_router(items_router)  # Adds all /items/* routes
app.include_router(auth_router)   # Adds all /auth/* routes

# That's it! Your app now has:
# GET  /users/
# GET  /items/
# POST /auth/login

# WHY THIS PATTERN?
# 1. Keeps main.py small and clean
# 2. Easy to find endpoints (users in users.py, etc.)
# 3. Can disable features by commenting one line
# 4. Teams can work on different routers without conflicts

Nested Router Structure (API Versioning Pattern)

# PROBLEM: Your API evolves but you need to support old clients
# SOLUTION: API versioning with nested routers

# Level 1: Version routers
api_v1 = Router(
    prefix="/api/v1",
    tags=["API v1"]  # Groups in documentation
)

api_v2 = Router(
    prefix="/api/v2",
    tags=["API v2"]  # Separate section in docs
)

# Level 2: Feature routers for each version

# ===== VERSION 1 ROUTES =====
users_v1 = Router(prefix="/users")
posts_v1 = Router(prefix="/posts")

@users_v1.get("/")
async def list_users_v1():
    """V1: Returns simple user list."""
    return {"users": ["Alice", "Bob"]}  # Just names

@posts_v1.get("/")
async def list_posts_v1():
    return {"posts": [...]}  # V1 format

# Connect V1 features to V1 API
api_v1.include_router(users_v1)  # Becomes /api/v1/users
api_v1.include_router(posts_v1)  # Becomes /api/v1/posts

# ===== VERSION 2 ROUTES =====
users_v2 = Router(prefix="/users")
posts_v2 = Router(prefix="/posts")

@users_v2.get("/")
async def list_users_v2():
    """V2: Returns detailed user objects."""
    return {
        "users": [
            {"id": 1, "name": "Alice", "email": "alice@example.com"},
            {"id": 2, "name": "Bob", "email": "bob@example.com"}
        ],
        "pagination": {"page": 1, "total": 2}  # V2 adds pagination
    }

@posts_v2.get("/")
async def list_posts_v2():
    return {"posts": [...], "meta": {...}}  # V2 format

# Connect V2 features to V2 API
api_v2.include_router(users_v2)  # Becomes /api/v2/users
api_v2.include_router(posts_v2)  # Becomes /api/v2/posts

# Level 3: Connect version routers to main app
app.include_router(api_v1)  # All V1 routes
app.include_router(api_v2)  # All V2 routes

# FINAL ROUTE STRUCTURE:
# /api/v1/users/  → Returns ["Alice", "Bob"]
# /api/v2/users/  → Returns [{id, name, email}, ...] with pagination
# /api/v1/posts/  → Old format
# /api/v2/posts/  → New format with metadata

# BENEFITS:
# 1. Old clients keep working with v1
# 2. New clients get enhanced v2 features
# 3. Can deprecate v1 after migration period
# 4. Clear separation in code and docs

Router with Override Parameters (Reuse with Modifications)

# SCENARIO: You have a router but need to mount it differently
# Example: Same routes for internal and external APIs

# Original router definition (e.g., in routes/users.py)
users_router = Router(
    prefix="/users",      # Base prefix
    tags=["Users"]        # Base tags
)

@users_router.get("/")
async def list_users():
    return {"users": [...]}

@users_router.get("/{user_id}")
async def get_user(user_id: int):
    return {"user": {...}}

# MOUNT 1: Public API with API key requirement
app.include_router(
    users_router,

    # OVERRIDE PREFIX - Prepends to router's prefix
    prefix="/api/v1",
    # Router has: /users
    # This adds: /api/v1
    # Result: /api/v1/users

    # ADD TAGS - Combines with router's tags
    tags=["Public API"],
    # Router has: ["Users"]
    # This adds: ["Public API"]
    # Result: ["Public API", "Users"]

    # ADD DEPENDENCIES - Extra requirements
    dependencies=[Inject(require_api_key)]
    # Now all routes need valid API key
)

# MOUNT 2: Internal API with authentication
app.include_router(
    users_router,  # Same router!

    prefix="/internal",  # Different prefix
    # Result: /internal/users

    tags=["Internal"],  # Different tags

    dependencies=[Inject(require_employee_auth)]
    # Different auth requirement
)

# NOW YOUR APP HAS:
# Public API:
#   GET /api/v1/users/       (requires API key)
#   GET /api/v1/users/{id}   (requires API key)
#
# Internal API:
#   GET /internal/users/     (requires employee auth)
#   GET /internal/users/{id} (requires employee auth)
#
# Same code, different access patterns!

# USE CASES:
# 1. Public vs internal APIs
# 2. Free vs premium endpoints
# 3. Different auth for different clients
# 4. A/B testing different API designs

Router Middleware

Router-Specific Middleware

from zenith.middleware import RateLimitMiddleware

# Create router with middleware
api_router = Router(
    prefix="/api",
    middleware=[
        RateLimitMiddleware({"requests_per_minute": 100})
    ]
)

@api_router.get("/data")
async def get_data():
    # Rate limiting applied only to this router's routes
    return {"data": "..."}

# Include in app
app.include_router(api_router)

Conditional Middleware

import os
from zenith.middleware import SecurityHeadersMiddleware, CORSMiddleware

# Different middleware for different environments
if os.getenv("ENVIRONMENT") == "production":
    middleware = [
        SecurityHeadersMiddleware({"force_https": True}),
        RateLimitMiddleware({"requests_per_minute": 60})
    ]
else:
    middleware = [
        CORSMiddleware({"allow_origins": ["*"]})
    ]

api_router = Router(
    prefix="/api",
    middleware=middleware
)

Router Events

Router Startup/Shutdown

from zenith import Router

router = Router(prefix="/data")

@router.on_event("startup")
async def startup_event():
    """Initialize router resources."""
    print("Data router starting up")
    # Initialize connections, caches, etc.

@router.on_event("shutdown")
async def shutdown_event():
    """Cleanup router resources."""
    print("Data router shutting down")
    # Close connections, cleanup resources

@router.get("/")
async def get_data():
    return {"data": "..."}

# Events are executed when router is included
app.include_router(router)

Multiple Event Handlers

router = Router(prefix="/services")

@router.on_event("startup")
async def init_database():
    """Initialize database connection."""
    print("Initializing database...")

@router.on_event("startup")
async def init_cache():
    """Initialize cache."""
    print("Initializing cache...")

@router.on_event("startup")
async def init_external_services():
    """Connect to external services."""
    print("Connecting to external services...")

# All startup handlers will be executed in order

Route Organization Patterns

Feature-Based Organization

app/routes/users.py

from zenith import Router
from app.services.users import UserService

router = Router(prefix="/users", tags=["Users"])

@router.get("/")
async def list_users(users: UserService = Inject()):
    return await users.list_users()

@router.post("/")
async def create_user(
    user_data: UserCreate,
    users: UserService = Inject()
):
    return await users.create_user(user_data)

# app/routes/posts.py
from zenith import Router
from app.services.posts import PostService

router = Router(prefix="/posts", tags=["Posts"])

@router.get("/")
async def list_posts(posts: PostService = Inject()):
    return await posts.list_posts()

# app/main.py
from app.routes import users, posts

app.include_router(users.router)
app.include_router(posts.router)

Versioned API Structure

app/routes/v1/__init__.py

from zenith import Router
from . import users, posts, auth

router = Router(prefix="/v1", tags=["API v1"])

router.include_router(auth.router)
router.include_router(users.router)
router.include_router(posts.router)

# app/routes/v2/__init__.py
from zenith import Router
from . import users, posts, auth

router = Router(prefix="/v2", tags=["API v2"])

router.include_router(auth.router)
router.include_router(users.router)
router.include_router(posts.router)

# app/main.py
from app.routes import v1, v2

app.include_router(v1.router, prefix="/api")
app.include_router(v2.router, prefix="/api")

Domain-Driven Structure

app/domains/user/routes.py

from zenith import Router
from .service import UserService
from .models import User, UserCreate, UserUpdate

router = Router(prefix="/users", tags=["Users"])

@router.get("/", response_model=List[User])
async def list_users(users: UserService = Inject()):
    return await users.list_users()

# app/domains/order/routes.py
from zenith import Router
from .service import OrderService
from .models import Order, OrderCreate

router = Router(prefix="/orders", tags=["Orders"])

@router.post("/", response_model=Order)
async def create_order(
    order_data: OrderCreate,
    orders: OrderService = Inject()
):
    return await orders.create_order(order_data)

# app/main.py
from app.domains.user.routes import router as user_router
from app.domains.order.routes import router as order_router

app.include_router(user_router)
app.include_router(order_router)

Router Testing

from zenith.testing import TestClient
from zenith import Zenith, Router
import pytest

# Create test router
test_router = Router(prefix="/test")

@test_router.get("/")
async def test_endpoint():
    return {"message": "test"}

@test_router.post("/items")
async def create_test_item(item: dict):
    return {"created": item}

# Test router in isolation
@pytest.mark.asyncio
async def test_router():
    app = Zenith()
    app.include_router(test_router)

    async with TestClient(app) as client:
        # Test GET
        response = await client.get("/test/")
        assert response.status_code == 200
        assert response.json() == {"message": "test"}

        # Test POST
        response = await client.post("/test/items", json={"name": "test"})
        assert response.status_code == 200
        assert response.json() == {"created": {"name": "test"}}

Advanced Router Features

Custom Route Classes

from zenith.routing import APIRoute
from typing import Callable

class TimingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request):
            import time
            start_time = time.time()
            response = await original_route_handler(request)
            process_time = time.time() - start_time
            response.headers["X-Process-Time"] = str(process_time)
            return response

        return custom_route_handler

# Use custom route class
router = Router(
    prefix="/timed",
    route_class=TimingRoute
)

@router.get("/")
async def timed_endpoint():
    return {"message": "This response includes timing headers"}

Router with Custom Dependency Resolution

class DatabaseRouter(Router):
    def __init__(self, db_url: str, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.db_url = db_url

    def get_database_session(self):
        # Custom dependency for this router
        return create_session(self.db_url)

# Use custom router
db_router = DatabaseRouter(
    db_url="postgresql://localhost/special_db",
    prefix="/special"
)

@db_router.get("/data")
async def get_special_data():
    # Uses the router's custom database session
    return {"data": "from special database"}

Tip

Organization Tip: Use routers to group related functionality and apply common configuration. This keeps your API organized and makes it easier to apply cross-cutting concerns like authentication and rate limiting.

Router Best Practices

1. Group by Feature - Organize routes by domain/feature, not by HTTP method
2. Use Prefixes - Apply consistent URL prefixes for better organization
3. Tag Appropriately - Use tags to group related endpoints in documentation
4. Apply Common Dependencies - Use router-level dependencies for shared requirements
5. Version Your APIs - Use separate routers for different API versions
6. Keep Routers Focused - Each router should have a single responsibility
7. Test Routers in Isolation - Test router functionality independently

Routers are essential for building maintainable, well-organized APIs in Zenith. They provide the structure and flexibility needed for complex applications while maintaining clean separation of concerns.