fastapi-zero-to-hero

name	fastapi-zero-to-hero
description	Complete FastAPI API development framework for Python. Provides comprehensive assistance for building APIs with routing, authentication (JWT, OAuth2, Better Auth), Pydantic models, database integration, and deployment using uv package manager. Use when users ask to build FastAPI applications, implement authentication, create API endpoints, or develop backend services in Python.

FastAPI Zero to Hero - Complete API Development Framework

Overview

This skill provides comprehensive assistance for FastAPI API development in Python, from basic setup to advanced features. It covers routing, authentication, database integration, testing, and deployment patterns using best practices with uv as the package manager.

What This Skill Does

• Creates FastAPI project structures with recommended organization
• Implements API routing with proper error handling
• Sets up authentication systems (JWT, OAuth2, Better Auth)
• Creates Pydantic models for request/response validation
• Configures database integration (SQLAlchemy/async)
• Provides testing and deployment patterns
• Follows FastAPI best practices and security guidelines
• Uses uv package manager for dependency management

What This Skill Does NOT Do

• Create frontend applications (React, Vue, etc.)
• Manage infrastructure (Docker, Kubernetes, cloud deployment)
• Handle specific business logic implementation beyond API patterns
• Provide complete application code without user requirements

---

Before Implementation

Gather context to ensure successful implementation:

Source	Gather
Codebase	Existing structure, patterns, conventions to integrate with
Conversation	User's specific API requirements, authentication needs, database preferences
Skill References	FastAPI documentation patterns, best practices, security guidelines
User Guidelines	Project-specific conventions, team standards, deployment requirements

Ensure all required context is gathered before implementing. Only ask user for THEIR specific requirements (domain expertise is in this skill).

---

Required Clarifications

Ask about USER'S context (not domain knowledge):

1. API scope: "What specific API endpoints or functionality do you need?"
2. Authentication: "Which authentication method do you prefer (JWT, OAuth2, Better Auth)?"
3. Database: "Which database are you planning to use (PostgreSQL, MySQL, etc.)?"
4. Deployment: "Where do you plan to deploy the API (Docker, cloud, etc.)?"

---

Workflow

1. Set up project structure and dependencies with uv
2. Create basic FastAPI application with proper configuration
3. Implement authentication system based on requirements
4. Design Pydantic models for data validation
5. Set up database integration with SQLAlchemy
6. Create API routes with proper error handling
7. Add testing framework and write tests
8. Prepare deployment configuration

---

Project Setup with uv

Installation with uv (recommended package manager)

# Install FastAPI with standard dependencies using uv
uv add "fastapi[standard]"
uv add uvicorn[standard]
uv add python-jose[cryptography]
uv add passlib[bcrypt]
uv add python-multipart
uv add sqlalchemy
uv add asyncpg  # For PostgreSQL async support
uv add python-dotenv
uv add pytest
uv add pytest-asyncio
uv add httpx  # For testing

Alternative: Install all dependencies at once

uv add "fastapi[standard]" uvicorn python-jose passlib python-multipart sqlalchemy asyncpg python-dotenv pytest pytest-asyncio httpx

Recommended Project Structure

my-fastapi-project/
├── main.py                 # Application entry point
├── app/
│   ├── __init__.py
│   ├── main.py            # FastAPI app instance
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   └── routes/
│   │   │       ├── __init__.py
│   │   │       ├── users.py
│   │   │       └── items.py
│   ├── models/            # Pydantic models
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/           # Database schemas
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   ├── database/
│   │   ├── __init__.py
│   │   └── database.py
│   ├── auth/
│   │   ├── __init__.py
│   │   └── auth.py
│   └── utils/
│       ├── __init__.py
│       └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_users.py
│   └── test_items.py
├── requirements.txt
├── .env
├── .gitignore
└── README.md

---

Core FastAPI Application Structure

main.py - Application Entry Point with Advanced Configuration

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uvicorn
import os
import logging
import datetime
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup logic
    logger.info("Starting up...")
    # Initialize database connections, cache, etc.
    # Example: await database.connect()
    yield
    # Shutdown logic
    logger.info("Shutting down...")
    # Cleanup: await database.disconnect()

# Create FastAPI app with lifespan and comprehensive configuration
app = FastAPI(
    title="My FastAPI Application",
    description="A comprehensive API built with FastAPI",
    version="1.0.0",
    lifespan=lifespan,
    # Additional configuration
    docs_url="/docs",  # Swagger UI
    redoc_url="/redoc",  # ReDoc
    openapi_url="/openapi.json",  # OpenAPI schema
)

# Add CORS middleware
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv("ALLOWED_ORIGINS", ).split(),
    allow_credentials=,
    allow_methods=[],
    allow_headers=[],
    
    allow_credentials=,
    allow_headers=[],
)



  ():
    logger.info()
    response =  call_next(request)
    logger.info()
     response



  ():
    logger.error()
     JSONResponse(
        status_code=,
        content={: }
    )


  ():
     {: , : }


  ():
     {: , : datetime.datetime.utcnow()}


 app.api.v1  router  api_v1_router
app.include_router(api_v1_router, prefix=)

 __name__ == :
    uvicorn.run(
        ,
        host=,
        port=(os.getenv(, )),
        reload=(os.getenv(, ).lower() == ),
        log_level=os.getenv(, )
    )

Advanced Routing and Path Operations

from fastapi import APIRouter, Path, Query, Body, status
from typing import List, Optional, Union
from pydantic import BaseModel, Field
import datetime

# Create API router
router = APIRouter(prefix="/users", tags=["users"])

# Path parameters with validation
@router.get("/{user_id}", summary="Get user by ID")
async def get_user(
    user_id: int = Path(..., ge=1, description="The ID of the user to retrieve"),
):
    """
    Retrieve a user by ID.

    - **user_id**: The unique identifier of the user
    """
    # Implementation here
    return {"user_id": user_id}

# Query parameters with advanced validation
@router.get("/", summary="Get multiple users")
async def get_users(
    skip: int = Query(0, ge=0, description="Number of users to skip"),
    limit: int = Query(100, ge=1, le=1000, description="Maximum number of users to return"),
    q: [] = Query(),
    active_only:  = Query(),
):
    
    
     {: skip, : limit, : q, : active_only}



  ():
    
    
     user



  ():
    
    
     {: user_id, : user_update, : notify}



  ():
    
    
    


 fastapi  UploadFile, File
 typing  


  ():
    
    contents =  file.read()
     {
        : file.filename,
        : file.content_type,
        : (contents)
    }


  ():
    
    results = []
     file  files:
        contents =  file.read()
        results.append({
            : file.filename,
            : file.content_type,
            : (contents)
        })
     {: results}

Dependency Injection Advanced Patterns

from fastapi import Depends, Header, HTTPException
from typing import Optional
import secrets

# Simple dependency
async def common_parameters(
    q: Optional[str] = None,
    skip: int = 0,
    limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}

# Dependency with sub-dependencies
async def verify_token(x_token: str = Header(...)):
    if not secrets.compare_digest(x_token, "fake-super-secret-token"):
        raise HTTPException(status_code=400, detail="X-Token header invalid")

async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key

# Complex dependency with database session
from sqlalchemy.ext.asyncio import AsyncSession

  () -> AsyncSession:
      AsyncSessionLocal()  session:
        :
             session
        :
             session.close()


  ():
    
    user =  get_user_from_token(db, token)
      user:
         HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=,
            headers={: },
        )
     user


  ():
      current_user.is_active:
         HTTPException(status_code=, detail=)
     current_user



  ():
     current_user

Background Tasks

from fastapi import BackgroundTasks
import asyncio

def send_email_task(email: str, message: str):
    """Simulate sending an email in the background"""
    print(f"Sending email to {email}: {message}")
    # In real app: send actual email
    time.sleep(2)  # Simulate async operation
    print("Email sent!")

@router.post("/send-email")
async def send_email(
    email: str,
    background_tasks: BackgroundTasks
):
    """
    Send email in background task.
    """
    background_tasks.add_task(send_email_task, email, "Welcome to our service!")
    return {"message": "Email will be sent in background"}

Custom Response Classes

from fastapi.responses import ORJSONResponse, UJSONResponse, HTMLResponse
from fastapi import Response

# Custom JSON response for performance
@router.get("/optimized-json", response_class=ORJSONResponse)
async def get_optimized_json():
    return {"message": "This uses orjson for faster serialization"}

# HTML response
@router.get("/html", response_class=HTMLResponse)
async def get_html():
    return """
    <html>
        <head>
            <title>FastAPI HTML Response</title>
        </head>
        <body>
            <h1>Hello from FastAPI!</h1>
        </body>
    </html>
    """

# Custom response with headers
@router.get("/custom-response")
async def get_custom_response(response: Response):
    response.headers["X-Custom-Header"] = "Custom value"
    return {"message": "Response with custom header"}

---

Authentication Systems

JWT Authentication with Security Best Practices

from datetime import datetime, timedelta
from typing import Optional
from fastapi import Depends, HTTPException, status, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
import secrets
import os
import bcrypt

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=True)

# Security configuration
SECRET_KEY = os.getenv("SECRET_KEY", secrets.token_urlsafe(32))
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

class Token(BaseModel):
    access_token: str
    token_type: str
    refresh_token: Optional[str] = None

class TokenData(BaseModel):
    username: Optional[str] = None
    scopes: list[str] = []

class ():
    username: 
    email: [] = 
    full_name: [] = 
    disabled: [] = 

 ():
    is_active:  = 
    is_admin:  = 

 ():
    hashed_password: 

 ():
    username: 
    email: 
    password: 
    full_name: [] = 

 ():
     pwd_context.verify(plain_password, hashed_password)

 ():
     pwd_context.(password)

 () -> [UserInDB]:
     username  db:
        user_dict = db[username]
         UserInDB(**user_dict)
     

  ():
    user = get_user(db, username)
      user:
        
        verify_password(password, )
         
      verify_password(password, user.hashed_password):
         
     user

 ():
    to_encode = data.copy()
     expires_delta:
        expire = datetime.utcnow() + expires_delta
    :
        expire = datetime.utcnow() + timedelta(minutes=)
    to_encode.update({: expire, : })
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
     encoded_jwt

 ():
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({: expire, : })
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
     encoded_jwt

  ():
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=,
        headers={: },
    )
    :
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username:  = payload.get()
        token_type:  = payload.get()

         username  :
             credentials_exception
         token_type != :
             credentials_exception

        token_data = TokenData(username=username)
     JWTError:
         credentials_exception

    
    user = get_user(fake_users_db, username=token_data.username)
     user  :
         credentials_exception
     user

  ():
     current_user.disabled:
         HTTPException(status_code=, detail=)
     current_user

  ():
      current_user.is_admin:
         HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=
        )
     current_user


fake_users_db = {
    : {
        : ,
        : ,
        : ,
        : get_password_hash(),
        : ,
        : ,
        : ,
    },
    : {
        : ,
        : ,
        : ,
        : get_password_hash(),
        : ,
        : ,
        : ,
    }
}


  ():
    user =  authenticate_user(fake_users_db, form_data.username, form_data.password)
      user:
         HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=,
            headers={: },
        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={: user.username, : []},
        expires_delta=access_token_expires
    )

    refresh_token = create_refresh_token(data={: user.username})

     {
        : access_token,
        : ,
        : refresh_token
    }


  ():
     current_user



  ():
    :
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        username:  = payload.get()
        token_type:  = payload.get()

         username    token_type != :
             HTTPException(status_code=, detail=)

        
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        new_access_token = create_access_token(
            data={: username, : []},
            expires_delta=access_token_expires
        )

         {: new_access_token, : }
     JWTError:
         HTTPException(status_code=, detail=)

OAuth2 with Authorization Code Flow (Google, GitHub, etc.)

from fastapi import Request
from fastapi.responses import RedirectResponse, JSONResponse
from authlib.integrations.starlette_client import OAuth
import os
from urllib.parse import urlencode

# OAuth setup for Google, GitHub, etc.
oauth = OAuth()

# Register Google OAuth
oauth.register(
    name='google',
    client_id=os.getenv('GOOGLE_CLIENT_ID'),
    client_secret=os.getenv('GOOGLE_CLIENT_SECRET'),
    server_metadata_url='https://accounts.google.com/.well-known/openid_configuration',
    client_kwargs={
        'scope': 'openid email profile'
    }
)

# Register GitHub OAuth
oauth.register(
    name='github',
    client_id=os.getenv('GITHUB_CLIENT_ID'),
    client_secret=os.getenv('GITHUB_CLIENT_SECRET'),
    access_token_url='https://github.com/login/oauth/token',
    authorize_url='https://github.com/login/oauth/authorize',
    api_base_url='https://api.github.com/',
    client_kwargs={'scope': 'user:email'}
)

@app.route('/auth/{provider}/login')
async def auth_login(request: Request, provider: str):
    if provider not in ['google', 'github']:
        raise HTTPException(status_code=404, detail="Provider not supported")

    redirect_uri = request.url_for()
      oauth.create_client(provider).authorize_redirect(request, redirect_uri)


  ():
    :
        token =  oauth.create_client(provider).authorize_access_token(request)
        user_info = token.get()   oauth.create_client(provider).userinfo(token=token)

        
        
        access_token = create_access_token(data={: user_info.get(, user_info.get())})

        
        params = urlencode({: access_token})
         RedirectResponse(url=)

     Exception  e:
        ()
         JSONResponse(status_code=, content={: })

 base64
 hashlib



  ():
     provider   [, ]:
         HTTPException(status_code=, detail=)

    
    code_verifier = secrets.token_urlsafe()
    code_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode()).digest()
    ).decode().rstrip()

    
    request.session[] = code_verifier

    redirect_uri = request.url_for()
      oauth.create_client(provider).authorize_redirect(
        request,
        redirect_uri,
        code_challenge=code_challenge,
        code_challenge_method=
    )

API Key Authentication

from fastapi.security import APIKeyHeader, APIKeyQuery
from typing import Optional
from fastapi import Security

# API Key authentication
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
api_key_query = APIKeyQuery(name="api_key", auto_error=False)

API_KEYS = os.getenv("API_KEYS", "").split(",")

async def get_api_key(api_key_header: str = Security(api_key_header),
                     api_key_query: str = Security(api_key_query)):
    api_key = api_key_header or api_key_query
    if api_key and api_key in API_KEYS:
        return api_key
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API Key"
        )

@app.get("/api/protected-endpoint")
async def protected_endpoint(api_key: str = Security(get_api_key)):
    return {"message": "Access granted with API key", "key": api_key[:8] + "..." if api_key else }

Better Auth Integration (Alternative approach)

# Better Auth is typically a frontend authentication solution
# For backend integration, you might implement custom auth middleware
from fastapi import Request, HTTPException
import jwt
import os
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend

# For Better Auth integration, you would typically:
# 1. Verify the session token from Better Auth
# 2. Extract user information
# 3. Potentially sync with your own user system

async def verify_better_auth_session(request: Request):
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Not authenticated")

    token = auth_header.split(" ")[1]
    try:
        # In a real Better Auth integration, you would:
        # - Fetch the public key from Better Auth
        # - Verify the JWT signature
        # - Extract user information

        # Example with a public key (in real app, fetch from Better Auth API)
        public_key_pem = os.getenv("BETTER_AUTH_PUBLIC_KEY")
        if not public_key_pem:
            raise HTTPException(status_code=500, detail="Public key not configured")

        
        public_key = serialization.load_pem_public_key(
            public_key_pem.encode(),
            backend=default_backend()
        )

        
        payload = jwt.decode(
            token,
            public_key,
            algorithms=[]
        )

        
         {
            : payload.get(),
            : payload.get(),
            : payload.get(),
            : payload.get()
        }
     jwt.InvalidTokenError:
         HTTPException(status_code=, detail=)
     Exception  e:
        ()
         HTTPException(status_code=, detail=)



  ():
     {: , : user_info}

Role-Based Access Control (RBAC)

from enum import Enum
from functools import wraps
from typing import List

class Role(str, Enum):
    USER = "user"
    MODERATOR = "moderator"
    ADMIN = "admin"
    SUPER_ADMIN = "super_admin"

def require_role(required_roles: List[Role]):
    def role_checker(current_user: User = Depends(get_current_active_user)):
        if current_user.is_admin or current_user.username == "admin":
            return current_user

        user_role = getattr(current_user, 'role', Role.USER)
        if user_role not in required_roles:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Access denied. Required roles: {', '.join([role.value for role in required_roles])}"
            )
        return current_user
    return role_checker

# Usage examples
@app.get("/admin-panel", dependencies=[Depends(require_role([Role.ADMIN, Role.SUPER_ADMIN]))])
async def admin_panel():
     {: , : current_user.username}


  ():
     {: , : current_user.username}

Security Headers and Rate Limiting

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# Rate limiting
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/limited-endpoint")
@limiter.limit("5/minute")  # 5 requests per minute per IP
async def limited_endpoint(request: Request):
    return {"message": "This endpoint is rate limited"}

# Security headers middleware
from starlette.middleware.base import BaseHTTPMiddleware

class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["X-Frame-Options"] = "DENY"
        response.headers["X-XSS-Protection"] = "1; mode=block"
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
        return response

app.add_middleware(SecurityHeadersMiddleware)

---

Pydantic Models

Advanced Pydantic Models with Validation

from pydantic import (
    BaseModel,
    EmailStr,
    field_validator,
    field_serializer,
    model_validator,
    ConfigDict,
    Field,
    HttpUrl,
    AnyUrl
)
from typing import Optional, List, Dict, Any, Union
from datetime import datetime, date
from enum import Enum
import re
import uuid

# Configuration model
class DatabaseConfig(BaseModel):
    host: str = Field(..., description="Database host")
    port: int = Field(5432, ge=1, le=65535, description="Database port")
    database: str = Field(..., description="Database name")
    username: str = Field(..., description="Database username")
    password: str = Field(..., description="Database password")

    model_config = ConfigDict(extra="forbid")  # Forbid extra fields

# Custom field types and validators
class PyObjectId(str):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type, handler):
        return handler(str)

class UserRole(, Enum):
    USER = 
    MODERATOR = 
    ADMIN = 
    SUPER_ADMIN = 

 ():
    email: EmailStr
    username:  = Field(..., min_length=, max_length=, pattern=)
    full_name: [] = Field(, min_length=, max_length=)
    bio: [] = Field(, max_length=)
    avatar_url: [HttpUrl] = 



     ():
          re.(, v):
             ValueError()
         v

 ():
    password:  = Field(..., min_length=)
    confirm_password: 


     ():
         .password != .confirm_password:
             ValueError()
         



     ():
         (v) < :
             ValueError()
          re.search(, v):
             ValueError()
          re.search(, v):
             ValueError()
          re.search(, v):
             ValueError()
          re.search(, v):
             ValueError()
         v

 ():
    email: [EmailStr] = 
    username: [] = Field(, min_length=, max_length=)
    full_name: [] = Field(, min_length=, max_length=)
    bio: [] = Field(, max_length=)
    avatar_url: [HttpUrl] = 

 ():
    : PyObjectId
    created_at: datetime
    updated_at: [datetime] = 
    is_active:  = 
    role: UserRole = UserRole.USER
    email_verified:  = 

    model_config = ConfigDict(from_attributes=)  


     () -> :
         (value)


     () -> :
         value.isoformat()  value  

 ():
    user: UserResponse
    posts_count:  = 
    followers_count:  = 
    following_count:  = 
    is_following:  = 


 (, Enum):
    DRAFT = 
    PUBLISHED = 
    ARCHIVED = 

 ():
    title:  = Field(..., min_length=, max_length=)
    description: [] = Field(, max_length=)
    price:  = Field(..., gt=)
    tags: [] = Field(default_factory=, max_items=)
    metadata: [, ] = Field(default_factory=)



     ():
         (v) > :
             ValueError()
         tag  v:
             (tag) > :
                 ValueError()
         v

 ():
    is_public:  = 

 ():
    title: [] = Field(, min_length=, max_length=)
    description: [] = Field(, max_length=)
    price: [] = Field(, gt=)
    tags: [[]] = Field(, max_items=)
    status: [ItemStatus] = 
    is_public: [] = 
    metadata: [[, ]] = Field()

 ():
    : PyObjectId
    owner_id: PyObjectId
    created_at: datetime
    updated_at: [datetime] = 
    status: ItemStatus = ItemStatus.DRAFT
    is_public:  = 
    views_count:  = 
    likes_count:  = 
    is_liked:  = 

    model_config = ConfigDict(from_attributes=)


     () -> :
         (value)


     () -> :
         value.isoformat()  value  


 ():
    page:  = Field(, ge=)
    limit:  = Field(, ge=, le=)
    sort_by:  = 
    sort_order:  = Field(, pattern=)


     () -> :
         (.page - ) * .limit

 ():
    total: 
    page: 
    limit: 
    pages: 
    has_next: 
    has_prev: 

 ():
    items: [ItemResponse]
    pagination: PaginationResponse


 ():
    filename: 
    content_type: 
    size:  = Field(..., gt=, le= *  * )  
    url: HttpUrl

 ():
    success_count: 
    failure_count: 
    errors: [[, ]] = Field(default_factory=)


 ():
    success:  = 
    message: 
    data: [] = 
    error_code: [] = 

 ():
    access_token: 
    token_type: 
    refresh_token: [] = 
    expires_in: [] = 

 ():
    username: [] = 
    scopes: [] = Field(default_factory=)


 ():
    detail: 
    error_code: [] = 
    timestamp: datetime = Field(default_factory=datetime.utcnow)


 ():
    app_name:  = 
    app_version:  = 
    debug:  = 
    database_url: 
    secret_key: 
    algorithm:  = 
    access_token_expire_minutes:  = 
    refresh_token_expire_days:  = 
    allowed_origins: [] = Field(default_factory=)

    model_config = ConfigDict(
        env_file=,
        env_file_encoding=,
        extra=
    )


 ():
    event_type: 
    event_id:  = Field(default_factory=: (uuid.uuid4()))
    timestamp: datetime = Field(default_factory=datetime.utcnow)
    data: [, ]
    signature: [] = 


 ():
    q:  = Field(..., min_length=, max_length=, description=)
    filters: [, ] = Field(default_factory=)
    page:  = Field(, ge=)
    limit:  = Field(, ge=, le=)
    sort_by: [] = 
    sort_order:  = Field(, pattern=)

Custom Pydantic Data Types and Validators

from pydantic import BaseModel, field_validator
from typing import Optional
import phonenumbers
from phonenumbers import NumberParseException
import ipaddress
from pydantic.functional_validators import AfterValidator
from typing_extensions import Annotated
import json

# Custom phone number validator
def validate_phone_number(v: str) -> str:
    try:
        parsed = phonenumbers.parse(v, None)
        if not phonenumbers.is_valid_number(parsed):
            raise ValueError("Invalid phone number")
        return phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164)
    except NumberParseException:
        raise ValueError("Invalid phone number format")

PhoneNumber = Annotated[str, AfterValidator(validate_phone_number)]

# Custom IP address validator
def validate_ip_address(v: str) -> str:
    try:
        ipaddress.IPv4Address(v)
        return v
    except ipaddress.AddressValueError:
        raise ValueError("Invalid IP address")

IPAddress = Annotated[str, AfterValidator(validate_ip_address)]

# Custom JSON validator
 () -> :
     (v, ):
         v
    :
         json.loads(v)
     json.JSONDecodeError:
         ValueError()

JSONType = Annotated[[, ], AfterValidator(validate_json)]

 ():
    phone: [PhoneNumber] = 
    ip_address: [IPAddress] = 
    metadata: [JSONType] = 
    tags: [] = Field(default_factory=)


 ():
    name: 
    price: 
    category: 
    tags: [] = []
    inventory:  = Field(..., ge=)
    is_available:  = 


     ():
         .inventory == :
            .is_available = 
         .inventory >    .is_available:
            
            ()
         



     ():
         (v) != ((v)):
             ValueError()
         v

Nested Models and Relationships

from typing import List, Optional
from pydantic import BaseModel, Field

class Address(BaseModel):
    street: str
    city: str
    state: str
    zip_code: str = Field(..., pattern=r'^\d{5}(-\d{4})?$')
    country: str = "US"

class Company(BaseModel):
    id: Optional[PyObjectId] = None
    name: str
    description: Optional[str] = None
    website: Optional[HttpUrl] = None
    address: Optional[Address] = None

class UserWithCompany(UserResponse):
    company: Optional[Company] = None
    colleagues: List[UserResponse] = Field(default_factory=list)

class OrderItem(BaseModel):
    product_id: PyObjectId
    quantity: int = Field(..., ge=1)
    price: float = Field(..., ge=0)

class Order(BaseModel):
    id: Optional[PyObjectId] = 
    user_id: PyObjectId
    items: [OrderItem]
    total_amount:  = Field(..., ge=)
    status:  = 
    shipping_address: Address
    billing_address: [Address] = 
    created_at: datetime = Field(default_factory=datetime.utcnow)



     ():
          v:
             ValueError()
         v


     () -> :
         (item.price * item.quantity  item  .items)

---

Database Integration

SQLAlchemy Async Setup with Advanced Configuration

from sqlalchemy import (
    create_engine,
    Column,
    Integer,
    String,
    DateTime,
    Boolean,
    Float,
    Text,
    Index,
    ForeignKey,
    UniqueConstraint,
    text
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, backref
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.pool import AsyncAdaptedQueuePool
from datetime import datetime
import os
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Database URL configuration
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql+asyncpg://user:password@localhost/dbname")

# Async engine with connection pooling
async_engine = create_async_engine(
    DATABASE_URL,
    poolclass=AsyncAdaptedQueuePool,
    pool_size=20,
    max_overflow=30,
    pool_pre_ping=True,  # Verify connections before use
    pool_recycle=300,    # Recycle connections after 5 minutes
    echo=bool(os.getenv("DB_ECHO", "False").lower() == "true")  # Enable SQL logging in debug
)

AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# Sync engine for non-async operations
sync_engine = create_engine(
    DATABASE_URL.replace("postgresql+asyncpg", "postgresql"),
    pool_size=20,
    max_overflow=30,
    pool_pre_ping=True,
    pool_recycle=,
    echo=(os.getenv(, ).lower() == )
)

SessionLocal = sessionmaker(autocommit=, autoflush=, bind=sync_engine)

Base = declarative_base()


  ():
      AsyncSessionLocal()  db:
        :
             db
        :
             db.close()

 ():
    db = SessionLocal()
    :
         db
    :
        db.close()


 :
    created_at = Column(DateTime, default=datetime.utcnow, nullable=)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=)

 (Base, TimestampMixin):
    __abstract__ = 
     = Column(Integer, primary_key=, index=)


 ():
    __tablename__ = 

    email = Column(String(), unique=, index=, nullable=)
    username = Column(String(), unique=, index=, nullable=)
    full_name = Column(String())
    hashed_password = Column(String(), nullable=)
    is_active = Column(Boolean, default=, nullable=)
    is_verified = Column(Boolean, default=, nullable=)
    role = Column(String(), default=, nullable=)

    
    items = relationship(, back_populates=, cascade=)
    profile = relationship(, back_populates=, uselist=, cascade=)

    
    __table_args__ = (
        Index(, ),
        Index(, ),
        Index(, ),
    )

 ():
    __tablename__ = 

    user_id = Column(Integer, ForeignKey(), unique=, nullable=)
    bio = Column(Text)
    avatar_url = Column(String())
    phone = Column(String())
    birth_date = Column(DateTime)

    
    user = relationship(, back_populates=)

 ():
    __tablename__ = 

    title = Column(String(), nullable=)
    description = Column(Text)
    price = Column(Float, nullable=)
    is_public = Column(Boolean, default=, nullable=)
    status = Column(String(), default=, nullable=)
    owner_id = Column(Integer, ForeignKey(), nullable=)

    
    owner = relationship(, back_populates=)
    tags = relationship(, back_populates=, cascade=)

    
    __table_args__ = (
        Index(, ),
        Index(, ),
        Index(, ),
    )

 ():
    __tablename__ = 

    item_id = Column(Integer, ForeignKey(), nullable=)
    tag_name = Column(String(), nullable=)

    
    item = relationship(, back_populates=)

    
    __table_args__ = (
        UniqueConstraint(, , name=),
        Index(, ),
    )


 ():
    __tablename__ = 

    table_name = Column(String(), nullable=)
    record_id = Column(Integer, nullable=)
    action = Column(String(), nullable=)  
    old_values = Column(Text)  
    new_values = Column(Text)  
    user_id = Column(Integer, ForeignKey())
    timestamp = Column(DateTime, default=datetime.utcnow, nullable=)

    user = relationship()


  ():
    
      async_engine.begin()  conn:
        
         conn.run_sync(Base.metadata.create_all)
        logger.info()

  ():
    
      async_engine.begin()  conn:
         conn.run_sync(Base.metadata.drop_all)
        logger.info()


 contextlib  asynccontextmanager


  ():
      AsyncSessionLocal()  session:
        :
             session
             session.commit()
         Exception:
             session.rollback()
            
        :
             session.close()

Async Database Operations and Repository Pattern

from typing import List, Optional, Dict, Any
from sqlalchemy import select, update, delete, and_, or_, func
from sqlalchemy.orm import selectinload
from sqlalchemy.exc import IntegrityError
import json

class BaseRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def create(self, model, **kwargs):
        """Create a new record"""
        try:
            instance = model(**kwargs)
            self.session.add(instance)
            await self.session.commit()
            await self.session.refresh(instance)
            return instance
        except IntegrityError as e:
            await self.session.rollback()
            raise ValueError(f"Integrity error: {str(e)}")

    async def get_by_id(self, model, id: int):
        """Get a record by ID"""
        stmt = select(model).where(model.id == )
        result =  .session.execute(stmt)
         result.scalar_one_or_none()

      ():
        
        stmt = select(model).where(model..in_(ids))
        result =  .session.execute(stmt)
         result.scalars().()

      ():
        
        stmt = update(model).where(model. == ).values(**kwargs)
        result =  .session.execute(stmt)

         result.rowcount == :
             

         .session.commit()

        
          .get_by_id(model, )

      ():
        
        stmt = delete(model).where(model. == )
        result =  .session.execute(stmt)

         result.rowcount == :
             

         .session.commit()
         

      ():
        
        stmt = select(model)

        
         filters:
            conditions = []
             key, value  filters.items():
                 (model, key):
                     (value, ):
                        conditions.append((model, key).in_(value))
                    :
                        conditions.append((model, key) == value)
             conditions:
                stmt = stmt.where(and_(*conditions))

        stmt = stmt.offset(skip).limit(limit)
        result =  .session.execute(stmt)
         result.scalars().()

      ():
        
        stmt = select(func.count(model.))

         filters:
            conditions = []
             key, value  filters.items():
                 (model, key):
                     (value, ):
                        conditions.append((model, key).in_(value))
                    :
                        conditions.append((model, key) == value)
             conditions:
                stmt = stmt.where(and_(*conditions))

        result =  .session.execute(stmt)
         result.scalar_one()

 ():
      ():
        
        stmt = select(User).where(User.username == username)
        result =  .session.execute(stmt)
         result.scalar_one_or_none()

      ():
        
        stmt = select(User).where(User.email == email)
        result =  .session.execute(stmt)
         result.scalar_one_or_none()

      ():
        
        stmt = select(User).where(User.is_active == ).offset(skip).limit(limit)
        result =  .session.execute(stmt)
         result.scalars().()

      ():
        
        search_filter = or_(
            User.username.contains(query),
            User.email.contains(query),
            User.full_name.contains(query)  query  
        )
        stmt = select(User).where(search_filter).offset(skip).limit(limit)
        result =  .session.execute(stmt)
         result.scalars().()

 ():
      ():
        
        stmt = select(Item).where(Item.owner_id == owner_id).offset(skip).limit(limit)
        result =  .session.execute(stmt)
         result.scalars().()

      ():
        
        stmt = select(Item).where(Item.is_public == ).offset(skip).limit(limit)
        result =  .session.execute(stmt)
         result.scalars().()

      ():
        
        stmt = (
            select(Item)
            .join(ItemTag)
            .where(ItemTag.tag_name.in_(tag_names))
            .offset(skip)
            .limit(limit)
        )
        result =  .session.execute(stmt)
         result.scalars().()

      ():
        
        stmt = (
            select(Item)
            .options(selectinload(Item.owner))
            .offset(skip)
            .limit(limit)
        )
        result =  .session.execute(stmt)
         result.scalars().()


 fastapi  Depends, HTTPException, status

  () -> UserRepository:
     UserRepository(db)

  () -> ItemRepository:
     ItemRepository(db)



  ():
    user =  repo.get_by_id(User, user_id)
      user:
         HTTPException(status_code=, detail=)
     user


  ():
    items =  repo.get_by_owner(user_id, skip, limit)
     items

Database Migrations with Alembic

# alembic.ini configuration example
"""
[alembic]
# path to migration scripts
script_location = alembic

# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s

# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .

# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =

# max length of characters to apply to the
# "slug" field
# max_length = 40

# version number format
# version_num_format = %04d

# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses
# os.pathsep. If this key is omitted entirely, it falls back to the legacy
# behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os  # Use os.pathsep. Default configuration used for new projects.

# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8

sqlalchemy.url = driver://user:pass@localhost/dbname


[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are invoked
# automatically whenever a new revision file is created.
# Options include:
#
# hooks = black, isort
# black.type = exec
# black.executable = black
# black.args = -l 79 REVISION_SCRIPT_FILENAME
# isort.type = exec
# isort.executable = isort
# isort.args = REVISION_SCRIPT_FILENAME
#

[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
"""

# Example migration file (generated by Alembic)
"""
Revision ID: abc123def456
Revises:
Create Date: 2023-10-01 12:00:00.000000

"""
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = 'abc123def456'
down_revision = None
branch_labels = None
depends_on = 

 () -> :
    
    op.create_table(
        ,
        sa.Column(, sa.Integer(), nullable=),
        sa.Column(, sa.String(length=), nullable=),
        sa.Column(, sa.String(length=), nullable=),
        sa.Column(, sa.String(length=), nullable=),
        sa.Column(, sa.String(length=), nullable=),
        sa.Column(, sa.Boolean(), nullable=, default=),
        sa.Column(, sa.Boolean(), nullable=, default=),
        sa.Column(, sa.String(length=), nullable=, default=),
        sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
        sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
        sa.PrimaryKeyConstraint(),
        sa.UniqueConstraint(),
        sa.UniqueConstraint()
    )

    
    op.create_table(
        ,
        sa.Column(, sa.Integer(), nullable=),
        sa.Column(, sa.Integer(), nullable=),
        sa.Column(, sa.Text(), nullable=),
        sa.Column(, sa.String(length=), nullable=),
        sa.Column(, sa.String(length=), nullable=),
        sa.Column(, sa.DateTime(), nullable=),
        sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
        sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
        sa.ForeignKeyConstraint([], [], ),
        sa.PrimaryKeyConstraint(),
        sa.UniqueConstraint()
    )

    
    op.create_table(
        ,
        sa.Column(, sa.Integer(), nullable=),
        sa.Column(, sa.String(length=), nullable=),
        sa.Column(, sa.Text(), nullable=),
        sa.Column(, sa.Float(), nullable=),
        sa.Column(, sa.Boolean(), nullable=, default=),
        sa.Column(, sa.String(length=), nullable=, default=),
        sa.Column(, sa.Integer(), nullable=),
        sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
        sa.Column(, sa.DateTime(), nullable=, server_default=sa.text()),
        sa.ForeignKeyConstraint([], [], ),
        sa.PrimaryKeyConstraint()
    )

    
    op.create_index(, , [])
    op.create_index(, , [])
    op.create_index(, , [])
    op.create_index(, , [])
    op.create_index(, , [])
    op.create_index(, , [])

 () -> :
    
    op.drop_index(, table_name=)
    op.drop_index(, table_name=)
    op.drop_index(, table_name=)
    op.drop_index(, table_name=)
    op.drop_index(, table_name=)
    op.drop_index(, table_name=)

    
    op.drop_table()
    op.drop_table()
    op.drop_table()

Database Connection Pooling and Performance Optimization

# Database configuration with performance optimization
from sqlalchemy import event
from sqlalchemy.pool import Pool
import time

# Connection pool configuration
DATABASE_CONFIG = {
    "pool_size": 20,
    "max_overflow": 30,
    "pool_pre_ping": True,
    "pool_recycle": 300,  # 5 minutes
    "pool_timeout": 30,
    "echo": bool(os.getenv("DB_ECHO", "False").lower() == "true")
}

# Connection event listeners for performance monitoring
@event.listens_for(sync_engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
    """Set SQLite pragmas for performance (if using SQLite)"""
    if "sqlite" in DATABASE_URL:
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.execute("PRAGMA journal_mode=WAL")
        cursor.execute("PRAGMA synchronous=NORMAL")
        cursor.close()

@event.listens_for(async_engine, "connect")
def set_async_sqlite_pragma(dbapi_connection, connection_record):
    """Set SQLite pragmas for async engine"""
    if   DATABASE_URL:
        cursor = dbapi_connection.cursor()
        cursor.execute()
        cursor.execute()
        cursor.execute()
        cursor.close()



 ():
    connection_record.start_time = time.time()


 ():
     (connection_record, ):
        total_time = time.time() - connection_record.start_time
         total_time > :  
            logger.warning()


 ():
      () -> [, ]:
        
        sql = 

        result =  .session.execute(text(sql), {: user_id})
        row = result.fetchone()

         row:
             {
                : row[],
                : row[],
                : row[],
                : row[],
                : (row[])  row[]  ,
                : row[]
            }
         

      () -> :
        
        stmt = Item.__table__.insert()
        result =  .session.execute(stmt, items_data)
         .session.commit()
         result.rowcount

      () -> [[, ]]:
        
        conditions = []
        params = {}

         filters.get():
            conditions.append()
            params[] = filters[]

         filters.get():
            conditions.append()
            params[] = filters[]

         filters.get():
            conditions.append()
            params[] = filters[]

         filters.get():
            conditions.append()
            params[] = filters[]

        where_clause =   conditions  

        sql = 

        params[] = filters.get(, )
        params[] = filters.get(, )

        result =  .session.execute(text(sql), params)
        rows = result.fetchall()

         [
            {
                : row[],
                : row[],
                : row[],
                : (row[]),
                : row[],
                : row[],
                : row[],
                : row[],
                : row[]
            }
             row  rows
        ]

---

Deployment and Testing Patterns

Production Deployment Configuration

# gunicorn.conf.py - Production WSGI configuration
import multiprocessing

# Server socket
bind = "0.0.0.0:8000"
backlog = 2048

# Worker processes
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
timeout = 30
keepalive = 2

# Restart settings
max_requests = 1000
max_requests_jitter = 100
preload_app = True
reload = False  # Disable reload in production

# Logging
accesslog = "-"
errorlog = "-"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# Process naming
proc_name = "fastapi_app"

# Security
limit_request_line = 4094
limit_request_fields = 100
limit_request_field_size = 8190

Docker Configuration for Production

# Dockerfile
FROM python:3.11-slim

# Install system dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# Install uv
RUN pip install uv

# Set working directory
WORKDIR /app

# Copy requirements and install dependencies
COPY requirements.txt .
RUN uv pip install -r requirements.txt

# Copy application code
COPY . .

# Create non-root user
RUN useradd --create-home --shell /bin/bash app \
    && chown -R app:app /app
USER app

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Run the application
CMD ["gunicorn", "main:app", "-c", "gunicorn.conf.py"]

Docker Compose for Development and Production

# docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:password@db:5432/myapp
      - REDIS_URL=redis://redis:6379/0
      - SECRET_KEY=your-super-secret-key-here
      - DEBUG=False
    depends_on:
      - db
      - redis
    volumes:
      - ./logs:/app/logs  # For log persistence
    restart: unless-stopped

  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=myapp
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    ports:
      - "5432:5432"
    restart: unless-stopped

  redis:
    image: 
    
       
    
       
     

  
     
    
       
       
    
       
       
    
       
     


  
  

Environment Configuration

# config.py - Application configuration
import os
from pydantic import BaseModel, Field
from typing import Optional, List

class Settings(BaseModel):
    # App settings
    app_name: str = "My FastAPI App"
    app_version: str = "1.0.0"
    debug: bool = Field(default=False, env="DEBUG")
    environment: str = Field(default="development", env="ENVIRONMENT")

    # Database settings
    database_url: str = Field(..., env="DATABASE_URL")
    database_pool_size: int = Field(default=20, env="DATABASE_POOL_SIZE")
    database_max_overflow: int = Field(default=30, env="DATABASE_MAX_OVERFLOW")

    # Security settings
    secret_key: str = Field(..., env="SECRET_KEY")
    algorithm: str = Field(default="HS256", env="ALGORITHM")
    access_token_expire_minutes: int = Field(default=30, env="ACCESS_TOKEN_EXPIRE_MINUTES")
    refresh_token_expire_days: int = Field(default=7, env="REFRESH_TOKEN_EXPIRE_DAYS")

    # CORS settings
    allowed_origins: List[str] = Field(default=[], env=)

    
    redis_url: [] = Field(default=, env=)

    
    log_level:  = Field(default=, env=)
    log_format:  = Field(default=, env=)

    
    sentry_dsn: [] = Field(default=, env=)
    mailgun_api_key: [] = Field(default=, env=)

     :
        env_file = 
        env_file_encoding = 
        case_sensitive = 


settings = Settings()


 contextlib  asynccontextmanager
 fastapi  FastAPI
 logging


logging.basicConfig(
    level=(logging, settings.log_level.upper()),
    =settings.log_format
)


  ():
    
    ()

    
     init_db()

    
     settings.redis_url:
        
        

    

    
    ()

app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    debug=settings.debug,
    lifespan=lifespan
)

Testing Framework Setup

# conftest.py - Pytest configuration
import pytest
import asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.main import app
from app.database import Base
import os

# Use in-memory SQLite for testing
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for each test case."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session", autouse=True)
async def create_test_database():
    """Create test database before tests and clean up after."""
    engine = create_async_engine(
        TEST_DATABASE_URL,
        poolclass=StaticPool,
        connect_args={"check_same_thread": False}
    )

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    yield engine

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

     engine.dispose()


  ():
    
      AsyncClient(app=app, base_url=)  ac:
         ac


  ():
    
    engine = create_async_engine(
        TEST_DATABASE_URL,
        poolclass=StaticPool,
        connect_args={: }
    )

    async_session = sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=
    )

      async_session()  session:
         session
         session.rollback()


 pytest
 httpx  AsyncClient


  ():
    
    response =  async_client.get()
     response.status_code == 
       response.json()


  ():
    
    response =  async_client.get()
     response.status_code == 
     response.json()[] == 


 app.auth  create_access_token
 jwt


  ():
    
    data = {: }
    token = create_access_token(data=data)

    
    payload = jwt.decode(token, os.getenv(), algorithms=[])
     payload[] == 


  ():
    
    
    token = create_access_token(data={: })

    
    response =  async_client.get(
        ,
        headers={: }
    )
     response.status_code == 


 app.models  User
 app.database  get_async_db
 sqlalchemy  select


  ():
    
    
    user = User(
        email=,
        username=,
        hashed_password=
    )
    db_session.add(user)
     db_session.commit()
     db_session.refresh(user)

    
    result =  db_session.execute(select(User).where(User. == user.))
    retrieved_user = result.scalar_one_or_none()

     retrieved_user   
     retrieved_user.email == 
     retrieved_user.username == 


 app.models  Item
 app.schemas  ItemCreate


  ():
    
    
    user = User(
        email=,
        username=,
        hashed_password=
    )
    db_session.add(user)
     db_session.commit()
     db_session.refresh(user)

    
    item_data = {
        : ,
        : ,
        : ,
        : user.
    }

    
    response =  async_client.post(, json=item_data)
     response.status_code == 

    item_response = response.json()
     item_response[] == 
     item_response[] == 
     item_response[] == user.



  ():
    
    
    user_data = {
        : ,
        : ,
        : ,
        : 
    }

    response =  async_client.post(, json=user_data)
     response.status_code == 

    
    login_data = {
        : ,
        : 
    }

    response =  async_client.post(, data=login_data)
     response.status_code == 

    token_data = response.json()
       token_data
     token_data[] == 

Performance Testing

# load_test.py - Load testing with Locust
from locust import HttpUser, task, between
import json

class FastAPIUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        """Login before starting tasks."""
        self.login()

    def login(self):
        """Login to get JWT token."""
        response = self.client.post("/token", data={
            "username": "testuser",
            "password": "testpassword"
        })

        if response.status_code == 200:
            token_data = response.json()
            self.token = token_data["access_token"]
            self.headers = {"Authorization": f"Bearer {self.token}"}
        else:
            self.token = None
            self.headers = {}

    @task(3)
    def get_items(self):
        """Get items endpoint."""
        self.client.get("/items/", headers=self.headers)


     ():
        
        item_data = {
            : ,
            : ,
            : 
        }
        .client.post(, json=item_data, headers=.headers)


     ():
        
        .client.get(, headers=.headers)


 ():
    
     ():
        
        user_data = {
            : ,
            : ,
            : 
        }
        
        
        

    result = benchmark(create_user)
    
     benchmark.stats[] <   


 pytest
 httpx  AsyncClient


  ():
    
    
     i  ():
        response =  async_client.get()

    
     response.status_code ==   


  ():
    
    malicious_payload = {
        : ,
        : 
    }

    
    response =  async_client.get(, params=malicious_payload)
    
     response.status_code != 



  ():
    
    response =  async_client.get()
     response.status_code == 


  ():
    
    invalid_data = {
        : ,  
        : -  
    }

    response =  async_client.post(, json=invalid_data)
     response.status_code ==