事务 A: UPDATE books SET status = 'BORROWED' WHERE id = 1;  COMMIT;
-- 此时进程崩溃或网络中断
事务 B: INSERT INTO borrows (...);  COMMIT;  -- 永远不会执行了

# app/models/borrow.py
from datetime import datetime

from sqlalchemy import DateTime, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column

from app.db import Base


class Borrow(Base):
    __tablename__ = "borrows"

    id: Mapped[int] = mapped_column(primary_key=True)
    book_id: Mapped[int] = mapped_column(
        ForeignKey("books.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
    )
    user_id: Mapped[int] = mapped_column(nullable=False, index=True)
    borrowed_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False
    )
    due_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False
    )
    returned_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True)
    )

# app/models/__init__.py
from app.models.author import Author
from app.models.book import Book
from app.models.borrow import Borrow

__all__ = ["Author", "Book", "Borrow"]

# app/dao/borrow_dao.py
from datetime import datetime, timedelta, UTC

from sqlalchemy.ext.asyncio import AsyncSession

from app.models.book import Book, BookStatus
from app.models.borrow import Borrow
from app.errors import BookUnavailableError, NotFoundError


async def borrow_book(
    db: AsyncSession,
    *,
    book_id: int,
    user_id: int,
    loan_days: int = 14,
) -> Borrow:
    """借书事务：校验图书可借、改 Book.status、新增 Borrow 行，单次 commit。"""
    book = await db.get(Book, book_id)
    if book is None or book.deleted_at is not None:
        raise NotFoundError(book_id)
    if book.status != BookStatus.AVAILABLE:
        raise BookUnavailableError(book_id)

    now = datetime.now(UTC)
    borrow = Borrow(
        book_id=book_id,
        user_id=user_id,
        borrowed_at=now,
        due_at=now + timedelta(days=loan_days),
    )
    book.status = BookStatus.BORROWED
    db.add(borrow)
    await db.commit()
    await db.refresh(borrow)
    return borrow


async def return_book(
    db: AsyncSession,
    *,
    borrow_id: int,
) -> Borrow:
    """还书事务：填充 Borrow.returned_at 与复位 Book.status，单次 commit。"""
    borrow = await db.get(Borrow, borrow_id)
    if borrow is None or borrow.returned_at is not None:
        raise NotFoundError(borrow_id)
    book = await db.get(Book, borrow.book_id)
    if book is not None:
        book.status = BookStatus.AVAILABLE
    borrow.returned_at = datetime.now(UTC)
    await db.commit()
    await db.refresh(borrow)
    return borrow

# app/errors.py
class BookUnavailableError(Exception):
    """图书当前不可借阅（状态非 AVAILABLE 或并发被占用）。"""

    def __init__(self, book_id: int) -> None:
        super().__init__(f"book {book_id} is not available")
        self.book_id = book_id

# app/api/books.py
from typing import Annotated

from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.dao import borrow_dao
from app.db import get_async_db
from app.errors import NotFoundError
from app.models.book import Book
from app.models.borrow import Borrow

router = APIRouter(prefix="/books", tags=["books"])
DBDep = Annotated[AsyncSession, Depends(get_async_db)]


@router.post("/{book_id}/borrow", status_code=201)
async def borrow_book(
    book_id: int,
    user_id: Annotated[int, Query(ge=1)],
    db: DBDep,
) -> Borrow:
    """借书，失败返回 404/409。"""
    return await borrow_dao.borrow_book(db, book_id=book_id, user_id=user_id)


@router.post("/{book_id}/return", status_code=200)
async def return_book(book_id: int, db: DBDep) -> Borrow:
    """还书：定位该书当前未归还的 Borrow 记录，调用 DAO 复位状态。"""
    result = await db.execute(
        select(Borrow)
        .where(Borrow.book_id == book_id, Borrow.returned_at.is_(None))
        .order_by(Borrow.id.desc())
    )
    record = result.scalar_one_or_none()
    if record is None:
        raise NotFoundError(book_id)
    return await borrow_dao.return_book(db, borrow_id=record.id)