Ep.03 FastAPI Project Structure: Scale Your API with Pydantic

Views: 2

To build a professional FastAPI project structure, use a modular architecture that separates concerns into dedicated directories: /app/api for routes, /app/models for Pydantic validation, and /app/services for business logic. This layout is managed through an Application Factory pattern and centralized configuration using pydantic-settings with .env files. This approach ensures your API is scalable, secure, and production-ready by decoupling environment-specific settings from your core application logic.

🎓 What You’ll Learn

By the end of this tutorial, you’ll understand:

• How to organize a FastAPI project for scalability
• Configuration management with environment variables
• Breaking your API into modular routers
• Settings management with Pydantic
• Best practices for production-ready code structure

---

📖 Why Project Structure Matters

The Problem with Single-File Applications

Our ep_02.py has everything in one file:

• Models
• Endpoints
• Business logic
• Configuration

What happens as your project grows?

ep_02.py  (150 lines)
    ↓
main.py  (500 lines)
    ↓
main.py  (2000 lines)  ❌ Unmaintainable!

Benefits of Proper Structure

Benefit	Why It Matters
Maintainability	Easy to find and fix bugs
Scalability	Add features without breaking existing code
Team Collaboration	Multiple developers can work simultaneously
Testing	Easier to write unit tests
Reusability	Share code across projects
Security	Separate secrets from code

---

🏗️ Professional FastAPI Project Structure

Here’s the structure we’ll build:

fastapi-ai-backend/
├── app/                          # Main application package
│   ├── __init__.py              # Makes 'app' a Python package
│   ├── main.py                  # Application entry point
│   ├── config.py                # Configuration & settings
│   ├── dependencies.py          # Shared dependencies
│   │
│   ├── api/                     # API routes
│   │   ├── __init__.py
│   │   ├── v1/                  # API version 1
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│   │   │   │   ├── __init__.py
│   │   │   │   ├── users.py     # User endpoints
│   │   │   │   ├── health.py    # Health check endpoints
│   │   │   │   └── ai.py        # AI endpoints (future)
│   │   │   └── api.py           # Router aggregator
│   │
│   ├── models/                  # Pydantic models
│   │   ├── __init__.py
│   │   ├── user.py              # User models
│   │   └── common.py            # Shared models
│   │
│   ├── schemas/                 # Database schemas (future)
│   │   └── __init__.py
│   │
│   ├── services/                # Business logic
│   │   ├── __init__.py
│   │   └── user_service.py      # User business logic
│   │
│   ├── core/                    # Core functionality
│   │   ├── __init__.py
│   │   ├── config.py            # Settings class
│   │   └── security.py          # Security utilities (future)
│   │
│   └── utils/                   # Utility functions
│       ├── __init__.py
│       └── logger.py            # Logging configuration
│
├── tests/                       # Test files
│   ├── __init__.py
│   └── test_users.py
│
├── .env                         # Environment variables (not in git!)
├── .env.example                 # Example env file (in git)
├── .gitignore
├── requirements.txt
├── README.md
└── main.py                      # Entry point (imports from app/)

---

🛠️ Step-by-Step Implementation

Step 1: Create the Project Structure

Let’s create all the folders and files:

# Make sure you're in your project root and venv is activated
cd ~/Documents/fastapi-ai-backend  # Adjust path as needed
source venv/bin/activate  # or venv\Scripts\activate on Windows

# Create directory structure
mkdir -p app/api/v1/endpoints
mkdir -p app/models
mkdir -p app/schemas
mkdir -p app/services
mkdir -p app/core
mkdir -p app/utils
mkdir -p tests

# Create __init__.py files (makes directories into Python packages)
touch app/__init__.py
touch app/api/__init__.py
touch app/api/v1/__init__.py
touch app/api/v1/endpoints/__init__.py
touch app/models/__init__.py
touch app/schemas/__init__.py
touch app/services/__init__.py
touch app/core/__init__.py
touch app/utils/__init__.py
touch tests/__init__.py

# On Windows, use this instead of 'touch':
# type nul > app/__init__.py
# (repeat for each __init__.py)

What is __init__.py?

• Makes a directory into a Python package
• Can be empty or contain initialization code
• Allows you to import from that directory
• Example: from app.models.user import User

---

Step 2: Environment Variables & Configuration

Why Environment Variables?

• Store secrets (API keys, database passwords)
• Different settings for development/production
• Don’t commit secrets to Git!

Create .env file in project root:

# .env
# This file contains sensitive information - NEVER commit to Git!

# Application Settings
APP_NAME="FastAPI AI Backend"
APP_VERSION="0.3.0"
DEBUG=True
ENVIRONMENT=development

# API Settings
API_V1_PREFIX=/api/v1
HOST=0.0.0.0
PORT=8000

# CORS Settings (we'll use this later)
ALLOWED_ORIGINS=http://localhost:3000,http://localhost:8000

# AI Settings (for future use)
OLLAMA_BASE_URL=http://localhost:11434
DEFAULT_AI_MODEL=llama2

# Database (for future use)
DATABASE_URL=sqlite:///./app.db

# Security (for future use)
SECRET_KEY=your-secret-key-change-this-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

Create .env.example (safe to commit to Git):

bash

# .env.example
# Copy this to .env and fill in your actual values

APP_NAME="FastAPI AI Backend"
APP_VERSION="0.3.0"
DEBUG=True
ENVIRONMENT=development

API_V1_PREFIX=/api/v1
HOST=0.0.0.0
PORT=8000

ALLOWED_ORIGINS=http://localhost:3000,http://localhost:8000

OLLAMA_BASE_URL=http://localhost:11434
DEFAULT_AI_MODEL=llama2

DATABASE_URL=sqlite:///./app.db

SECRET_KEY=change-this-to-a-random-secret-key
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

Update .gitignore:

# Add to .gitignore
.env
.env.local

---

Step 3: Install Required Packages

pip install python-dotenv pydantic-settings
pip freeze > requirements.txt

What we installed:

• python-dotenv: Loads environment variables from .env file
• pydantic-settings: Validates and manages settings with Pydantic

---

Step 4: Create Core Configuration

Create app/core/config.py:

"""
Core configuration module
Manages all application settings using Pydantic Settings
"""

from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import List
from functools import lru_cache


class Settings(BaseSettings):
    """
    Application settings
    
    These values are loaded from environment variables or .env file
    Pydantic validates the types automatically
    """
    
    # Application
    APP_NAME: str = "FastAPI AI Backend"
    APP_VERSION: str = "0.3.0"
    DEBUG: bool = False
    ENVIRONMENT: str = "production"
    
    # API
    API_V1_PREFIX: str = "/api/v1"
    HOST: str = "0.0.0.0"
    PORT: int = 8000
    
    # CORS
    ALLOWED_ORIGINS: str = "http://localhost:3000"
    
    @property
    def allowed_origins_list(self) -> List[str]:
        """Convert comma-separated string to list"""
        return [origin.strip() for origin in self.ALLOWED_ORIGINS.split(",")]
    
    # AI Settings
    OLLAMA_BASE_URL: str = "http://localhost:11434"
    DEFAULT_AI_MODEL: str = "llama2"
    
    # Database (for future use)
    DATABASE_URL: str = "sqlite:///./app.db"
    
    # Security (for future use)
    SECRET_KEY: str = "change-this-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    # Pydantic Settings Configuration
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=True,
        extra="ignore"  # Ignore extra fields in .env
    )


@lru_cache()
def get_settings() -> Settings:
    """
    Create settings instance (cached)
    
    @lru_cache ensures we only create one Settings instance
    and reuse it throughout the application
    
    Returns:
        Settings: Application settings
    """
    return Settings()


# Convenience: Get settings instance
settings = get_settings()

🔍 Key Concepts Explained:

1. BaseSettings: Pydantic class that loads from environment variables
2. @property: Converts method into attribute (computed field)
3. @lru_cache():
   • Least Recently Used cache
   • Stores function result
   • Next call returns cached value (faster!)
   • Ensures we have ONE settings instance app-wide
4. model_config: Tells Pydantic:
   • Where to find .env file
   • How to handle extra variables
   • Case sensitivity rules

---

Step 5: Create Common Models

Create app/models/common.py:

"""
Common Pydantic models used across the application
"""

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


class HealthCheck(BaseModel):
    """Health check response model"""
    status: str = Field(..., description="Service status")
    version: str = Field(..., description="API version")
    environment: str = Field(..., description="Environment (dev/prod)")
    timestamp: datetime = Field(default_factory=datetime.now)
    
    class Config:
        json_schema_extra = {
            "example": {
                "status": "healthy",
                "version": "0.3.0",
                "environment": "development",
                "timestamp": "2024-01-15T10:30:00"
            }
        }


class MessageResponse(BaseModel):
    """Generic message response"""
    message: str = Field(..., description="Response message")
    success: bool = Field(default=True, description="Operation success status")
    
    class Config:
        json_schema_extra = {
            "example": {
                "message": "Operation completed successfully",
                "success": True
            }
        }


class ErrorResponse(BaseModel):
    """Error response model"""
    detail: str = Field(..., description="Error details")
    error_code: Optional[str] = Field(None, description="Error code")
    
    class Config:
        json_schema_extra = {
            "example": {
                "detail": "Resource not found",
                "error_code": "NOT_FOUND"
            }
        }

---

Step 6: Create User Models

Create app/models/user.py:

"""
User-related Pydantic models
"""

from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List
from datetime import datetime
from enum import Enum


class UserRole(str, Enum):
    """User role enumeration"""
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"


class UserBase(BaseModel):
    """Base user model with common fields"""
    username: str = Field(..., min_length=3, max_length=50, description="Username")
    email: EmailStr = Field(..., description="Email address")
    full_name: Optional[str] = Field(None, description="Full name")
    role: UserRole = Field(default=UserRole.USER, description="User role")


class UserCreate(UserBase):
    """Model for creating a new user"""
    password: str = Field(..., min_length=8, description="Password")
    
    class Config:
        json_schema_extra = {
            "example": {
                "username": "johndoe",
                "email": "john@example.com",
                "full_name": "John Doe",
                "password": "securepass123",
                "role": "user"
            }
        }


class UserUpdate(BaseModel):
    """Model for updating a user (all fields optional)"""
    username: Optional[str] = Field(None, min_length=3, max_length=50)
    email: Optional[EmailStr] = None
    full_name: Optional[str] = None
    role: Optional[UserRole] = None
    is_active: Optional[bool] = None
    tags: Optional[List[str]] = None


class User(UserBase):
    """Complete user model (internal)"""
    id: int = Field(..., description="User ID", gt=0)
    is_active: bool = Field(default=True, description="Is user active")
    created_at: datetime = Field(default_factory=datetime.now)
    tags: List[str] = Field(default=[])
    
    class Config:
        from_attributes = True  # Allows creation from ORM models (future)


class UserResponse(UserBase):
    """User response model (public-facing, no password)"""
    id: int
    is_active: bool
    created_at: datetime
    tags: List[str]
    
    class Config:
        from_attributes = True
        json_schema_extra = {
            "example": {
                "id": 1,
                "username": "johndoe",
                "email": "john@example.com",
                "full_name": "John Doe",
                "role": "user",
                "is_active": True,
                "created_at": "2024-01-15T10:30:00",
                "tags": ["developer", "python"]
            }
        }

🔍 Model Inheritance Pattern:

UserBase (common fields)
    ├── UserCreate (+ password)
    ├── User (+ id, is_active, created_at, tags)
    └── UserResponse (public version of User)

UserUpdate (all optional, separate hierarchy)

Why this pattern?

• DRY: Don’t Repeat Yourself – shared fields in base
• Security: UserResponse never includes password
• Flexibility: Each use case gets the right fields

---

Step 7: Create User Service (Business Logic)

Create app/services/user_service.py:

"""
User service - Business logic for user operations

This separates business logic from API endpoints
"""

from typing import List, Optional
from datetime import datetime
from fastapi import HTTPException, status

from app.models.user import User, UserCreate, UserUpdate, UserRole


class UserService:
    """
    User service class
    
    Handles all user-related business logic
    In a real app, this would interact with a database
    """
    
    def __init__(self):
        """Initialize the service with in-memory storage"""
        self._users: List[User] = []
        self._next_id: int = 1
    
    def create_user(self, user_data: UserCreate) -> User:
        """
        Create a new user
        
        Args:
            user_data: User creation data
        
        Returns:
            Created user
        
        Raises:
            HTTPException: If username already exists
        """
        # Check for duplicate username
        if self.get_user_by_username(user_data.username):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Username '{user_data.username}' already exists"
            )
        
        # Check for duplicate email
        if self.get_user_by_email(user_data.email):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Email '{user_data.email}' already registered"
            )
        
        # Create new user
        new_user = User(
            id=self._next_id,
            username=user_data.username,
            email=user_data.email,
            full_name=user_data.full_name,
            role=user_data.role,
            is_active=True,
            created_at=datetime.now(),
            tags=[]
        )
        
        self._users.append(new_user)
        self._next_id += 1
        
        return new_user
    
    def get_all_users(
        self,
        skip: int = 0,
        limit: int = 10,
        role: Optional[UserRole] = None
    ) -> List[User]:
        """
        Get all users with optional filtering
        
        Args:
            skip: Number of records to skip (pagination)
            limit: Maximum records to return
            role: Filter by role
        
        Returns:
            List of users
        """
        users = self._users
        
        # Filter by role if specified
        if role:
            users = [u for u in users if u.role == role]
        
        # Apply pagination
        return users[skip : skip + limit]
    
    def get_user_by_id(self, user_id: int) -> User:
        """
        Get user by ID
        
        Args:
            user_id: User identifier
        
        Returns:
            User object
        
        Raises:
            HTTPException: If user not found
        """
        for user in self._users:
            if user.id == user_id:
                return user
        
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User with ID {user_id} not found"
        )
    
    def get_user_by_username(self, username: str) -> Optional[User]:
        """
        Get user by username
        
        Args:
            username: Username to search for
        
        Returns:
            User if found, None otherwise
        """
        for user in self._users:
            if user.username == username:
                return user
        return None
    
    def get_user_by_email(self, email: str) -> Optional[User]:
        """
        Get user by email
        
        Args:
            email: Email to search for
        
        Returns:
            User if found, None otherwise
        """
        for user in self._users:
            if user.email == email:
                return user
        return None
    
    def update_user(self, user_id: int, user_update: UserUpdate) -> User:
        """
        Partially update a user
        
        Args:
            user_id: User to update
            user_update: Fields to update
        
        Returns:
            Updated user
        
        Raises:
            HTTPException: If user not found
        """
        user = self.get_user_by_id(user_id)
        
        # Get only the fields that were set
        update_data = user_update.model_dump(exclude_unset=True)
        
        # Check for username conflict (if username is being updated)
        if "username" in update_data:
            existing = self.get_user_by_username(update_data["username"])
            if existing and existing.id != user_id:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="Username already exists"
                )
        
        # Check for email conflict (if email is being updated)
        if "email" in update_data:
            existing = self.get_user_by_email(update_data["email"])
            if existing and existing.id != user_id:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail="Email already registered"
                )
        
        # Apply updates
        for field, value in update_data.items():
            setattr(user, field, value)
        
        return user
    
    def delete_user(self, user_id: int) -> None:
        """
        Delete a user
        
        Args:
            user_id: User to delete
        
        Raises:
            HTTPException: If user not found
        """
        for idx, user in enumerate(self._users):
            if user.id == user_id:
                self._users.pop(idx)
                return
        
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"User with ID {user_id} not found"
        )
    
    def get_user_count(self) -> int:
        """Get total number of users"""
        return len(self._users)


# Create a global instance (singleton pattern)
user_service = UserService()

🔍 Service Layer Pattern:

Why separate business logic?

• ✅ Single Responsibility: Endpoints handle HTTP, services handle logic
• ✅ Reusability: Use same logic from multiple endpoints
• ✅ Testability: Test business logic without HTTP layer
• ✅ Maintainability: Changes to logic don’t affect API structure

---

Step 8: Create Health Check Endpoint

Create app/api/v1/endpoints/health.py:

"""
Health check endpoints
"""

from fastapi import APIRouter, Depends
from app.models.common import HealthCheck
from app.core.config import Settings, get_settings

router = APIRouter(tags=["Health"])


@router.get("/health", response_model=HealthCheck)
async def health_check(settings: Settings = Depends(get_settings)):
    """
    Health check endpoint
    
    Returns current system status and version information
    """
    return HealthCheck(
        status="healthy",
        version=settings.APP_VERSION,
        environment=settings.ENVIRONMENT
    )


@router.get("/")
async def root(settings: Settings = Depends(get_settings)):
    """Root endpoint"""
    return {
        "message": f"Welcome to {settings.APP_NAME}",
        "version": settings.APP_VERSION,
        "docs": "/docs",
        "health": "/api/v1/health"
    }

🔍 New Concept: Depends()

settings: Settings = Depends(get_settings)

What is Dependency Injection?

• FastAPI calls get_settings() automatically
• Result is passed to your function
• Function is called once per request (unless cached)
• Promotes code reuse and testability

Real-world analogy: Instead of going to the kitchen yourself to get ingredients (settings), you declare “I need flour” and someone brings it to you.

---

Step 9: Create User Endpoints

Create app/api/v1/endpoints/users.py:

"""
User management endpoints
"""

from fastapi import APIRouter, Depends, status
from typing import List, Optional

from app.models.user import (
    User,
    UserCreate,
    UserUpdate,
    UserResponse,
    UserRole
)
from app.services.user_service import UserService, user_service

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


def get_user_service() -> UserService:
    """
    Dependency to get user service instance
    
    In the future, this could create a new instance per request
    or handle database sessions
    """
    return user_service


@router.post(
    "",
    response_model=UserResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create a new user"
)
async def create_user(
    user_data: UserCreate,
    service: UserService = Depends(get_user_service)
):
    """
    Create a new user with the following information:
    
    - **username**: Unique username (3-50 characters)
    - **email**: Valid email address
    - **full_name**: Optional full name
    - **password**: Password (min 8 characters)
    - **role**: User role (admin, user, guest)
    """
    return service.create_user(user_data)


@router.get(
    "",
    response_model=List[UserResponse],
    summary="Get all users"
)
async def get_users(
    skip: int = 0,
    limit: int = 10,
    role: Optional[UserRole] = None,
    service: UserService = Depends(get_user_service)
):
    """
    Retrieve users with optional filtering:
    
    - **skip**: Number of records to skip (for pagination)
    - **limit**: Maximum number of records to return
    - **role**: Filter by user role
    """
    return service.get_all_users(skip=skip, limit=limit, role=role)


@router.get(
    "/{user_id}",
    response_model=UserResponse,
    summary="Get user by ID"
)
async def get_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    """
    Get a specific user by their ID
    """
    return service.get_user_by_id(user_id)


@router.patch(
    "/{user_id}",
    response_model=UserResponse,
    summary="Update user"
)
async def update_user(
    user_id: int,
    user_update: UserUpdate,
    service: UserService = Depends(get_user_service)
):
    """
    Update user information (partial update)
    
    Only provided fields will be updated
    """
    return service.update_user(user_id, user_update)


@router.delete(
    "/{user_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Delete user"
)
async def delete_user(
    user_id: int,
    service: UserService = Depends(get_user_service)
):
    """
    Delete a user by ID
    """
    service.delete_user(user_id)
    return None


@router.get(
    "/stats/count",
    summary="Get user statistics"
)
async def get_user_stats(
    service: UserService = Depends(get_user_service)
):
    """
    Get user statistics
    """
    return {
        "total_users": service.get_user_count()
    }

🔍 Router Prefix:

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

• prefix: All routes get /users prepended
  • @router.get("") → /users
  • @router.get("/{user_id}") → /users/{user_id}
• tags: Groups endpoints in documentation

---

Step 10: Aggregate Routers

Create app/api/v1/api.py:

"""
API v1 router aggregator

Combines all v1 endpoint routers
"""

from fastapi import APIRouter
from app.api.v1.endpoints import users, health

# Create main v1 router
api_router = APIRouter()

# Include all endpoint routers
api_router.include_router(health.router)
api_router.include_router(users.router)

# Future routers will be added here:
# api_router.include_router(ai.router)
# api_router.include_router(chat.router)

Why this pattern?

• Central place to manage all routes
• Easy to add/remove entire feature sets
• Clean separation of concerns
• Version control (v1, v2, etc.)

---

Step 11: Create Main Application

Create app/main.py:

"""
FastAPI AI Backend - Main Application

This is the core application file that ties everything together
"""

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

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


def create_application() -> FastAPI:
    """
    Application factory pattern
    
    Creates and configures the FastAPI application
    
    Returns:
        FastAPI: Configured application instance
    """
    
    app = FastAPI(
        title=settings.APP_NAME,
        version=settings.APP_VERSION,
        description="""
        A production-ready FastAPI backend for AI applications.
        
        ## Features
        * User management with CRUD operations
        * RESTful API design
        * Automatic API documentation
        * Modular architecture
        * Environment-based configuration
        
        ## Coming Soon
        * AI model integration (Ollama)
        * Authentication & Authorization
        * Database integration
        * WebSocket support for streaming
        """,
        debug=settings.DEBUG,
        docs_url="/docs",
        redoc_url="/redoc",
        openapi_url="/openapi.json"
    )
    
    # Configure CORS
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.allowed_origins_list,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    # Include API routers
    app.include_router(
        api_router,
        prefix=settings.API_V1_PREFIX
    )
    
    return app


# Create the application instance
app = create_application()


# Optional: Add startup/shutdown events
@app.on_event("startup")
async def startup_event():
    """
    Code to run when the application starts
    """
    print(f"🚀 Starting {settings.APP_NAME} v{settings.APP_VERSION}")
    print(f"📝 Environment: {settings.ENVIRONMENT}")
    print(f"🔧 Debug mode: {settings.DEBUG}")
    print(f"📚 API Docs: http://{settings.HOST}:{settings.PORT}/docs")


@app.on_event("shutdown")
async def shutdown_event():
    """
    Code to run when the application shuts down
    """
    print(f"👋 Shutting down {settings.APP_NAME}")

🔍 Application Factory Pattern:

Instead of creating the app at module level:

app = FastAPI()  # ❌ Hard to test, configure

We use a function:

def create_application() -> FastAPI:
    app = FastAPI()
    # ... configuration ...
    return app

app = create_application()  # ✅ Flexible, testable

Benefits:

• Easy to create multiple app instances (testing)
• Centralized configuration
• Clean initialization logic

---

Step 12: Create Entry Point

Create main.py in project root (outside app/):

"""
Application entry point

Run with: uvicorn main:app --reload
"""

from app.main import app

# This file exists so we can run: uvicorn main:app
# which is simpler than: uvicorn app.main:app

if __name__ == "__main__":
    import uvicorn
    from app.core.config import settings
    
    uvicorn.run(
        "app.main:app",
        host=settings.HOST,
        port=settings.PORT,
        reload=settings.DEBUG
    )

---

Step 13: Create a README

Create README.md:

# FastAPI AI Backend

A production-ready FastAPI backend for AI applications with modular architecture.

## Features

- ✅ RESTful API design
- ✅ Automatic API documentation
- ✅ Environment-based configuration
- ✅ Modular router structure
- ✅ Service layer pattern
- ✅ Pydantic data validation
- ✅ CORS support
- 🔜 AI model integration (Ollama)
- 🔜 Database integration
- 🔜 Authentication & authorization

## Project Structure

fastapi-ai-backend/ ├── app/ # Main application package │ ├── api/ # API routes │ ├── core/ # Core functionality │ ├── models/ # Pydantic models │ ├── services/ # Business logic │ └── utils/ # Utilities ├── tests/ # Test files ├── .env # Environment variables ├── requirements.txt # Dependencies └── main.py # Entry point

## Setup

### 1. Clone and Navigate
```bash
cd fastapi-ai-backend
```

### 2. Create Virtual Environment
```bash
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
```

### 3. Install Dependencies
```bash
pip install -r requirements.txt
```

### 4. Configure Environment
```bash
cp .env.example .env
# Edit .env with your settings
```

### 5. Run the Application
```bash
# Method 1: Using uvicorn directly
uvicorn main:app --reload

# Method 2: Using Python
python main.py
```

## API Documentation

Once running, visit:
- **Swagger UI**: http://localhost:8000/docs
- **ReDoc**: http://localhost:8000/redoc
- **OpenAPI JSON**: http://localhost:8000/openapi.json

## API Endpoints

### Health
- `GET /api/v1/health` - Health check

### Users
- `POST /api/v1/users` - Create user
- `GET /api/v1/users` - List users
- `GET /api/v1/users/{id}` - Get user
- `PATCH /api/v1/users/{id}` - Update user
- `DELETE /api/v1/users/{id}` - Delete user
- `GET /api/v1/users/stats/count` - User statistics

## Development

### Running Tests
```bash
pytest
```

### Code Style
```bash
# Format code
black app/

# Lint
flake8 app/
```

## Environment Variables

See `.env.example` for all available configuration options.

Key variables:
- `DEBUG`: Enable debug mode
- `ENVIRONMENT`: Environment name (development/production)
- `API_V1_PREFIX`: API version prefix
- `ALLOWED_ORIGINS`: CORS allowed origins

## Contributing

1. Follow the existing code structure
2. Add tests for new features
3. Update documentation
4. Use type hints

## License

MIT

---

🎯 Final Project Structure

Your complete structure should now look like this:

fastapi-ai-backend/
├── app/
│   ├── __init__.py
│   ├── main.py                      ✅ Application factory
│   │
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── api.py               ✅ Router aggregator
│   │       └── endpoints/
│   │           ├── __init__.py
│   │           ├── health.py        ✅ Health endpoints
│   │           └── users.py         ✅ User endpoints
│   │
│   ├── core/
│   │   ├── __init__.py
│   │   └── config.py                ✅ Configuration
│   │
│   ├── models/
│   │   ├── __init__.py
│   │   ├── common.py                ✅ Common models
│   │   └── user.py                  ✅ User models
│   │
│   ├── services/
│   │   ├── __init__.py
│   │   └── user_service.py          ✅ User business logic
│   │
│   ├── schemas/                     (empty for now)
│   │   └── __init__.py
│   │
│   └── utils/                       (empty for now)
│       └── __init__.py
│
├── tests/
│   └── __init__.py
│
├── venv/
├── .env                             ✅ Environment variables
├── .env.example                     ✅ Example env file
├── .gitignore
├── main.py                          ✅ Entry point
├── README.md                        ✅ Documentation
└── requirements.txt

---

🧪 Testing Your New Structure

Method 1: Run the Server

# Make sure you're in project root with venv activated
python main.py

# Or use uvicorn directly
uvicorn main:app --reload

Method 2: Test the Endpoints

Visit: http://localhost:8000/docs

You should see:

• Organized by tags (Health, Users)
• All your endpoints under /api/v1/
• Beautiful documentation with your descriptions

Method 3: Use curl

# Health check
curl http://localhost:8000/api/v1/health

# Create user
curl -X POST http://localhost:8000/api/v1/users \
  -H "Content-Type: application/json" \
  -d '{
    "username": "testuser",
    "email": "test@example.com",
    "password": "password123"
  }'

# Get users
curl http://localhost:8000/api/v1/users

# Get user stats
curl http://localhost:8000/api/v1/users/stats/count

---

📊 Understanding the Architecture

Request Flow Diagram

Client Request
    ↓
main.py (Entry Point)
    ↓
app/main.py (FastAPI App)
    ↓
app/api/v1/api.py (Router Aggregator)
    ↓
app/api/v1/endpoints/users.py (Endpoint)
    ↓
app/services/user_service.py (Business Logic)
    ↓
app/models/user.py (Data Validation)
    ↓
Response to Client

Separation of Concerns

Layer	Responsibility	Files
Entry Point	Start application	main.py
App Factory	Create & configure app	app/main.py
Configuration	Manage settings	app/core/config.py
API Routing	Route requests	app/api/v1/
Endpoints	Handle HTTP	app/api/v1/endpoints/
Services	Business logic	app/services/
Models	Data validation	app/models/

---

🔧 Common Issues & Solutions

Issue 1: “ModuleNotFoundError: No module named ‘app’”

Solution: Make sure you’re running from project root:

# Wrong
cd app/
python main.py  ❌

# Right
cd fastapi-ai-backend/  # Project root
python main.py  ✅

Issue 2: Environment variables not loading

Solution: Check .env file location:

# .env must be in project root
fastapi-ai-backend/
├── .env          ✅ Here!
└── app/
    └── .env      ❌ Not here

Issue 3: CORS errors in browser

Solution: Update ALLOWED_ORIGINS in .env:

ALLOWED_ORIGINS=http://localhost:3000,http://localhost:8000

---

📚 Key Concepts Summary

1. Project Structure

Single Responsibility: Each file/folder has one purpose
Separation of Concerns: API, Logic, Data are separate
Modular Design: Easy to add/remove features

2. Configuration Management

# Load from .env automatically
settings = get_settings()

# Use throughout app
@app.get("/")
def root(settings: Settings = Depends(get_settings)):
    return {"name": settings.APP_NAME}

3. Dependency Injection

# FastAPI calls this function for you
def get_user_service() -> UserService:
    return user_service

# Automatically injected
@app.get("/users")
def get_users(service: UserService = Depends(get_user_service)):
    return service.get_all_users()

4. Router Organization

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

# Add routes
@router.get("")  # /users
@router.get("/{id}")  # /users/{id}

# Include in app
app.include_router(router, prefix="/api/v1")

---

🎓 Homework / Practice

Exercise 1: Add a Product Feature

Create a complete product feature:

1. app/models/product.py – Product models
2. app/services/product_service.py – Product logic
3. app/api/v1/endpoints/products.py – Product endpoints
4. Add to app/api/v1/api.py

Exercise 2: Environment-Specific Settings

Create different settings for dev/staging/prod:

• .env.development
• .env.staging
• .env.production

Load based on ENVIRONMENT variable.

Exercise 3: Custom Middleware

Add a middleware to log all requests:

• Request method, path, timestamp
• Response status code, duration

---

🚀 What’s Next?

In Blog Post 4, we’ll cover:

• Pydantic Deep Dive: Advanced validation, custom validators
• Request/Response lifecycle: Understanding FastAPI internals
• Error handling: Custom exception handlers, standard error responses
• Data validation: Field validation, custom validators, nested models

This will make our API more robust and production-ready!

---

Congratulations! 🎉 You’ve built a professional, scalable FastAPI application structure. Your code is now:

✅ Organized and maintainable
✅ Easy to test
✅ Ready for team collaboration
✅ Scalable for future features
✅ Production-ready architecture