PostgreSQL con FastAPI

  • Python
  • Gabriele Romanato

In questo articolo configuriamo un backend FastAPI con PostgreSQL usando SQLAlchemy 2.x in modalità asincrona, migrazioni Alembic e impostazioni tipizzate con Pydantic v2.

Installazione pacchetti

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install "fastapi[standard]" sqlalchemy[asyncio] asyncpg alembic pydantic-settings python-dotenv

Struttura consigliata del progetto

.
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── database.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   └── routers/
│       └── items.py
├── alembic.ini
├── alembic/  # generato da Alembic
└── .env

Variabili d’ambiente

Crea un file .env nella root:

DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/mydb
ENV=dev

Configurazione con Pydantic Settings

app/config.py

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
  database_url: str
  env: str = "dev"
  model_config = SettingsConfigDict(env_file=".env", env_prefix="", extra="ignore")

settings = Settings()

Engine asincrono e sessione

app/database.py

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from .config import settings

engine = create_async_engine(
  settings.database_url,
  echo=False,
  pool_pre_ping=True,
)

SessionLocal = async_sessionmaker(
  bind=engine,
  class_=AsyncSession,
  expire_on_commit=False,
)

async def get_session() -> AsyncGenerator[AsyncSession, None]:
async with SessionLocal() as session:
  yield session

Definizione dei modelli

app/models.py

from typing import Annotated
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, text, func

class Base(DeclarativeBase):
  pass

IdPK = Annotated[int, mapped_column(primary_key=True)]
Timestamp = Annotated["datetime",mapped_column(server_default=func.now(), nullable=False)]

class Item(Base):
  tablename = "items"

  id: Mapped[IdPK]
  name: Mapped[str] = mapped_column(String(120), index=True)
  description: Mapped[str | None] = mapped_column(String(500))
  created_at: Mapped["datetime"] = mapped_column(server_default=func.now(), nullable=False)

Schemi con Pydantic

app/schemas.py

from pydantic import BaseModel, Field

class ItemBase(BaseModel):
  name: str = Field(min_length=1, max_length=120)
  description: str | None = Field(default=None, max_length=500)

class ItemCreate(ItemBase):
  pass

class ItemRead(ItemBase):
  id: int
  model_config = {"from_attributes": True}

CRUD asincrono

app/crud.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from . import models, schemas

async def create_item(db: AsyncSession, data: schemas.ItemCreate) -> models.Item:
  item = models.Item(name=data.name, description=data.description)
  db.add(item)
  await db.commit()
  await db.refresh(item)
  return item

async def list_items(db: AsyncSession) -> list[models.Item]:
  res = await db.execute(select(models.Item).order_by(models.Item.id.desc()))
  return list(res.scalars().all())

async def get_item(db: AsyncSession, item_id: int) -> models.Item | None:
  res = await db.execute(select(models.Item).where(models.Item.id == item_id))
  return res.scalar_one_or_none()

Router REST

app/routers/items.py

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from ..schemas import ItemCreate, ItemRead
from ..crud import create_item, list_items, get_item
from ..database import get_session

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

@router.post("", response_model=ItemRead, status_code=status.HTTP_201_CREATED)
async def create(data: ItemCreate, db: AsyncSession = Depends(get_session)):
  return await create_item(db, data)

@router.get("", response_model=list[ItemRead])
async def read_all(db: AsyncSession = Depends(get_session)):
  return await list_items(db)

@router.get("/{item_id}", response_model=ItemRead)
async def read_one(item_id: int, db: AsyncSession = Depends(get_session)):
  item = await get_item(db, item_id)
  if not item:
    raise HTTPException(status_code=404, detail="Item non trovato")
  return item

Avvio applicazione e ciclo di vita

app/main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from .database import engine
from .models import Base
from .routers import items

@asynccontextmanager
async def lifespan(app: FastAPI):
  # Esegue le create solo in dev; in prod usa Alembic
  async with engine.begin() as conn:
  await conn.run_sync(Base.metadata.create_all)
  yield
  await engine.dispose()

app = FastAPI(lifespan=lifespan)
app.include_router(items.router)

Migrazioni con Alembic

Inizializza Alembic:

alembic init alembic

Configura la URL nel alembic.ini (puoi anche leggere da .env nel file env.py).

alembic/env.py (estratto per async e SQLAlchemy):

from alembic import context
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from app.models import Base
import os

config = context.config
config.set_main_option("sqlalchemy.url", os.getenv("DATABASE_URL"))

target_metadata = Base.metadata

def run_migrations_offline():
  url = config.get_main_option("sqlalchemy.url")
  context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
  with context.begin_transaction():
    context.run_migrations()

async def run_migrations_online():
  connectable = async_engine_from_config(
    config.get_section(config.config_ini_section, {}),
    prefix="sqlalchemy.",
    poolclass=pool.NullPool,
  )
async with connectable.connect() as connection:
  await connection.run_sync(do_run_migrations)

async def do_run_migrations(connection: Connection):
  context.configure(connection=connection, target_metadata=target_metadata)
  with context.begin_transaction():
    context.run_migrations()

  if context.is_offline_mode():
    run_migrations_offline()
  else:
  import asyncio
  asyncio.run(run_migrations_online())

Crea e applica una migration:

alembic revision -m "create items" --autogenerate
alembic upgrade head

Esecuzione locale

uvicorn app.main:app --reload

Visita la documentazione interattiva su /docs (Swagger UI) o /redoc.

Test rapido con HTTPie o curl

# Crea un item
http POST :8000/items name="Notebook" description="Formato A5"

Connessione via Docker (opzionale)

Esempio rapido di docker-compose.yml per PostgreSQL:

services:
  db:
    image: postgres
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: mydb
    ports:
      - "127.0.0.1:5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
volumes:
  pgdata: {}

Note su performance e produzione

  • Usa pool_pre_ping=True per gestire connessioni inattive.
  • Preferisci Alembic per lo schema in ambienti reali; evita create_all in produzione.
  • Imposta limiti di pool e timeout secondo il carico e l’orchestratore (ad es. Gunicorn/Uvicorn workers).
  • Con Pydantic v2 usa model_validate / from_attributes=True per convertire ORM → schema.

Con questa base puoi scalare in sicurezza: aggiungi nuovi modelli, versiona lo schema con Alembic, proteggi gli endpoint con OAuth2/JWT e osserva le prestazioni con strumenti APM. FastAPI e PostgreSQL, insieme, offrono velocità di sviluppo e solidità in produzione.

Torna su