Building Production-Ready APIs with Nexios

Feb 1, 2025 — Tutorial, REST API, Python, Framework, Authentication

Building production-ready APIs requires more than just handling HTTP requests. You need robust authentication, input validation, error handling, and comprehensive documentation. Nexios makes all of this straightforward with its utility-first approach.

Setting Up a Production API Structure

Let’s build a complete user management API that demonstrates production-ready patterns:

from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, EmailStr
from nexios import NexiosApp
from nexios.http import Request, Response
from nexios.auth import BaseUser, JWTAuthBackend, AuthenticationMiddleware, auth, create_jwt

# Pydantic models for request/response validation
class UserCreate(BaseModel):
    """Model for creating a new user."""
    name: str
    email: EmailStr
    age: Optional[int] = None

class UserResponse(BaseModel):
    """Model for user response data."""
    id: int
    name: str
    email: str
    age: Optional[int]
    created_at: datetime

# User model for authentication
class User(BaseUser):
    def __init__(self, id: str, username: str, email: str, is_admin: bool = False):
        self.id = id
        self.username = username
        self.email = email
        self.is_admin = is_admin

    @property
    def is_authenticated(self) -> bool:
        return True

    @property
    def identity(self) -> str:
        return self.id

    @property
    def display_name(self) -> str:
        return self.username

    @classmethod
    async def load_user(cls, identity: str):
        # Replace with your database lookup
        user_data = await get_user_by_id(identity)
        if user_data:
            return cls(
                id=user_data["id"],
                username=user_data["username"],
                email=user_data["email"],
                is_admin=user_data.get("is_admin", False)
            )
        return None

# Create application with comprehensive configuration
app = NexiosApp(
    title="Production User API",
    version="1.0.0",
    description="A production-ready user management API built with Nexios"
)

# Set up JWT authentication
jwt_backend = JWTAuthBackend()
app.add_middleware(AuthenticationMiddleware(user_model=User, backend=jwt_backend))

Implementing JWT Authentication & Authorization

Nexios provides built-in JWT authentication with the @auth() decorator:

# Login endpoint to generate JWT tokens
@app.post("/auth/login")
async def login(request: Request, response: Response):
    """Authenticate user and return JWT token."""
    credentials = await request.json

    # Validate credentials (replace with your logic)
    user = await authenticate_user(credentials.get("email"), credentials.get("password"))
    if not user:
        return response.json({"error": "Invalid credentials"}, status=401)

    # Create JWT token with user data
    token = create_jwt({
        "sub": user.identity,
        "username": user.username,
        "email": user.email,
        "is_admin": user.is_admin
    })

    return response.json({
        "access_token": token,
        "token_type": "bearer",
        "user": {
            "id": user.identity,
            "username": user.username,
            "email": user.email
        }
    })

# Helper function to check admin permissions
async def require_admin_access(request: Request, response: Response):
    """Ensure current user has admin privileges."""
    if not request.user.is_admin:
        return response.json({"error": "Admin access required"}, status=403)
    return True

Function-Based Routes with JWT Authentication

Nexios uses function-based routes with the @auth() decorator for protected endpoints:

@app.get("/users")
@auth()  # Requires valid JWT token
async def get_users(request: Request, response: Response):
    """Get all users with pagination and filtering."""
    # Query parameters with validation
    limit = min(int(request.query_params.get('limit', '10')), 100)
    offset = int(request.query_params.get('offset', '0'))
    search = request.query_params.get('search', '')

    # Database query (mock implementation)
    users = await get_users_from_db(limit, offset, search)
    total = await count_users_in_db(search)

    return response.json({
        "users": [UserResponse(**user).dict() for user in users],
        "pagination": {
            "total": total,
            "limit": limit,
            "offset": offset,
            "has_next": offset + limit < total
        },
        "current_user": request.user.display_name
    })

@app.post("/users")
@auth()  # Requires valid JWT token
async def create_user(request: Request, response: Response):
    """Create a new user with validation."""
    # Check admin permissions
    if not request.user.is_admin:
        return response.json({"error": "Admin access required"}, status=403)

    try:
        # Parse and validate request data
        data = await request.json
        user_data = UserCreate(**data)

        # Check if user already exists
        existing_user = await find_user_by_email(user_data.email)
        if existing_user:
            return response.json(
                {"error": "User with this email already exists"},
                status=409
            )

        # Create user in database
        new_user = await create_user_in_db(user_data.dict())

        return response.json(
            UserResponse(**new_user).dict(),
            status=201
        )

    except Exception as e:
        return response.json(
            {"error": "Invalid user data", "details": str(e)},
            status=400
        )

Advanced Route Handlers with Path Parameters

Individual endpoints with path parameters and JWT authentication:

@app.get("/users/{user_id:int}")
@auth()  # Requires valid JWT token
async def get_user_detail(request: Request, response: Response):
    """Get detailed user information."""
    user_id = request.path_params.user_id

    # Check permissions (users can only see their own data unless admin)
    if str(user_id) != request.user.identity and not request.user.is_admin:
        return response.json({"error": "Access denied"}, status=403)

    user = await get_user_by_id(user_id)
    if not user:
        return response.json({"error": "User not found"}, status=404)

    return response.json(UserResponse(**user).dict())

@app.put("/users/{user_id:int}")
@auth()  # Requires valid JWT token
async def update_user(request: Request, response: Response):
    """Update user information."""
    user_id = request.path_params.user_id

    # Check permissions
    if str(user_id) != request.user.identity and not request.user.is_admin:
        return response.json({"error": "Access denied"}, status=403)

    try:
        data = await request.json
        updated_user = await update_user_in_db(user_id, data)

        if not updated_user:
            return response.json({"error": "User not found"}, status=404)

        return response.json(UserResponse(**updated_user).dict())

    except Exception as e:
        return response.json({"error": str(e)}, status=400)

@app.delete("/users/{user_id:int}")
@auth()  # Requires valid JWT token
async def delete_user(request: Request, response: Response):
    """Delete a user (admin only)."""
    if not request.user.is_admin:
        return response.json({"error": "Admin access required"}, status=403)

    user_id = request.path_params.user_id
    success = await delete_user_from_db(user_id)

    if not success:
        return response.json({"error": "User not found"}, status=404)

    return response.json({"message": "User deleted successfully"})

Production Middleware Stack

Add essential middleware for production environments:

from nexios_contrib.etag import ETagMiddleware
from nexios_contrib.trusted import TrustedHostMiddleware

# Security middleware
app.add_middleware(TrustedHostMiddleware(
    allowed_hosts=["api.yourapp.com", "localhost"]
))

# Performance middleware
app.add_middleware(ETagMiddleware())

# Custom logging middleware
@app.add_middleware
async def request_logging(request: Request, response: Response, call_next):
    """Log all requests with timing and user info."""
    start_time = datetime.now()

    # Extract user info if available
    user_id = "anonymous"
    if hasattr(request, 'user') and request.user and request.user.is_authenticated:
        user_id = request.user.identity

    # Process request
    result = await call_next()

    # Log request details
    duration = (datetime.now() - start_time).total_seconds()
    print(
        f"{request.method} {request.url} - "
        f"User: {user_id} - "
        f"Status: {response.status_code} - "
        f"Duration: {duration:.3f}s"
    )

    return result

# CORS middleware for frontend integration
@app.add_middleware
async def cors_middleware(request: Request, response: Response, call_next):
    """Handle CORS for frontend applications."""
    if request.method == "OPTIONS":
        response.headers["Access-Control-Allow-Origin"] = "*"
        response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, OPTIONS"
        response.headers["Access-Control-Allow-Headers"] = "Authorization, Content-Type"
        return response.text("", status=200)

    result = await call_next()
    response.headers["Access-Control-Allow-Origin"] = "*"
    return result

Error Handling & Validation

Nexios provides built-in error handling using add_exception_handler:

from nexios.exceptions import HTTPException
from pydantic import ValidationError

# Custom exception classes
class UserNotFoundError(Exception):
    def __init__(self, user_id: int):
        self.user_id = user_id
        super().__init__(f"User with ID {user_id} not found")

class ValidationError(Exception):
    def __init__(self, message: str, details: dict = None):
        self.message = message
        self.details = details or {}
        super().__init__(message)

# Exception handlers using add_exception_handler
@app.add_exception_handler(UserNotFoundError)
async def handle_user_not_found(request: Request, response: Response, exc: UserNotFoundError):
    """Handle user not found errors."""
    return response.json(
        {"error": "User not found", "user_id": exc.user_id},
        status=404
    )

@app.add_exception_handler(ValidationError)
async def handle_validation_error(request: Request, response: Response, exc: ValidationError):
    """Handle validation errors."""
    return response.json(
        {"error": "Validation failed", "message": exc.message, "details": exc.details},
        status=400
    )

@app.add_exception_handler(ValueError)
async def handle_value_error(request: Request, response: Response, exc: ValueError):
    """Handle general value errors."""
    return response.json(
        {"error": "Invalid input", "message": str(exc)},
        status=400
    )

# Handle HTTP status codes
@app.add_exception_handler(404)
async def handle_not_found(request: Request, response: Response, exc):
    """Handle 404 errors."""
    return response.json(
        {"error": "Resource not found", "path": str(request.url)},
        status=404
    )

@app.add_exception_handler(500)
async def handle_internal_error(request: Request, response: Response, exc):
    """Handle internal server errors."""
    return response.json(
        {"error": "Internal server error", "message": "Something went wrong"},
        status=500
    )

# Example usage in routes
@app.get("/users/{user_id:int}/profile")
@auth()
async def get_user_profile(request: Request, response: Response):
    """Get user profile with proper error handling."""
    user_id = request.path_params.user_id

    # This will trigger UserNotFoundError handler if user doesn't exist
    user = await get_user_by_id(user_id)
    if not user:
        raise UserNotFoundError(user_id)

    return response.json(UserResponse(**user).dict())

Complete Example with Database Helpers

Here’s a complete example with mock database functions:

# Mock database functions (replace with your actual database)
async def authenticate_user(email: str, password: str):
    """Authenticate user credentials."""
    # Replace with your authentication logic
    if email == "admin@example.com" and password == "secret":
        return User(id="1", username="admin", email=email, is_admin=True)
    return None

async def get_user_by_id(user_id: int):
    """Get user by ID from database."""
    # Mock user data
    return {
        "id": user_id,
        "name": "John Doe",
        "email": "john@example.com",
        "age": 30,
        "created_at": datetime.now()
    }

async def get_users_from_db(limit: int, offset: int, search: str):
    """Get users with pagination."""
    # Mock users data
    return [
        {"id": 1, "name": "John Doe", "email": "john@example.com", "age": 30, "created_at": datetime.now()},
        {"id": 2, "name": "Jane Smith", "email": "jane@example.com", "age": 25, "created_at": datetime.now()}
    ]

async def count_users_in_db(search: str):
    """Count total users."""
    return 2

async def create_user_in_db(user_data: dict):
    """Create new user in database."""
    return {
        "id": 3,
        **user_data,
        "created_at": datetime.now()
    }

# Run the application
if __name__ == "__main__":
    print("Starting Production API with JWT Authentication...")
    print("API Documentation: http://localhost:8000/docs")
    print("Login with: POST /auth/login")
    print("  Body: {\"email\": \"admin@example.com\", \"password\": \"secret\"}")

    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

Key Production Benefits

🔒 Built-in JWT Authentication: Native JWT support with @auth() decorator and token management

📊 Automatic Documentation: Every endpoint generates interactive API docs at /docs

🚀 Performance: Async-first design with efficient middleware pipeline

🧪 Function-Based Routes: Clean, simple route handlers with built-in authentication

📈 Scalable: Modular architecture with community extensions via nexios-contrib

Next Steps

In our next post, we’ll explore how to leverage the nexios-contrib package for even more powerful utilities like Redis integration, advanced caching, and monitoring tools.

Ready to build production APIs? Start with:

pip install nexios

Your production-ready API is just a few lines of code away!