SQLModel: основы

SQLModel — библиотека для работы с БД в FastAPI, объединяющая SQLAlchemy и Pydantic.

Установка

pip install sqlmodel

Модели

from sqlmodel import SQLModel, Field, Relationship
from typing import Optional, List
from datetime import datetime

class Project(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(max_length=200)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    tasks: List["Task"] = Relationship(back_populates="project")

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(max_length=200, index=True)
    status: str = Field(default="todo")
    priority: int = Field(default=1)
    project_id: Optional[int] = Field(default=None, foreign_key="project.id")

    project: Optional[Project] = Relationship(back_populates="tasks")

Подключение к БД

from sqlmodel import create_engine, Session, SQLModel

DATABASE_URL = "sqlite:///./db.sqlite3"
# Для PostgreSQL:
# DATABASE_URL = "postgresql://user:pass@localhost/mydb"

engine = create_engine(DATABASE_URL, echo=True)  # echo=True логирует SQL

def create_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

CRUD операции

from sqlmodel import select

def create_task(session: Session, title: str, project_id: int) -> Task:
    task = Task(title=title, project_id=project_id)
    session.add(task)
    session.commit()
    session.refresh(task)  # обновить из БД (получить id)
    return task

def get_task(session: Session, task_id: int) -> Task | None:
    return session.get(Task, task_id)

def list_tasks(session: Session, status: str | None = None) -> list[Task]:
    query = select(Task)
    if status:
        query = query.where(Task.status == status)
    return session.exec(query).all()

def update_task(session: Session, task: Task, **kwargs) -> Task:
    for key, value in kwargs.items():
        setattr(task, key, value)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task

def delete_task(session: Session, task: Task):
    session.delete(task)
    session.commit()

Схемы без table=True

class TaskCreate(SQLModel):      # только для API, без table=True
    title: str
    priority: int = 1

class TaskResponse(SQLModel):
    id: int
    title: str
    status: str
    created_at: datetime
    model_config = {"from_attributes": True}