FastAPI + PostgreSQL: পূর্ণ CRUD

docker run -d --name notes-db -e POSTGRES_USER=app -e POSTGRES_PASSWORD=secret \
  -e POSTGRES_DB=notes -p 5432:5432 postgres:17

uv add "fastapi[standard]" "sqlalchemy[asyncio]" asyncpg alembic pydantic-settings

# .env
DATABASE_URL=postgresql+asyncpg://app:secret@localhost:5432/notes
JWT_SECRET=dev-only-change-me

# models/base.py
from datetime import datetime
from sqlalchemy import func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(server_default=func.now(), onupdate=func.now())

# models/user.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin


class User(TimestampMixin, Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    name: Mapped[str] = mapped_column(String(100))
    password_hash: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)

    notes: Mapped[list["Note"]] = relationship(back_populates="owner")

# models/note.py
from datetime import datetime
from sqlalchemy import ForeignKey, Index, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base, TimestampMixin


class Note(TimestampMixin, Base):
    __tablename__ = "notes"
    __table_args__ = (
        Index("ix_notes_owner_updated", "owner_id", "updated_at"),   # list query এর জন্য
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"))
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text, default="")
    is_pinned: Mapped[bool] = mapped_column(default=False)
    image_key: Mapped[str | None] = mapped_column(String(300))
    deleted_at: Mapped[datetime | None]

    owner: Mapped["User"] = relationship(back_populates="notes")

# models/__init__.py   ← Alembic যেন সব model চেনে
from app.models.base import Base
from app.models.user import User
from app.models.note import Note

__all__ = ["Base", "User", "Note"]

uv run alembic init -t async migrations

from app.core.config import get_settings
from app.models import Base

config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata

uv run alembic revision --autogenerate -m "users and notes"
# migrations/versions/xxxx_users_and_notes.py খুলে পড়ো!
uv run alembic upgrade head

# schemas/note.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field


class NoteCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    content: str = Field(default="", max_length=50_000)
    is_pinned: bool = False


class NoteUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    content: str | None = Field(default=None, max_length=50_000)
    is_pinned: bool | None = None


class NoteOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)    # SQLAlchemy object থেকে পড়তে পারবে

    id: int
    title: str
    content: str
    is_pinned: bool
    created_at: datetime
    updated_at: datetime


class NotePage(BaseModel):
    items: list[NoteOut]
    total: int
    limit: int
    offset: int

# repositories/note_repo.py
from sqlalchemy import Select, select, func, or_


class NoteRepository:
    def __init__(self, db) -> None:
        self.db = db

    def _base(self, user_id: int, search: str | None) -> Select:
        stmt = select(Note).where(Note.owner_id == user_id, Note.deleted_at.is_(None))
        if search:
            pattern = f"%{search}%"
            stmt = stmt.where(or_(Note.title.ilike(pattern), Note.content.ilike(pattern)))
        return stmt

    async def page(self, user_id: int, limit: int, offset: int, search: str | None) -> tuple[list[Note], int]:
        base = self._base(user_id, search)
        total = await self.db.scalar(select(func.count()).select_from(base.subquery()))
        rows = await self.db.scalars(
            base.order_by(Note.is_pinned.desc(), Note.updated_at.desc()).limit(limit).offset(offset)
        )
        return list(rows), total or 0

    async def page(self, user_id: int, limit: int, offset: int, search: str | None) -> NotePage:
        items, total = await self.notes.page(user_id, limit, offset, search)
        return NotePage(items=items, total=total, limit=limit, offset=offset)

@router.get("", response_model=NotePage)
async def list_notes(
    user: CurrentUser,
    svc: ServiceDep,
    page: PageDep,
    search: str | None = Query(None, max_length=100),
):
    return await svc.page(user.id, page.limit, page.offset, search)

from sqlalchemy.exc import IntegrityError

class ConflictError(AppError):
    status_code = 409
    code = "CONFLICT"


async def register(self, data: UserCreate) -> User:
    user = User(email=data.email.lower(), name=data.name, password_hash=hash_password(data.password))
    self.db.add(user)
    try:
        await self.db.commit()
    except IntegrityError as e:
        await self.db.rollback()
        raise ConflictError("এই email দিয়ে আগেই account আছে") from e
    return user