FastAPI Python Setup and Configuration

Overview

FastAPI setup with modern async patterns and high-performance API development, following AzmX development standards for scalable microservices architecture.

Core Dependencies

From typical FastAPI requirements.txt:

fastapi==0.110.3
uvicorn==0.26.0                  # ASGI server
gunicorn==21.2.0                 # Production server
pydantic==2.5.3                  # Data validation
sqlalchemy==2.0.25               # ORM with async support
asyncpg==0.29.0                  # PostgreSQL async adapter
alembic==1.13.1                  # Database migrations
redis==5.0.1                     # Caching and sessions
httpx==0.27.0                    # Async HTTP client
python-jose==3.3.0               # JWT authentication
python-multipart==0.0.6          # File upload support

Project Structure

app/
├── main.py                      # FastAPI application entry point
├── requirements.txt
├── .env
├── alembic/                     # Database migrations
│   ├── alembic.ini
│   └── versions/
├── app/
│   ├── __init__.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py            # Settings configuration
│   │   ├── database.py          # Database connection
│   │   └── security.py          # Authentication utilities
│   ├── api/
│   │   ├── __init__.py
│   │   ├── deps.py              # Dependencies
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── api.py           # API router
│   │       └── endpoints/
│   │           ├── auth.py
│   │           ├── users.py
│   │           └── items.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py              # Base model classes
│   │   ├── user.py
│   │   └── item.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py              # Pydantic schemas
│   │   └── item.py
│   └── services/
│       ├── __init__.py
│       ├── user.py              # Business logic
│       └── item.py
└── tests/
    ├── __init__.py
    ├── conftest.py
    └── api/

Settings Configuration

Core Configuration (app/core/config.py)

from pydantic_settings import BaseSettings
from pydantic import Field, validator
from typing import List, Optional
from functools import lru_cache
import os

class Settings(BaseSettings):
    """Application settings."""

    # Application
    APP_NAME: str = Field(default="AzmX FastAPI")
    APP_VERSION: str = Field(default="1.0.0")
    DEBUG: bool = Field(default=False)
    API_V1_STR: str = Field(default="/api/v1")

    # Database
    DATABASE_URL: str = Field(..., env="DATABASE_URL")
    DATABASE_POOL_SIZE: int = Field(default=10)
    DATABASE_MAX_OVERFLOW: int = Field(default=20)

    # Security
    SECRET_KEY: str = Field(..., min_length=32)
    ACCESS_TOKEN_EXPIRE_MINUTES: int = Field(default=30)
    REFRESH_TOKEN_EXPIRE_DAYS: int = Field(default=7)
    ALGORITHM: str = Field(default="HS256")

    # CORS
    BACKEND_CORS_ORIGINS: List[str] = Field(
        default=["http://localhost:3000", "http://localhost:3001"]
    )

    # Redis
    REDIS_URL: Optional[str] = Field(default=None)
    CACHE_TTL: int = Field(default=300)

    # Environment specific
    ENVIRONMENT: str = Field(default="development")

    @validator("BACKEND_CORS_ORIGINS", pre=True)
    def assemble_cors_origins(cls, v):
        if isinstance(v, str) and not v.startswith("["):
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, (list, str)):
            return v
        raise ValueError(v)

    class Config:
        env_file = ".env"
        case_sensitive = True

@lru_cache()
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

Database Configuration (app/core/database.py)

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from .config import settings

# Async engine for FastAPI
async_engine = create_async_engine(
    settings.DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://"),
    pool_size=settings.DATABASE_POOL_SIZE,
    max_overflow=settings.DATABASE_MAX_OVERFLOW,
    echo=settings.DEBUG,
)

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

Base = declarative_base()

# Dependency
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

FastAPI Application Setup

Main Application (main.py)

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager

from app.core.config import settings
from app.api.v1.api import api_router


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Manage application lifespan events."""
    # Startup
    print(f"Starting {settings.APP_NAME}")
    yield
    # Shutdown
    print("Shutting down application")


def create_app() -> FastAPI:
    """Create and configure FastAPI application."""
    app = FastAPI(
        title=settings.APP_NAME,
        version=settings.APP_VERSION,
        openapi_url=f"{settings.API_V1_STR}/openapi.json",
        docs_url="/docs" if settings.DEBUG else None,
        redoc_url="/redoc" if settings.DEBUG else None,
        lifespan=lifespan
    )

    # Configure CORS
    if settings.BACKEND_CORS_ORIGINS:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"],
        )

    # Include routers
    app.include_router(api_router, prefix=settings.API_V1_STR)

    return app

app = create_app()

@app.get("/")
def read_root():
    """Health check endpoint."""
    return {"message": f"{settings.APP_NAME} is running"}


@app.get("/health")
def health_check():
    """Detailed health check."""
    return {
        "status": "healthy",
        "app_name": settings.APP_NAME,
        "version": settings.APP_VERSION,
        "environment": settings.ENVIRONMENT
    }

API Router Configuration (app/api/v1/api.py)

from fastapi import APIRouter
from app.api.v1.endpoints import auth, users, items

api_router = APIRouter()

api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(items.router, prefix="/items", tags=["items"])

Environment Configuration

Development Environment (.env.development)

# Application
APP_NAME=AzmX FastAPI Development
DEBUG=true
ENVIRONMENT=development

# Database
DATABASE_URL=postgresql://postgres:password@localhost:5433/fastapi_dev

# Security
SECRET_KEY=your-super-secret-key-for-development-minimum-32-chars
ACCESS_TOKEN_EXPIRE_MINUTES=60

# CORS
BACKEND_CORS_ORIGINS=http://localhost:3000,http://localhost:3001,http://localhost:8080

# Redis
REDIS_URL=redis://localhost:6379/0
CACHE_TTL=300

# Logging
LOG_LEVEL=INFO

Production Environment (.env.production)

# Application
APP_NAME=AzmX FastAPI Production
DEBUG=false
ENVIRONMENT=production

# Database
DATABASE_URL=postgresql://user:password@prod-host:5432/fastapi_prod
DATABASE_POOL_SIZE=20
DATABASE_MAX_OVERFLOW=40

# Security (use strong secrets in production)
SECRET_KEY=${PRODUCTION_SECRET_KEY}
ACCESS_TOKEN_EXPIRE_MINUTES=30

# CORS (restrict to actual frontend domains)
BACKEND_CORS_ORIGINS=https://app.azmx.sa,https://admin.azmx.sa

# Redis
REDIS_URL=${PRODUCTION_REDIS_URL}
CACHE_TTL=600

# Monitoring
SENTRY_DSN=${SENTRY_DSN}
LOG_LEVEL=WARNING

Pydantic Schemas

User Schemas (app/schemas/user.py)

from pydantic import BaseModel, EmailStr, Field, ConfigDict
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    """Base user schema."""
    email: EmailStr
    first_name: str = Field(..., min_length=1, max_length=50)
    last_name: str = Field(..., min_length=1, max_length=50)
    is_active: bool = True

class UserCreate(UserBase):
    """User creation schema."""
    password: str = Field(..., min_length=8, max_length=100)

class UserUpdate(BaseModel):
    """User update schema."""
    first_name: Optional[str] = Field(None, min_length=1, max_length=50)
    last_name: Optional[str] = Field(None, min_length=1, max_length=50)
    email: Optional[EmailStr] = None
    is_active: Optional[bool] = None

class UserInDB(UserBase):
    """User as stored in database."""
    model_config = ConfigDict(from_attributes=True)

    id: int
    created_at: datetime
    updated_at: datetime

class User(UserInDB):
    """User response schema."""
    pass

# Authentication schemas
class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: Optional[int] = None

Authentication Setup

Security Utilities (app/core/security.py)

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import HTTPException, status
from .config import settings

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

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """Create JWT access token."""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(
            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
        )
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(
        to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
    )
    return encoded_jwt

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify password."""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """Hash password."""
    return pwd_context.hash(password)

def decode_access_token(token: str):
    """Decode and validate JWT token."""
    try:
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
        )
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials",
            )
        return user_id
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate credentials",
        )

Deployment Configuration

Gunicorn Configuration (gunicorn.conf.py)

import multiprocessing
import os

# 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
max_requests = 1000
max_requests_jitter = 100

# Timeout
timeout = 30
keepalive = 2

# Logging
loglevel = os.getenv("LOG_LEVEL", "info")
accesslog = "-"
errorlog = "-"

# Process naming
proc_name = "fastapi-azmx"

# Server mechanics
preload_app = True
daemon = False
pidfile = "/tmp/gunicorn.pid"
user = None
group = None
tmp_upload_dir = None

# SSL (if needed)
# keyfile = "/path/to/keyfile"
# certfile = "/path/to/certfile"

Docker Configuration (Dockerfile)

FROM python:3.11-slim

WORKDIR /app

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

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Create non-root user
RUN adduser --disabled-password --gecos '' appuser
RUN chown -R appuser:appuser /app
USER appuser

EXPOSE 8000

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

CMD ["gunicorn", "--config", "gunicorn.conf.py", "main:app"]

Testing Configuration

Test Configuration (tests/conftest.py)

import pytest
import asyncio
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from httpx import AsyncClient

from main import app
from app.core.database import get_db, Base
from app.core.config import settings

# Test database URL
TEST_DATABASE_URL = "postgresql+asyncpg://postgres:password@localhost:5433/test_db"

# Test engine and session
test_engine = create_async_engine(TEST_DATABASE_URL, echo=True)
TestSessionLocal = sessionmaker(
    test_engine, class_=AsyncSession, expire_on_commit=False
)

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

@pytest.fixture(scope="session")
async def setup_database():
    """Setup test database."""
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def db_session(setup_database):
    """Create test database session."""
    async with TestSessionLocal() as session:
        yield session

@pytest.fixture
def client():
    """Create test client."""
    return TestClient(app)

@pytest.fixture
async def async_client():
    """Create async test client."""
    async with AsyncClient(app=app, base_url="http://test") as client:
        yield client

Performance Optimizations

Database Query Optimization

# Use async/await properly
from sqlalchemy import select
from sqlalchemy.orm import selectinload

async def get_user_with_items(db: AsyncSession, user_id: int):
    """Get user with related items efficiently."""
    result = await db.execute(
        select(User)
        .options(selectinload(User.items))
        .where(User.id == user_id)
    )
    return result.scalar_one_or_none()

# Batch operations
async def create_multiple_items(db: AsyncSession, items_data: List[dict]):
    """Create multiple items efficiently."""
    items = [Item(**item_data) for item_data in items_data]
    db.add_all(items)
    await db.commit()
    return items

Caching Configuration

# Redis caching
from redis.asyncio import Redis
from typing import Optional
import json

redis_client: Optional[Redis] = None

async def init_redis():
    """Initialize Redis connection."""
    global redis_client
    if settings.REDIS_URL:
        redis_client = Redis.from_url(settings.REDIS_URL, decode_responses=True)

async def cache_get(key: str):
    """Get value from cache."""
    if redis_client:
        value = await redis_client.get(key)
        return json.loads(value) if value else None
    return None

async def cache_set(key: str, value, ttl: int = settings.CACHE_TTL):
    """Set value in cache."""
    if redis_client:
        await redis_client.set(key, json.dumps(value, default=str), ex=ttl)

Development Commands

Useful Development Scripts

# Development server
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# Run tests
pytest tests/ -v

# Run tests with coverage
pytest tests/ --cov=app --cov-report=html

# Database migrations
alembic init alembic                    # Initialize migrations
alembic revision --autogenerate -m "Initial migration"
alembic upgrade head                    # Apply migrations

# Format code
black app/ tests/
isort app/ tests/

# Lint code
flake8 app/ tests/
mypy app/

# Generate requirements
pip-compile requirements.in             # If using pip-tools

References

• FastAPI Documentation
• Pydantic Documentation
• SQLAlchemy Async Documentation
• Alembic Documentation