1
from sqlalchemy.ext.asyncio import create_async_engine
2

3
ASYNC_DATABASE_URL = "mysql+aiomysql://root:password@localhost:3306/mydb?charset=utf8mb4"
4

5
async_engine = create_async_engine(
6
    ASYNC_DATABASE_URL,
7
    echo=True,          # 打印 SQL 语句（调试用，生产环境关闭）
8
    pool_size=10,       # 连接池大小
9
    max_overflow=20,    # 超出连接池后最多可创建的连接数
10
)

1
mysql+aiomysql://用户名:密码@主机:端口/数据库名?charset=utf8mb4

1
from sqlalchemy.ext.asyncio import AsyncSession
2
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
3
from sqlalchemy import DateTime, String, Float, func
4
from datetime import datetime
5

6
# 基类：所有模型都继承它
7
class Base(DeclarativeBase):
8
    create_time: Mapped[datetime] = mapped_column(
9
        DateTime,
10
        default=func.now,           # Python 层面：插入时调用
11
        server_default=func.now(),  # 数据库层面：DDL 默认值
12
        comment="创建时间"
13
    )
14
    update_time: Mapped[datetime] = mapped_column(
15
        DateTime,
16
        default=func.now,
17
        onupdate=func.now(),        # 更新记录时自动设置
18
        server_default=func.now(),
19
        comment="更新时间"
20
    )
21

22
# 模型类：对应数据库中的一张表
23
class Book(Base):
24
    __tablename__ = "book"
25

26
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍ID")
27
    bookname: Mapped[str] = mapped_column(String(255), comment="书名")
28
    author: Mapped[str] = mapped_column(String(255), comment="作者")
29
    price: Mapped[float] = mapped_column(Float, comment="价格")
30
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

1
async def create_table():
2
    async with async_engine.begin() as conn:
3
        await conn.run_sync(Base.metadata.create_all)

1
from contextlib import asynccontextmanager
2
from fastapi import FastAPI
3

4
@asynccontextmanager
5
async def lifespan(app):
6
    print("应用启动...")
7
    await create_table()            # 启动时自动建表
8
    yield                           # 应用运行中
9
    print("应用关闭...")
10
    await async_engine.dispose()    # 关闭时释放连接池
11

12
app = FastAPI(lifespan=lifespan)

1
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
2

3
AsyncSessionLocal = async_sessionmaker(
4
    bind=async_engine,          # 绑定引擎
5
    class_=AsyncSession,        # 使用异步会话
6
    expire_on_commit=False      # 提交后对象不过期（推荐）
7
)

1
async def get_database():
2
    async with AsyncSessionLocal() as session:
3
        try:
4
            yield session           # 提供给路由
5
            await session.commit()  # 无异常则提交
6
        except Exception:
7
            await session.rollback()  # 有异常则回滚
8
        finally:
9
            await session.close()     # 关闭会话

1
from fastapi import Depends
2
from sqlalchemy import select
3

4
@app.get("/books")
5
async def get_books(db: AsyncSession = Depends(get_database)):
6
    result = await db.execute(select(Book))
7
    return result.scalars().all()

1
from pydantic import BaseModel
2

3
class BookCreate(BaseModel):
4
    bookname: str
5
    author: str
6
    price: float
7
    publisher: str
8

9
@app.post("/books")
10
async def create_book(data: BookCreate, db: AsyncSession = Depends(get_database)):
11
    book = Book(**data.model_dump())  # Pydantic 模型转 ORM 对象
12
    db.add(book)
13
    await db.commit()
14
    return book

1
@app.put("/books/{id}")
2
async def update_book(id: int, data: BookCreate, db: AsyncSession = Depends(get_database)):
3
    book = await db.get(Book, id)       # ① 查询
4
    if book is None:
5
        raise HTTPException(status_code=404, detail="书籍不存在")
6

7
    book.bookname = data.bookname       # ② 修改属性
8
    book.author = data.author
9
    book.price = data.price
10
    book.publisher = data.publisher
11

12
    await db.commit()                   # ③ 提交
13
    return book

1
@app.delete("/books/{id}")
2
async def delete_book(id: int, db: AsyncSession = Depends(get_database)):
3
    book = await db.get(Book, id)       # ① 查询
4
    if book is None:
5
        raise HTTPException(status_code=404, detail="书籍不存在")
6

7
    await db.delete(book)               # ② 标记删除
8
    await db.commit()                   # ③ 提交
9
    return {"msg": "删除成功"}