大模型应用开发 10.FastAPI ② ORM

失败 羞辱 恨意 欺骗 欲望 这些黑色生命力才是真正的动力源泉

                                                                                        —— 26.6.1

一、ORM 简介

        ORM（Object-RelationalMapping，对象关系映射）是一种编程技术，用于在面向对象编程语言和关系型数据库之间建立映射。它允许开发者通过操作对象的方式与数据库进行交互，而无需直接编写复杂的SQL语句。

优势：

① 减少重复的 SQL 代码

② 代码更简洁易读

③ 自动处理数据库连接和事务

④ 自动防止 SQL 注入攻击

1.ORM 分类

2.ORM 使用流程

安装 ——> 建库、建表 ——> 操作数据

Ⅰ、安装SQLAlchemy ORM

pip install "sqlalchemy[asyncio]" aiomysql -i https://mirrors.aliyun.com/pypi/simple/ –trusted-host mirrors.aliyun.com

Ⅱ、建表

创建数据库引擎 ——> 定义模型类 ——> 启动应用时建表

① ORM – 创建数据库引擎

使用 create_async_engine 创建异步引擎

create_async_engine()：是 SQLAlchemy（Python 主流 ORM 数据库框架）中创建异步数据库引擎的核心工厂函数，用于生成支持异步 I/O操作的数据库连接引擎，是 Python 异步编程（async/await）操作关系型数据库的入口。

与同步版的 create_engine 不同，create_async_engine 返回的引擎支持 await 语法，需要配合 async/await 使用，适用于 FastAPI、aiohttp、Sanic 等异步 Web 框架。

# 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:954926928lcl@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

② ORM – 定义模型类

1.基类，基础DeclarativeBase（包含通用属性和字段的映射）

2.定义数据库表对应的模型类

DeclarativeBase：是 SQLAlchemy 2.0 版本中，用于定义ORM 模型类（数据库表映射类）的核心基类。 它是声明式 ORM的基础，所有自定义的数据库模型（对应数据库表）都必须继承这个基类。

mapped_column()：SQLAlchemy 2.0 现代 ORM 最核心的字段定义，专门搭配 DeclarativeBase 基类使用。

# 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, onupdate=func.now())

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

③ ORM – 创建数据库表

1.从连接池获取异步连接，开启事务，执行ORM操作

2.FastAPI应用启动时，创建数据库表

async with 异步引擎.begin() as coon：开启一个异步数据库事务连接，自动管理：

• 自动获取连接
• 自动提交事务（成功）
• 自动回滚事务（失败）
• 自动释放连接

async：标记一个函数 / 代码块是 异步函数，可以使用 await。

• 定义异步函数：async def 函数名():
• 异步上下文管理器：async with
• 异步迭代：async for

await：等待一个异步操作完成，并获取结果。

• 只能在 async def 函数里用
• 只能等待 异步对象（异步方法、异步连接、异步查询）
• 不会卡住整个程序（非阻塞）

.run_sync()：把同步代码放到异步环境里安全执行。

• 固定用法：await conn.run_sync(同步函数)

Base.metadata.create_all：根据你写的 ORM 模型类，自动在数据库里创建所有表。

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

④ 完整代码

import uvicorn
from fastapi import FastAPI
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:xxxxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 启动代码
if __name__ == "__main__":
uvicorn.run("demo1_create:app", host="127.0.0.1", port=8000, reload=True)

练习

        需求：使用 SQLAlchemy ORM 创建用户表，包含字段如下：用户 id、用户名、密码、创建时间、更新时间

import uvicorn
from fastapi import FastAPI
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

class Base(DeclarativeBase):
__abstract__ = True
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class User(Base):
__tablename__ = "user"
id: Mapped[int] = mapped_column(primary_key=True, comment="用户id")
userName: Mapped[str] = mapped_column(String(255), comment="用户名")
password: Mapped[str] = mapped_column(String(255), comment="密码")

async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 启动代码
if __name__ == "__main__":
uvicorn.run("demo2_practice1:app", host="127.0.0.1", port=8000, reload=True)

Ⅲ、路由匹配中使用ORM

核心：创建依赖项，使用Depends注入到处理函数

① 创建异步数据库引擎

create_async_engine()：创建异步数据库引擎，负责管理数据库连接池、发送 SQL、执行事务。

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

② 定义模型类

DeclarativeBase：所有 ORM 模型的基类，用于生成数据表、管理字段。

mapped_column()：定义数据表字段（列），替代旧版 Column （）。

func.now()：数据库层面获取当前时间。

datetime.now：Python 层面获取当前时间。

String()：定义字符串类型字段。

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

③ 建表

async_engine.begin()：开启一个异步数据库事务（Transaction）

• 建立数据库连接
• 开启事务
• 自动管理提交 / 回滚
• 用完自动释放连接

.run_sync()：在异步连接中，运行同步函数（同步转异步执行）

• 因为 是同步方法create_all()
• 但我们用的是异步连接
• 所以必须用 包装执行.run_sync()

Base.metadata.create_all：根据 ORM 模型类，自动创建所有数据库表

• 读取所有继承 Base 的模型
• 自动生成 CREATE TABLE 语句
• 表不存在才创建，不会覆盖

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

④ 绑定会话类

async_sessionmaker()：创建异步会话工厂，用于生成数据库会话。

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

⑤ 设置依赖项

AsyncSessionLocal()：生成数据库会话实例，用于执行 SQL。

yield：在依赖项中返回会话，并保留上下文，实现自动提交 / 回滚。

AsyncSessionLocal().commit()：提交事务，保存增删改操作。

AsyncSessionLocal().rollback()：回滚事务，出错时撤销操作。

AsyncSessionLocal().close()：关闭会话，释放数据库连接。

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

⑥ depends注入使用依赖项

Depends()：FastAPI 依赖注入，自动获取数据库会话。

db.execute()：执行 SQL 语句（select / insert / update / delete）。

select()：构建查询语句。

.scalars()：从查询结果中提取 ORM 对象。

.scalars().all()：获取所有查询结果，返回列表。

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
# 查询
result = await db.execute(select(Book))
book = result.scalars().all()
return book

⑦ 完整代码

import uvicorn
from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from sqlalchemy import select

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
# 查询
result = await db.execute(select(Book))
book = result.scalars().all()
return book

if __name__ == "__main__":
uvicorn.run('demo3_routeMatching:app', host="127.0.0.1", port=8000, reload=True)

Ⅳ、操作数据 – 查询

数据库操作：查询 select()、新增 add()、更新 先查再改（重新赋值）、删除 delete()

① 数据库操作 – 查询

核心语句：await db.execute(select(模型类))，返回一个ORM对象

获取所有数据

        scalars().all()

获取单条数据

        scalars().first()

        get(模型类，主键值)

import uvicorn
from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from sqlalchemy import select

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
# 返回ORM对象
result = await db.execute(select(Book))
# 查询1 获取所有
# book = result.scalars().all()

# 查询2 获取第一个
# book = result.scalars().first()

# 查询3 根据主键获取单条数据
book = await db.get(Book, 3)
return book

if __name__ == "__main__":
uvicorn.run('demo4_query1:app', host="127.0.0.1", port=8000, reload=True)

② 数据库操作 – 条件查询

select(Book).where(条件1，条件2，……)

条件：

        比较判断：==；<;>;<=;>=等

        模糊查询：like()；%：零个、一个或多个字符；_：一个单个字符

        与非查询：&;|;~

        包含查询：in_()

import uvicorn
from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from sqlalchemy import select

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
# 返回ORM对象
# result = await db.execute(select(Book))
# 查询1 获取所有
# book = result.scalars().all()

# 查询2 获取第一个
# book = result.scalars().first()

# 查询3 根据主键获取单条数据
book = await db.get(Book, 3)
return book

@app.get("/book/fuzzy_query1")
async def get_fuzzy_query_book1(db: AsyncSession = Depends(get_database)):
# like()模糊查询： % 任意个字符
result = await db.execute(select(Book).where(Book.author.like("曹%")))
books = result.scalars().all()
return books

@app.get("/book/fuzzy_query2")
async def get_fuzzy_query_book2(db: AsyncSession = Depends(get_database)):
# like()模糊查询： _ 一个单个字符
result = await db.execute(select(Book).where(Book.bookName.like("平%")))
books = result.scalars().all()
return books

@app.get("/book/search_book1")
async def get_search_book(db: AsyncSession = Depends(get_database)):
result = await db.execute(select(Book).where(Book.price >= 20))
books = result.scalars().all()
return books

@app.get("/book/search_book2")
async def get_search_book(db: AsyncSession = Depends(get_database)):
# 逻辑运算符，扩充条件： 与 &
result = await db.execute(select(Book).where ((Book.price >= 20) & Book.author.like("曹%")))
books = result.scalars().all()
return books

@app.get("/book/search_book3")
async def get_search_book(db: AsyncSession = Depends(get_database)):
# 逻辑运算符，扩充条件： 或 |
result = await db.execute(select(Book).where ((Book.price >= 20) | Book.author.like("曹%")))
books = result.scalars().all()
return books

@app.get("/book/search_book4")
async def get_search_book(db: AsyncSession = Depends(get_database)):
# 逻辑运算符，扩充条件： 非 ~
result = await db.execute(select(Book).where ((Book.price >= 20) & ~Book.author.like("曹%")))
books = result.scalars().all()
return books

@app.get("/book/search_book5")
async def get_search_book(db: AsyncSession = Depends(get_database)):
# in_() 包含
id_list = [1, 2, 3, 4]
result = await db.execute(select(Book).where(Book.id.in_(id_list)))
books = result.scalars().all()
return books

@app.get("/book/{book_id}")
async def get_book_by_id(book_id: int, db: AsyncSession = Depends(get_database)):
result = await db.execute(select(Book).where(Book.id == book_id))
book = result.scalar_one_or_none()
return book

if __name__ == "__main__":
uvicorn.run('demo4_query:app', host="127.0.0.1", port=8000, reload=True)

③ 数据库操作 – 聚合查询

聚合计算：func.方法（模型类.属性）

        count：统计行数量

        avg：求平均值

        max：求最大值

        min：求最小值

        sum：求和

import uvicorn
from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from sqlalchemy import select

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
# 返回ORM对象
# result = await db.execute(select(Book))
# 查询1 获取所有
# book = result.scalars().all()

# 查询2 获取第一个
# book = result.scalars().first()

# 查询3 根据主键获取单条数据
book = await db.get(Book, 3)
return book

@app.get("/book/count_query1")
async def get_count(db: AsyncSession = Depends(get_database)):
# func.count()计数
result = await db.execute(select(func.count(Book.id)))
# scalar() 用于提取一个数值 ——> 标量值
num = result.scalar()
return num

@app.get("/book/count_query2")
async def get_count(db: AsyncSession = Depends(get_database)):
# func.max() 最大值
result = await db.execute(select(func.max(Book.price)))
# scalar() 用于提取一个数值 ——> 标量值
num = result.scalar()
return num

@app.get("/book/count_query3")
async def get_count(db: AsyncSession = Depends(get_database)):
# func.sum() 求和
result = await db.execute(select(func.sum(Book.price)))
# scalar() 用于提取一个数值 ——> 标量值
num = result.scalar()
return num

@app.get("/book/count_query4")
async def get_count(db: AsyncSession = Depends(get_database)):
# func.avg() 求平均
result = await db.execute(select(func.avg(Book.price)))
# scalar() 用于提取一个数值 ——> 标量值
num = result.scalar()
return num

@app.get("/book/{book_id}")
async def get_book_by_id(book_id: int, db: AsyncSession = Depends(get_database)):
result = await db.execute(select(Book).where(Book.id == book_id))
book = result.scalar_one_or_none()
return book

if __name__ == "__main__":
uvicorn.run('demo4_query:app', host="127.0.0.1", port=8000, reload=True)

④ 数据库操作 – 分页查询

分页查询：select().offset().limit()

        offset：跳过的记录数

        limit：返回的记录数

offset值 = （当前页码 – 1） * 每页数量 limit

import uvicorn
from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from sqlalchemy import select

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

@app.get("/book/get_book_list")
async def get_book_list(
page: int = 1,
page_size: int = 2,
db: AsyncSession = Depends(get_database)
):
# （页码 – 1） * 每页数量
skip = (page – 1) * page_size
stmt = select(Book).offset(skip).limit(page_size)
result = await db.execute(stmt)
books = result.scalars().all()
return books

if __name__ == "__main__":
uvicorn.run('demo5_splitPageQuery:app', host="127.0.0.1", port=8000, reload=True)

⭐ ORM – 查询 – 总结

核心思路：

        select() ——> db.execute()：从ORM对象获取数据——>响应结果

        db.get(模型类，主键值)

从ORM对象获取数据的方式

获取所有数据

        scalars().all()

获取单条数据

        scalars().first()：提取第一个数据

        scalar_one_or_none()：提取一个或null

        scalar()：提取标量值（配合聚合查询使用）

Ⅴ、操作数据 – 新增

核心步骤：定义ORM对象 ——> 添加对象到事务：add(对象) ——> commit 提交到数据库

import uvicorn
from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from pydantic import BaseModel

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

class BookBase(BaseModel):
id: int
bookName: str
author: str
price: float
publisher: str

# 用户输入图书信息（id、书名、作者、价格、出版社） ——> 新增
# 用户输入 ——> 参数 ——> 请求体参数
@app.post("/book/add_book")
async def add_book(book: BookBase, db : AsyncSession = Depends(get_database)):
# ORM对象 ——> add ——> commit
book_obj = Book(**book.__dict__)
db.add(book_obj)
await db.commit()
return book

if __name__ == "__main__":
uvicorn.run('demo6_addDatabase', host="127.0.0.1", port=8000, reload=True)

Ⅴ、操作数据 – 更新

核心步骤：查询 get ——> 属性重新赋值 ——> commit 提交到数据库

Ⅵ、操作数据 – 删除

核心步骤：查询 get ——> delete 删除 ——> commit 提交到数据库

import uvicorn
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import DateTime, func, String, Float
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from pydantic import BaseModel

app = FastAPI()

# 1. 创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://xxxx@localhost:3306/FastAPI_db?charset=utf8"

async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo = True, # 可选，输出SQL日志
pool_size = 10, # 设置连接池中保持的持久连接数
max_overflow = 20 # 设置连接池允许创建的额外连接数
)

# 2. 定义模型类：基类+表对应的模型类
# 基类：创建时间、更新时间; 书籍表：id、书名、作者、价格、出版社
class Base(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, comment="创建时间")
update_time: Mapped[datetime] = mapped_column(
DateTime, insert_default=func.now(), default=datetime.now, onupdate=func.now(), comment="修改时间")

class Book(Base):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
bookName: Mapped[str] = mapped_column(String(255), comment="书名")
author: Mapped[str] = mapped_column(String(255), comment = "作者")
price: Mapped[float] = mapped_column(Float, comment="价格")
publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

# 3.建表： 定义函数建表 ——> FastAPI 启动的时候调用建表的函数
async def create_tables():
# 获取异步引擎，创建事务 – 建表
async with async_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all) # Base 模型类的元数据创建

@app.on_event("startup")
async def startup_event():
await create_tables()

@app.get("/")
async def root():
return {"message": "Hello World"}

# 需求：查询功能的接口，查询图书 ——> 依赖注入：创建依赖项获取数据库会话 + Depends 注入路由处理函数
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine, # 绑定数据库引擎
class_=AsyncSession, # 指定会话类
expire_on_commit=False # 提交后会话对象不过期，不重新查询数据库
)

# 依赖项：用于获取数据库会话
async def get_database():
async with AsyncSessionLocal() as session:
try:
yield session # 返回数据库会话给路由处理函数
await session.commit() # 无异常，提交事务
except Exception:
await session.rollback() # 有异常则回滚
raise
finally:
await session.close() # 关闭会话

class BookBase(BaseModel):
id: int
bookName: str
author: str
price: float
publisher: str

class BookDelate(BaseModel):
bookName: str
author: str
price: float
publisher: str

# 修改图书信息：先查再改
# 设计思路：路径参数书籍id
@app.post("/book/delate_book/{book_id}")
async def delete_book(book_id: int, data: BookBase, db : AsyncSession = Depends(get_database)):
db_book = await db.get(Book, book_id)
if db_book is None:
raise HTTPException(
status_code=404,
detail="查无此书"
)

# 重新赋值
await db.delete(db_book)
await db.commit()
return {"msg": "删除图书成功"}

if __name__ == "__main__":
uvicorn.run('demo8_deleteDatabase:app', host="127.0.0.1", port=8000, reload=True)

⭐Ⅶ、整体操作流程

导入依赖 → 初始化FastAPI应用 → 配置数据库连接 → 定义ORM模型 → 配置自动建表 → 创建会话工厂 → 定义数据库依赖 → 编写CRUD路由 → 启动应用