SQLAlchemy and database sessions

# pip install sqlalchemy "psycopg[binary]" alembic
# db.py
from collections.abc import Iterator
from typing import Annotated
from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

engine = create_engine("postgresql+psycopg://user:pass@localhost/shop",
                       pool_pre_ping=True)
SessionLocal = sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

def get_db() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session

Db = Annotated[Session, Depends(get_db)]

# models.py
from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from db import Base

class Author(Base):
    __tablename__ = "author"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120))
    books: Mapped[list["Book"]] = relationship(back_populates="author")

class Book(Base):
    __tablename__ = "book"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    author_id: Mapped[int] = mapped_column(ForeignKey("author.id"))
    author: Mapped[Author] = relationship(back_populates="books")

# schemas.py
from pydantic import BaseModel, ConfigDict, Field

class BookCreate(BaseModel):
    title: str = Field(min_length=1)
    author_id: int

class AuthorOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str

class BookOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    title: str
    author: AuthorOut

# main.py (imports db.Db, models.Book, schemas.*)
from fastapi import FastAPI, HTTPException
from sqlalchemy import select
from db import Db
from models import Author, Book
from schemas import BookCreate, BookOut

app = FastAPI()

@app.post("/books", response_model=BookOut, status_code=201)
def create_book(data: BookCreate, db: Db):
    author = db.get(Author, data.author_id)
    if author is None:
        raise HTTPException(404, "Author not found")
    book = Book(title=data.title, author=author)
    db.add(book)
    db.commit()
    db.refresh(book)
    return book

from fastapi import FastAPI, HTTPException, Query
from sqlalchemy import select
from sqlalchemy.orm import selectinload, joinedload
from db import Db
from models import Author, Book
from schemas import BookOut

app = FastAPI()

@app.get("/books", response_model=list[BookOut])
def list_books(db: Db,
               limit: int = Query(default=20, ge=1, le=100),
               offset: int = Query(default=0, ge=0)):
    stmt = (select(Book)
            .options(joinedload(Book.author))
            .order_by(Book.id)
            .limit(limit).offset(offset))
    return db.scalars(stmt).all()

@app.get("/authors/{author_id}")
def author_with_books(author_id: int, db: Db):
    stmt = (select(Author)
            .options(selectinload(Author.books))
            .where(Author.id == author_id))
    author = db.scalar(stmt)
    if author is None:
        raise HTTPException(404, "Author not found")
    return {"id": author.id,
            "titles": [book.title for book in author.books]}

from sqlalchemy import select

# Fixed whitelist for sorting; never accept arbitrary SQL text.
columns = {"title": Book.title, "id": Book.id}
column = columns.get(sort, Book.id)
stmt = select(Book).where(Book.title.ilike(f"%{q}%"))
stmt = stmt.order_by(column, Book.id).limit(51)
items = db.scalars(stmt).all()

# For deep pages, use keyset pagination:
stmt = (select(Book)
        .where(Book.id > last_seen_id)
        .order_by(Book.id).limit(21))
# Return at most 20 items and use item 21 to detect more.
# Encode cursors so clients do not depend on internal IDs.

from sqlalchemy import Column, ForeignKey, Integer, Table
from sqlalchemy.orm import Mapped, mapped_column, relationship
from db import Base

book_tag = Table(
    "book_tag", Base.metadata,
    Column("book_id", ForeignKey("book.id"), primary_key=True),
    Column("tag_id", ForeignKey("tag.id"), primary_key=True),
)

class Tag(Base):
    __tablename__ = "tag"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    books: Mapped[list["Book"]] = relationship(
        secondary=book_tag, back_populates="tags"
    )

# Add this field to the earlier Book class:
# tags: Mapped[list[Tag]] = relationship(
#     secondary=book_tag, back_populates="books")

# db.add(Book(title="Dune", author=author, tags=[Tag(name="sf")]))

from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from db import Db
from models import Book
from pydantic import BaseModel, Field

class BookPatch(BaseModel):
    title: str | None = Field(default=None, min_length=1)

@app.patch("/books/{book_id}")
def patch_book(book_id: int, patch: BookPatch, db: Db):
    book = db.get(Book, book_id)
    if book is None:
        raise HTTPException(404, "Book not found")
    for field, value in patch.model_dump(exclude_unset=True).items():
        if field == "title" and value is None:
            raise HTTPException(422, "Title cannot be null")
        setattr(book, field, value)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, "Conflict")
    db.refresh(book)
    return {"id": book.id, "title": book.title}

from sqlalchemy import select

with SessionLocal.begin() as db:
    order = db.scalar(select(Order).where(Order.id == order_id)
                      .with_for_update())
    if order is None:
        raise HTTPException(404, "Order not found")
    order.paid = True
    db.add(Payment(order_id=order_id, amount_cents=price))
# Commits on normal exit, rolls back on exception.
# Lock semantics and isolation vary across databases.
# Do not do slow network calls while holding a row lock.

# pip install sqlalchemy asyncpg
from collections.abc import AsyncIterator
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import (
    AsyncSession, async_sessionmaker, create_async_engine,
)
from sqlalchemy.orm import selectinload
from models import Author, Book

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/shop",
    pool_pre_ping=True,
)
SessionMaker = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncIterator[AsyncSession]:
    async with SessionMaker() as session:
        yield session

Db = Annotated[AsyncSession, Depends(get_db)]
app = FastAPI()

@app.get("/authors/{author_id}")
async def read_author(author_id: int, db: Db):
    stmt = (select(Author).options(selectinload(Author.books))
            .where(Author.id == author_id))
    author = await db.scalar(stmt)
    if author is None:
        raise HTTPException(404, "Author not found")
    return {"id": author.id,
            "books": [b.title for b in author.books]}

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession

async def add_book(session: AsyncSession,
                   author_id: int, title: str) -> Book:
    async with session.begin():
        author = await session.get(Author, author_id)
        if author is None:
            raise HTTPException(404, "Author not found")
        book = Book(title=title, author_id=author_id)
        session.add(book)
        await session.flush()       # obtain book.id, not yet committed
    return book  # committed after the block

# Each concurrent task must create its own AsyncSession.
# Lazy relationship access can fail under async: eager load it.