Software Development

The Unfolding Power of FastAPI: A Deep Dive into Project Initialization and Architecture

The inherent flexibility of FastAPI is its most compelling attribute, offering developers a vast canvas upon which to construct robust and scalable web applications. This freedom, while empowering, also presents a unique challenge, akin to assembling a complex entity from disparate parts – a Frankenstein’s monster of code. This article delves into the initial stages of building a FastAPI project, focusing on establishing a solid architectural foundation and integrating essential tools for efficient development.

The Genesis of a FastAPI Project: Embracing the "Frankenstein" Approach

The term "Frankenstein" aptly describes the FastAPI development experience. Unlike frameworks that impose rigid structures, FastAPI allows developers to select and integrate a variety of libraries and architectural patterns. This modularity grants immense power to tailor projects to specific needs, but it also means that the resulting application’s elegance and efficiency depend heavily on the choices made during its creation. A well-architected project can be a marvel of engineering, while a poorly planned one can become unwieldy and difficult to maintain.

This guide aims to provide a best-practice approach to initializing a FastAPI project, drawing upon current documentation and experienced development patterns. It is crucial to note that the landscape of software development evolves rapidly. Therefore, always cross-reference the information presented here with the latest official documentation, especially if this article is more than six months old.

Setting the Stage: Environment Management with UV

Efficient project management begins with a robust and fast environment manager. While traditional Python virtual environments using venv are well-understood and widely used, the introduction of uv by Astral signifies a significant leap forward in speed and efficiency for package installation and environment management.

To initiate a project, navigate to your desired directory and create a new project folder, here named "franky." Within this directory, the uv init command bootstraps a new project, creating essential files such as main.py, README.md, and pyproject.toml. This command also implicitly initializes a Git repository, which can be retained or removed based on project requirements.

# Create the project directory
mkdir franky
cd franky

# Initialize the project with uv
uv init

Following the initialization, the next step involves installing the core dependencies. This curated list of libraries provides a strong foundation for a modern FastAPI application:

  • fastapi[standard]: The core FastAPI framework, including common optional dependencies for enhanced functionality.
  • pydantic-settings: For managing application settings and configurations, particularly from environment variables.
  • python-dotenv: To load environment variables from a .env file, facilitating local development configuration.
  • sqlmodel: A powerful library that combines Pydantic and SQLAlchemy, simplifying the definition of data models and database interactions.
  • uvicorn: A high-performance ASGI server essential for running FastAPI applications.
  • alembic: A database migration tool that allows for incremental changes to the database schema over time.
  • httpx: An asynchronous HTTP client, useful for making requests to other services or for testing.
  • pytest: The de facto standard for Python testing, enabling the creation of comprehensive test suites.
  • pytest-asyncio: A plugin that enhances pytest‘s ability to test asynchronous code.
  • greenlet: A low-level C extension that enables cooperative multitasking, often used by libraries like SQLAlchemy for managing asynchronous operations.
# Add core dependencies using uv
uv add 'fastapi[standard]'
uv add pydantic-settings
uv add python-dotenv
uv add sqlmodel
uv add uvicorn
uv add alembic
uv add httpx
uv add pytest
uv add pytest_asyncio
uv add greenlet

The pyproject.toml file will now reflect these dependencies, providing a centralized record of the project’s requirements.

# franky/pyproject.toml
[project]
name = "franky"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
    "aiosqlite>=0.22.1",
    "alembic>=1.18.4",
    "fastapi[standard]>=0.136.0",
    "greenlet>=3.4.0",
    "httpx>=0.28.1",
    "pydantic-settings>=2.13.1",
    "pytest>=9.0.3",
    "pytest-asyncio>=1.3.0",
    "python-dotenv>=1.2.2",
    "sqlmodel>=0.0.38",
    "uvicorn>=0.44.0",
]

To verify the setup, a basic "Hello World" endpoint can be added to main.py:

# franky/main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return "message": "Hello World"

This application can be launched using uv‘s run command:

uv run fastapi dev

This command will start the development server, typically accessible at http://127.0.0.1:8000.

Core Project Configuration: Settings and Logging

Establishing a clear configuration management system is paramount for any application. This involves centralizing settings and defining how they are loaded. The pydantic-settings library, combined with python-dotenv, offers an elegant solution.

A Config class, inheriting from BaseSettings, can be defined to manage application parameters, including database connection details and debug modes.

# franky/src/core/config.py
import os
from dotenv import load_dotenv
from pydantic_settings import BaseSettings

load_dotenv()

class Config(BaseSettings):
    app_name: str = "Franky"
    debug: bool = True
    db_name: str = os.getenv("DB_NAME")

    @property
    def db_url(self):
        # Using sqlite+aiosqlite for asynchronous SQLite operations
        return f"sqlite+aiosqlite:///./self.db_name"

config = Config()

A .env file in the project root will store the database name:

# franky/.env
DB_NAME=db.sqlite3

The choice of sqlite+aiosqlite highlights the commitment to asynchronous operations, crucial for maintaining high performance in web applications.

Logging is another critical aspect of application monitoring and debugging. A simple setup function can define the logging level and format.

# franky/src/core/logging.py
import logging

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s [%(name)s] %(message)s"
    )

These configurations are then integrated into the main FastAPI application instance.

# franky/main.py (snippet)
from fastapi import FastAPI
from src.core.logging import setup_logging
from src.core.config import config

setup_logging()

app = FastAPI(title=config.app_name)
# ... rest of the main app setup

The updated logging format provides more context, including timestamps and module names, which is invaluable for troubleshooting.

Unified Response and Exception Handling

A consistent API response structure enhances usability and predictability for consumers. The goal is to present a unified schema for both successful operations and errors.

Successful Response Schema:


  "success": true,
  "message": "Operation successful",
  "data": 
    // ... actual response data
  

Error Response Schema:


  "success": false,
  "message": "An error occurred",
  "data": null

This is achieved through a combination of a generic response model and a middleware that intercepts and wraps responses.

# franky/src/core/models.py
from typing import Generic, TypeVar
from pydantic import BaseModel

T = TypeVar("T")

class IResponse(BaseModel, Generic[T]):
    success: bool = True
    message: str = "Operation successful"
    data: T | None = None

The UnifiedResponseMiddleware inspects outgoing responses. If a response is JSON and indicates success (status code less than 400), it wraps the data within the IResponse schema. It also intelligently skips modifying responses intended for API documentation endpoints like /docs and /redoc.

# franky/src/core/middlewares.py
import json
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.concurrency import iterate_in_threadpool

from src.core.models import IResponse

class UnifiedResponseMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Skip wrapping for docs
        if request.url.path in ["/openapi.json", "/docs", "/redoc"]:
            return await call_next(request)

        response = await call_next(request)

        # Only wrap JSON
        content_type = response.headers.get("content-type", "")
        if response.status_code < 400 and "application/json" in content_type:
            body = b"".join([section async for section in response.body_iterator])

            if not body:
                return response

            try:
                data = json.loads(body.decode("utf-8"))

                # Check if already wrapped
                if isinstance(data, dict) and "success" in data:
                    response.body_iterator = iterate_in_threadpool(iter([body]))
                    return response

                # Wrap
                wrapped_data = IResponse(data=data).model_dump_json()

                # Clean Headers
                headers = dict(response.headers)
                headers.pop("Content-Length", None)
                headers.pop("content-length", None)

                return Response(
                    content=wrapped_data,
                    status_code=response.status_code,
                    headers=headers,
                    media_type="application/json"
                )
            except (json.JSONDecodeError, UnicodeDecodeError):
                response.body_iterator = iterate_in_threadpool(iter([body]))
                return response

        return response

To handle errors uniformly, custom exception handlers are implemented. These handlers catch various exception types, including StarletteHTTPException, RequestValidationError, and generic Exception, and format them into the predefined error response schema.

# franky/src/core/exceptions.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

async def common_exception_handler(request: Request, exc: Exception):
    status_code = 400
    message = str(exc)

    # Check if it's a Starlette/FastAPI HTTPException
    if isinstance(exc, StarletteHTTPException):
        status_code = exc.status_code
        message = exc.detail

    return JSONResponse(
        status_code=status_code,
        content=
            "success": False,
            "message": message,
            "data": None
        
    )

def setup_exception_handlers(app: FastAPI):
    app.add_exception_handler(StarletteHTTPException, common_exception_handler)
    app.add_exception_handler(RequestValidationError, common_exception_handler)
    app.add_exception_handler(Exception, common_exception_handler)

These handlers and middleware are then registered with the FastAPI application.

# franky/main.py (updated snippet)
from fastapi import FastAPI
from src.core.logging import setup_logging
from src.core.config import config
from src.core.exceptions import setup_exception_handlers
from src.core.middlewares import UnifiedResponseMiddleware

setup_logging()

app = FastAPI(title=config.app_name)

setup_exception_handlers(app)
app.add_middleware(UnifiedResponseMiddleware)

@app.get("/")
async def root():
    return "message": "Hello World"

@app.get("/error")
async def trigger_error():
    raise Exception("Serious error occurred")

Testing the /error endpoint now reveals the unified error response structure.

Modular Application Structure

A well-defined modular structure is key to maintaining large and complex applications. A common pattern involves organizing code into distinct modules, each responsible for a specific domain or feature.

--src
  --module1
    --__init__.py
    --dependencies.py
    --models.py
    --router.py
    --service.py
  ...

This approach promotes separation of concerns, making the codebase easier to navigate, test, and scale.

The Appointments Module: A Practical Example

To illustrate this modularity, an appointments module is introduced. This module encompasses models, services, dependencies, and routers related to appointment management.

Models: SQLModel is used to define the data structures for appointments, including their status, base attributes, and specific schemas for creation, update, and reading.

# franky/src/appointments/models.py
from datetime import datetime, UTC
from enum import Enum
from typing import Optional

from sqlmodel import SQLModel, Field

class AppointmentStatus(str, Enum):
    scheduled = "scheduled"
    completed = "completed"
    cancelled = "cancelled"

class AppointmentBase(SQLModel):
     str = Field(min_length=1, max_length=255)
    description: Optional[str] = Field(default=None, max_length=1000)
    start_time: datetime
    end_time: datetime
    location: Optional[str] = Field(default=None, max_length=500)
    status: AppointmentStatus = AppointmentStatus.scheduled

class Appointment(AppointmentBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

class AppointmentCreate(AppointmentBase):
    pass

class AppointmentUpdate(SQLModel):
     Optional[str] = Field(default=None, min_length=1, max_length=255)
    description: Optional[str] = Field(default=None, max_length=1000)
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None
    location: Optional[str] = Field(default=None, max_length=500)
    status: Optional[AppointmentStatus] = None

class AppointmentRead(AppointmentBase):
    id: int
    created_at: datetime
    updated_at: datetime

Service: The AppointmentService class encapsulates the business logic for interacting with the database, providing CRUD (Create, Read, Update, Delete) operations. It leverages asynchronous SQLAlchemy sessions for efficient data manipulation.

# franky/src/appointments/service.py
from datetime import datetime, UTC
from typing import Optional, Sequence

from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select

from src.appointments.models import Appointment, AppointmentCreate, AppointmentUpdate

class AppointmentService:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def create(self, data: AppointmentCreate) -> Appointment:
        appointment = Appointment.model_validate(data)
        self.session.add(appointment)
        await self.session.commit()
        await self.session.refresh(appointment)
        return appointment

    async def get(self, appointment_id: int) -> Optional[Appointment]:
        return await self.session.get(Appointment, appointment_id)

    async def list(self, offset: int = 0, limit: int = 20) -> Sequence[Appointment]:
        result = await self.session.execute(
            select(Appointment).offset(offset).limit(limit)
        )
        return result.scalars().all()

    async def update(
        self, appointment_id: int, data: AppointmentUpdate
    ) -> Optional[Appointment]:
        appointment = await self.session.get(Appointment, appointment_id)
        if not appointment:
            return None
        updates = data.model_dump(exclude_unset=True)
        for key, value in updates.items():
            setattr(appointment, key, value)
        appointment.updated_at = datetime.now(UTC)
        self.session.add(appointment)
        await self.session.commit()
        await self.session.refresh(appointment)
        return appointment

    async def delete(self, appointment_id: int) -> bool:
        appointment = await self.session.get(Appointment, appointment_id)
        if not appointment:
            return False
        await self.session.delete(appointment)
        await self.session.commit()
        return True

Dependencies: A dependency function (get_appointment_service) is created to inject the AppointmentService into routes, ensuring that each request handler receives a fresh database session.

# franky/src/appointments/dependencies.py
from typing import Annotated

from fastapi import Depends

from src.core.dependencies import SessionDep
from src.appointments.service import AppointmentService

def get_appointment_service(session: SessionDep) -> AppointmentService:
    return AppointmentService(session)

AppointmentServiceDep = Annotated[AppointmentService, Depends(get_appointment_service)]

Router: The APIRouter defines the API endpoints for appointments, including POST for creation, GET for listing and retrieval, PATCH for updates, and DELETE for removal.

# franky/src/appointments/router.py
from fastapi import APIRouter, HTTPException, Query

from src.appointments.dependencies import AppointmentServiceDep
from src.appointments.models import AppointmentCreate, AppointmentRead, AppointmentUpdate

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

@router.post("/", response_model=AppointmentRead, status_code=201)
async def create_appointment(data: AppointmentCreate, service: AppointmentServiceDep):
    return await service.create(data)

@router.get("/", response_model=list[AppointmentRead])
async def list_appointments(
    service: AppointmentServiceDep,
    offset: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
):
    return await service.list(offset=offset, limit=limit)

@router.get("/appointment_id", response_model=AppointmentRead)
async def get_appointment(appointment_id: int, service: AppointmentServiceDep):
    appointment = await service.get(appointment_id)
    if not appointment:
        raise HTTPException(status_code=404, detail="Appointment not found")
    return appointment

@router.patch("/appointment_id", response_model=AppointmentRead)
async def update_appointment(
    appointment_id: int,
    data: AppointmentUpdate,
    service: AppointmentServiceDep,
):
    appointment = await service.update(appointment_id, data)
    if not appointment:
        raise HTTPException(status_code=404, detail="Appointment not found")
    return appointment

@router.delete("/appointment_id", status_code=204)
async def delete_appointment(appointment_id: int, service: AppointmentServiceDep):
    deleted = await service.delete(appointment_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Appointment not found")

Finally, the appointments_router is included in the main FastAPI application.

# franky/main.py (updated snippet)
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlmodel import SQLModel, create_engine # Import SQLModel for metadata
from src.core.logging import setup_logging
from src.core.config import config
from src.core.exceptions import setup_exception_handlers
from src.core.middlewares import UnifiedResponseMiddleware
from src.appointments.router import router as appointments_router
from src.core.dependencies import engine # Import the engine from core dependencies

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        # Ensure tables are created if they don't exist
        await conn.run_sync(SQLModel.metadata.create_all)
    yield

app = FastAPI(title=config.app_name, lifespan=lifespan)

setup_exception_handlers(app)
app.add_middleware(UnifiedResponseMiddleware)

app.include_router(appointments_router)

The lifespan context manager is a temporary measure for initial setup, ensuring database tables are created. This will be superseded by a more robust migration strategy.

Database Migrations with Alembic

For managing database schema evolution, Alembic is the standard tool. Its integration with asynchronous SQLAlchemy and SQLModel requires specific configuration.

Initialize Alembic with the asynchronous template:

uv run alembic init -t async migrations

The generated Alembic scripts need to be adapted to work with SQLModel and asynchronous operations.

migrations/script.py.mako:

"""$message

Revision ID: $up_revision
Revises: $ comma,n
Create Date: $create_date

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
import sqlmodel             # NEW
$imports if imports else ""
...

migrations/env.py: This file is crucial for Alembic to recognize your models and connect to the database.

# franky/migrations/env.py
import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlmodel import SQLModel   # NEW

from alembic import context

from src.appointments.models import Appointment   # NEW

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = config # Use the global config object

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = SQLModel.metadata   # UPDATED

# ... rest of the env.py file

The alembic.ini file must be updated to point to your database URL.

# franky/alembic.ini (snippet)
sqlalchemy.url = sqlite+aiosqlite:///./db.sqlite3   # UPDATED

With the configuration in place, create the initial migration:

uv run alembic revision --autogenerate -m "init"

This command will generate a new migration script in migrations/versions/. Applying this migration will create the necessary database tables.

uv run alembic upgrade head

Once migrations are set up, the SQLModel.metadata.create_all call in main.py can be removed, relying solely on Alembic for schema management.

The final project structure consolidates these components into a well-organized and scalable application.

franky/
├── main.py                 # FastAPI app, lifespan, route registration
├── alembic.ini             # Alembic configuration (DB URL for CLI migrations)
├── pyproject.toml          # Dependencies and project metadata
├── uv.lock                 # Locked dependency versions
└── migrations/             # Alembic migration environment
    ├── env.py              # Async SQLAlchemy / SQLModel metadata for autogenerate
    └── versions/           # Revision scripts
└── src/
    ├── core/               # Shared infrastructure
    │   ├── config.py       # Settings (env / `.env`)
    │   ├── dependencies.py # Async SQLAlchemy engine and session
    │   ├── exceptions.py   # Exception handlers
    │   ├── logging.py
    │   ├── middlewares.py  # Unified JSON response wrapper
    │   └── models.py       # `IResponse`, pagination types
    └── appointments/       # Appointments domain
        ├── dependencies.py # Service DI
        ├── models.py       # SQLModel entities and request/response schemas
        ├── router.py       # HTTP routes under `/appointments`
        └── service.py      # Business logic

This comprehensive setup provides a robust foundation for building sophisticated FastAPI applications, emphasizing flexibility, maintainability, and developer efficiency. Future articles will explore further enhancements, such as user management and advanced authentication mechanisms.

Related Articles

Leave a Reply

Your email address will not be published. Required fields are marked *

Back to top button
Tech Survey Info
Privacy Overview

This website uses cookies so that we can provide you with the best user experience possible. Cookie information is stored in your browser and performs functions such as recognising you when you return to our website and helping our team to understand which sections of the website you find most interesting and useful.