Skip to content

Multi-Router Example

This example demonstrates how to organize a large API using multiple routers.

Basic Router Composition

Creating Sub-Routers

from flask import Flask
from pydantic import BaseModel
from fastopenapi.routers import FlaskRouter
from fastopenapi.errors import ResourceNotFoundError

app = Flask(__name__)

# Main router
main_router = FlaskRouter(
    app=app,
    title="E-Commerce API",
    version="2.0.0",
    description="Multi-router e-commerce API example"
)

# Users sub-router
users_router = FlaskRouter()

class User(BaseModel):
    id: int
    username: str
    email: str

@users_router.get("/{user_id}", response_model=User, tags=["Users"])
def get_user(user_id: int):
    """Get user by ID"""
    return User(id=user_id, username="john", email="john@example.com")

@users_router.get("/", response_model=list[User], tags=["Users"])
def list_users():
    """List all users"""
    return [
        User(id=1, username="john", email="john@example.com"),
        User(id=2, username="jane", email="jane@example.com")
    ]

@users_router.post("/", response_model=User, status_code=201, tags=["Users"])
def create_user(user: User):
    """Create a new user"""
    return user

# Products sub-router
products_router = FlaskRouter()

class Product(BaseModel):
    id: int
    name: str
    price: float

@products_router.get("/{product_id}", response_model=Product, tags=["Products"])
def get_product(product_id: int):
    """Get product by ID"""
    return Product(id=product_id, name="Product", price=99.99)

@products_router.get("/", response_model=list[Product], tags=["Products"])
def list_products():
    """List all products"""
    return [
        Product(id=1, name="Laptop", price=999.99),
        Product(id=2, name="Mouse", price=29.99)
    ]

# Include sub-routers with prefixes
main_router.include_router(users_router, prefix="/users")
main_router.include_router(products_router, prefix="/products")

# Main router endpoints
@main_router.get("/health", tags=["Health"])
def health_check():
    """Health check endpoint"""
    return {"status": "healthy"}

if __name__ == "__main__":
    print("API available at:")
    print("  GET  /users - List users")
    print("  GET  /users/{id} - Get user")
    print("  POST /users - Create user")
    print("  GET  /products - List products")
    print("  GET  /products/{id} - Get product")
    print("  GET  /health - Health check")
    print("")
    print("Docs at http://localhost:5000/docs")
    app.run(debug=True)

Modular API Structure

Project Structure

my_api/
 main.py
 config.py
 dependencies.py
 routers/
     __init__.py
     users.py
     products.py
     orders.py
     admin.py

routers/users.py

from pydantic import BaseModel
from fastopenapi import Query, Path, Depends
from fastopenapi.routers import FlaskRouter
from fastopenapi.errors import ResourceNotFoundError
from dependencies import get_db, get_current_user

router = FlaskRouter()

class User(BaseModel):
    id: int
    username: str
    email: str

class UserCreate(BaseModel):
    username: str
    email: str
    password: str

@router.get("/", response_model=list[User], tags=["Users"])
def list_users(
    page: int = Query(1, ge=1),
    limit: int = Query(10, ge=1, le=100),
    db = Depends(get_db)
):
    """List users with pagination"""
    # Fetch from database
    users = db.query_users(page, limit)
    return users

@router.get("/{user_id}", response_model=User, tags=["Users"])
def get_user(
    user_id: int = Path(..., ge=1),
    db = Depends(get_db)
):
    """Get user by ID"""
    user = db.get_user(user_id)
    if not user:
        raise ResourceNotFoundError(f"User {user_id} not found")
    return user

@router.post("/", response_model=User, status_code=201, tags=["Users"])
def create_user(
    user: UserCreate,
    db = Depends(get_db)
):
    """Create a new user"""
    created_user = db.create_user(user)
    return created_user

@router.delete("/{user_id}", status_code=204, tags=["Users"])
def delete_user(
    user_id: int = Path(..., ge=1),
    current_user = Depends(get_current_user),
    db = Depends(get_db)
):
    """Delete a user (requires authentication)"""
    db.delete_user(user_id)
    return None

routers/products.py

from pydantic import BaseModel
from fastopenapi import Query, Path, Depends
from fastopenapi.routers import FlaskRouter
from fastopenapi.errors import ResourceNotFoundError, AuthorizationError
from dependencies import get_db, get_current_user

router = FlaskRouter()

class Product(BaseModel):
    id: int
    name: str
    description: str
    price: float
    stock: int

class ProductCreate(BaseModel):
    name: str
    description: str
    price: float
    stock: int

@router.get("/", response_model=list[Product], tags=["Products"])
def list_products(
    category: str = Query(None),
    min_price: float = Query(None, ge=0),
    max_price: float = Query(None, ge=0),
    db = Depends(get_db)
):
    """List products with filters"""
    products = db.query_products(
        category=category,
        min_price=min_price,
        max_price=max_price
    )
    return products

@router.get("/{product_id}", response_model=Product, tags=["Products"])
def get_product(
    product_id: int = Path(..., ge=1),
    db = Depends(get_db)
):
    """Get product by ID"""
    product = db.get_product(product_id)
    if not product:
        raise ResourceNotFoundError(f"Product {product_id} not found")
    return product

@router.post("/", response_model=Product, status_code=201, tags=["Products"])
def create_product(
    product: ProductCreate,
    current_user = Depends(get_current_user),
    db = Depends(get_db)
):
    """Create a new product (admin only)"""
    # Check if user is admin
    if not current_user.is_admin:
        raise AuthorizationError("Admin access required")

    created_product = db.create_product(product)
    return created_product

routers/orders.py

from pydantic import BaseModel
from datetime import datetime
from fastopenapi import Path, Depends
from fastopenapi.routers import FlaskRouter
from fastopenapi.errors import ResourceNotFoundError, AuthorizationError
from dependencies import get_db, get_current_user

router = FlaskRouter()

class OrderItem(BaseModel):
    product_id: int
    quantity: int
    price: float

class Order(BaseModel):
    id: int
    user_id: int
    items: list[OrderItem]
    total: float
    status: str
    created_at: datetime

class OrderCreate(BaseModel):
    items: list[OrderItem]

@router.get("/", response_model=list[Order], tags=["Orders"])
def list_my_orders(
    current_user = Depends(get_current_user),
    db = Depends(get_db)
):
    """List current user's orders"""
    orders = db.get_user_orders(current_user.id)
    return orders

@router.get("/{order_id}", response_model=Order, tags=["Orders"])
def get_order(
    order_id: int = Path(..., ge=1),
    current_user = Depends(get_current_user),
    db = Depends(get_db)
):
    """Get order by ID"""
    order = db.get_order(order_id)

    if not order:
        raise ResourceNotFoundError(f"Order {order_id} not found")

    # Verify ownership
    if order.user_id != current_user.id:
        raise AuthorizationError("You can only view your own orders")

    return order

@router.post("/", response_model=Order, status_code=201, tags=["Orders"])
def create_order(
    order: OrderCreate,
    current_user = Depends(get_current_user),
    db = Depends(get_db)
):
    """Create a new order"""
    created_order = db.create_order(current_user.id, order)
    return created_order

routers/admin.py

from fastopenapi import Depends
from fastopenapi.routers import FlaskRouter
from dependencies import get_admin_user

router = FlaskRouter()

@router.get("/stats", tags=["Admin"])
def get_stats(admin = Depends(get_admin_user)):
    """Get platform statistics (admin only)"""
    return {
        "total_users": 1000,
        "total_products": 500,
        "total_orders": 2500
    }

@router.get("/users", tags=["Admin"])
def list_all_users(admin = Depends(get_admin_user)):
    """List all users (admin only)"""
    return {"users": []}

@router.delete("/users/{user_id}", status_code=204, tags=["Admin"])
def delete_user_admin(
    user_id: int,
    admin = Depends(get_admin_user)
):
    """Delete any user (admin only)"""
    # Delete user
    return None

main.py

from flask import Flask
from fastopenapi.routers import FlaskRouter

# Import all routers
from routers import users, products, orders, admin

app = Flask(__name__)

# Create main router
api_router = FlaskRouter(
    app=app,
    title="E-Commerce Platform API",
    version="2.0.0",
    description="Complete e-commerce API with multiple routers"
)

# Include all sub-routers
api_router.include_router(users.router, prefix="/users")
api_router.include_router(products.router, prefix="/products")
api_router.include_router(orders.router, prefix="/orders")
api_router.include_router(admin.router, prefix="/admin")

# Root endpoint
@api_router.get("/", tags=["Root"])
def root():
    """API root endpoint"""
    return {
        "message": "E-Commerce API",
        "version": "2.0.0",
        "docs": "/docs"
    }

if __name__ == "__main__":
    app.run(debug=True)

Nested Routers

Deep Nesting

# Comments router (nested under posts)
comments_router = FlaskRouter()

@comments_router.get("/", tags=["Comments"])
def list_comments(post_id: int):
    """List comments for a post"""
    return []

@comments_router.post("/", status_code=201, tags=["Comments"])
def create_comment(post_id: int, content: str):
    """Create a comment on a post"""
    return {"post_id": post_id, "content": content}

# Posts router
posts_router = FlaskRouter()

@posts_router.get("/{post_id}", tags=["Posts"])
def get_post(post_id: int):
    """Get post by ID"""
    return {"id": post_id, "title": "Post"}

# Include comments under posts
posts_router.include_router(
    comments_router,
    prefix="/{post_id}/comments"
)

# Main router
main_router.include_router(posts_router, prefix="/posts")

# Results in routes:
# GET  /posts/{post_id}
# GET  /posts/{post_id}/comments
# POST /posts/{post_id}/comments

Versioned APIs

from datetime import datetime
from flask import Flask
from pydantic import BaseModel
from fastopenapi.routers import FlaskRouter

app = Flask(__name__)

# Version 1 router
v1_router = FlaskRouter()

@v1_router.get("/users", tags=["Users V1"])
def list_users_v1():
    """Legacy user list (v1)"""
    return [{"id": 1, "name": "John"}]

# Version 2 router (with improvements)
v2_router = FlaskRouter()

class UserV2(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime

@v2_router.get("/users", response_model=list[UserV2], tags=["Users V2"])
def list_users_v2():
    """Improved user list (v2)"""
    return [
        UserV2(
            id=1,
            username="john",
            email="john@example.com",
            created_at=datetime.now()
        )
    ]

# Main router
main_router = FlaskRouter(
    app=app,
    title="Versioned API",
    version="2.0.0"
)

# Include versioned routers
main_router.include_router(v1_router, prefix="/v1")
main_router.include_router(v2_router, prefix="/v2")

# Routes:
# GET /v1/users
# GET /v2/users

Domain-Driven Design

# Authentication domain
auth_router = FlaskRouter()

@auth_router.post("/register", tags=["Authentication"])
def register(user: UserCreate):
    return {"message": "User registered"}

@auth_router.post("/login", tags=["Authentication"])
def login(credentials: LoginRequest):
    return {"access_token": "..."}

@auth_router.post("/logout", tags=["Authentication"])
def logout():
    return {"message": "Logged out"}

# Blog domain
blog_router = FlaskRouter()

@blog_router.get("/posts", tags=["Blog"])
def list_posts():
    return []

@blog_router.post("/posts", tags=["Blog"])
def create_post(post: PostCreate):
    return post

# E-commerce domain
shop_router = FlaskRouter()

@shop_router.get("/products", tags=["Shop"])
def list_products():
    return []

@shop_router.post("/cart/items", tags=["Shop"])
def add_to_cart(item: CartItem):
    return item

# Combine all domains
main_router = FlaskRouter(app=app)
main_router.include_router(auth_router, prefix="/auth")
main_router.include_router(blog_router, prefix="/blog")
main_router.include_router(shop_router, prefix="/shop")

Shared Dependencies

# dependencies.py
from fastopenapi import Depends, Header
from fastopenapi.errors import AuthenticationError, AuthorizationError

def get_db():
    """Database dependency (shared across all routers)"""
    db = DatabaseSession()
    try:
        yield db
    finally:
        db.close()

def get_current_user(authorization: str = Header(...)):
    """Authentication dependency (shared)"""
    # Verify token
    user = verify_token(authorization)
    if not user:
        raise AuthenticationError("Invalid token")
    return user

def get_admin_user(user = Depends(get_current_user)):
    """Admin authorization (shared)"""
    if not user.is_admin:
        raise AuthorizationError("Admin required")
    return user

# All routers can use these dependencies
# routers/users.py
@router.get("/")
def list_users(db = Depends(get_db)):
    return db.query_users()

# routers/admin.py
@router.get("/stats")
def get_stats(admin = Depends(get_admin_user)):
    return {"stats": "..."}

Best Practices

1. Organize by Feature/Domain

routers/
 auth.py        # Authentication
 users.py       # User management
 products.py    # Products
 orders.py      # Orders
 admin.py       # Admin operations

2. Use Consistent Prefixes

main_router.include_router(users_router, prefix="/users")
main_router.include_router(products_router, prefix="/products")
main_router.include_router(orders_router, prefix="/orders")

3. Share Common Dependencies

# Define once, use everywhere
def get_db():
    ...

# All routers use the same dependency
@users_router.get("/")
def list_users(db = Depends(get_db)):
    ...

@products_router.get("/")
def list_products(db = Depends(get_db)):
    ...

4. Use Tags for Organization

@users_router.get("/", tags=["Users"])
@products_router.get("/", tags=["Products"])
@orders_router.get("/", tags=["Orders"])

5. Version Your API

main_router.include_router(v1_router, prefix="/api/v1")
main_router.include_router(v2_router, prefix="/api/v2")

See Also