目录
- 一，什么是ORM？
- 二，如何使用？
- 1.安装
- 2.配置异步数据库引擎
- 3.定义模型
- 3.1 Mapped[T] 的作用
- 3.2 mapped_column() 常用参数详解
- 3.3 模型
- 4.获取数据库 Session (Dependency)
一，什么是ORM？
ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，它负责在关系型数据库（如 MySQL、PostgreSQL）和面向对象编程语言（如 Python）之间建立一座“桥梁”。
简单来说就是ORM 允许你用 Python 类来定义表结构，用操作对象来代替 SQL 语句。
Python 生态中有多个 ORM 方案，比如 Django ORM、Peewee、Tortoise-ORM。而 SQLAlchemy 是最流行的选择。
SQLAlchemy具有诸多优点：
- 功能强大：同时提供ORM（对象操作）和Core（SQL 表达式），灵活度极高。
- 数据库无关：一套代码支持 MySQL、PostgreSQL、SQLite 等，切换仅改连接串。
- 高性能可控：支持懒加载/预加载、批量操作，避免 N+1 问题。
- 异步原生支持：
sqlalchemy.ext.asyncio可与 FastAPI 完美配合，不阻塞事件循环。 - 迁移成熟：配合 Alembic 实现自动化数据库版本管理。
- 社区事实标准：文档齐全，生态庞大，长期维护可靠。
二，如何使用？
1.安装
首先创建一个FastAPI项目，安装sqlalchemy（ORM 框架）、aiomysql（异步驱动）以及 pydantic。
注：还可以用同步驱动，但是在 FastAPI 这类异步 Web 框架中，使用它会因阻塞事件循环而浪费掉框架最大的并发优势，因此在高并发生产环境中应用不广。
在控制台运行：pip install sqlalchemy aiomysql fastapi uvicorn
2.配置异步数据库引擎
如下，
fromsqlalchemy.ext.asyncioimportcreate_async_engine,AsyncSession,async_sessionmakerfromsqlalchemy.ormimportDeclarativeBase# 数据库连接配置DATABASE_URL="mysql+aiomysql://user:password@localhost:3306/dbname"# 创建异步引擎engine=create_async_engine(DATABASE_URL,echo=True)# 创建异步 Session 工厂AsyncSessionLocal=async_sessionmaker(bind=engine,class_=AsyncSession,expire_on_commit=False)# 定义模型基类classBase(DeclarativeBase):passDATABASE_URL用来告诉 SQLAlchemy 用什么驱动连接哪个数据库服务，通常的写法是：
mysql+aiomysql://用户名:密码@主机:端口/数据库名?可选参数create_async_engine(DATABASE_URL, echo=True) 用来创建一个异步引擎。
第一个参数DATABASE_URL就是之前设置的固定格式的字符串变量。
第二个参数echo=True让引擎输出所有执行的 SQL 语句到控制台（便于开发调试）。
还可以通过pool_size=10设置连接池中保持的常驻连接数（默认 5），max_overflow=20来设置当连接池满时，额外临时创建的连接数（默认 10）
AsyncSessionLocal以及后面的一串代码创建了一个异步 Session 工厂，作用是每次调用它时生成一个独立的数据库会话（AsyncSession）。
bind=engine：把之前创建的异步引擎绑定给 Session 工厂，这样生成的 Session 就知道该连接哪个数据库。class_=AsyncSession：指定要生成的 Session 类型为异步版本（默认也是AsyncSession，这里显式写出来更清晰）。expire_on_commit=False：控制提交后的行为。True（默认）：commit()后，Session 中所有对象的属性会被标记为“过期”，下次访问时会自动重新从数据库加载。这在异步中可能导致不必要的延迟加载（而且异步模式懒加载可能有问题）。False：提交后不使对象过期，你可以继续访问对象的属性（不会触发额外查询），性能更好、行为更可控。在 FastAPI + SQLAlchemy 异步模式下，通常建议设为False。
最后的定义模型类通常会另外创建一个文件来设置，而且一般会写表的创建时间和更新时间，如下：
fromdatetimeimportdatetimefromsqlalchemyimportfuncfromsqlalchemy.ormimportMapped,mapped_column,DeclarativeBaseclassBase(DeclarativeBase):__abstract__=True# 表示这是一个抽象基类，不生成表created_at:Mapped[datetime]=mapped_column(server_default=func.now())updated_at:Mapped[datetime]=mapped_column(server_default=func.now(),onupdate=func.now())然后让其他模型继承 Base 即可。（还可以用混入类的方式来写创建时间之类的，更灵活）
3.定义模型
示例：
fromsqlalchemy.ormimportMapped,mapped_columnfromsqlalchemyimportStringclassUser(Base):__tablename__="users"id:Mapped[int]=mapped_column(primary_key=True)username:Mapped[str]=mapped_column(String(50),unique=True)email:Mapped[str]=mapped_column(String(100))3.1 Mapped[T] 的作用
- 类型提示：告诉静态类型检查器（如 mypy、Pyright）这个属性在 ORM 实例中是什么 Python 类型。
- 运行时信息：SQLAlchemy 会扫描 Mapped 注解，结合 mapped_column 生成元数据。
- 支持可选类型：Mapped[Optional[str]] 或 Mapped[str | None] 表示数据库字段可为 NULL。
3.2 mapped_column() 常用参数详解
| 参数 | 说明 | 示例 |
|---|---|---|
| 类型 | 第一个位置参数，指定 SQL 类型 | mapped_column(String(100)) |
primary_key | 是否为主键 | primary_key=True |
unique | 是否唯一约束 | unique=True |
nullable | 是否可为 NULL（默认 True，主键自动为 False） | nullable=False |
default | Python 层面的默认值（插入时若未提供则使用此值） | default="draft",default=uuid.uuid4 |
server_default | 数据库层面的默认值，使用数据库函数 | server_default=func.now() |
index | 是否创建索引 | index=True |
autoincrement | 自增（整数主键默认自增） | autoincrement=True |
comment | 列注释 | comment="用户名称" |
name | 数据库中的列名，若不指定则使用属性名 | name="user_name" |
注：
- mapped_column 本身在类定义时执行，不涉及同步/异步问题。
- 类型注解与 mapped_column 的 SQL 类型要匹配：比如 Mapped[str] 但 mapped_column(Integer) 会报错。
- 若不写类型注解：可以只用 mapped_column()，但失去了类型检查优势。
- 可选字段必须使用 Optional 或 | None，否则 mypy 会报错。
3.3 模型
模型（Model）用来规定Python 对象如何映射到数据库中的表结构，也就是“对象关系映射”（ORM）的核心：
- 类名
User→ 对应数据库中的表名（__tablename__ = "users"） - 类属性（如
id,username,email）→ 对应表中的列 Mapped[int]→ 指定该列在 Python 中的数据类型mapped_column(...)→ 定义列的数据库细节：类型、是否主键、是否唯一、默认值等
简单说：你写的User类就像一张“设计图纸”，SQLAlchemy 根据这张图纸在数据库中自动创建（或匹配）相应的users表，之后你对 PythonUser对象的增删改查操作，会被自动翻译成对数据库表的 SQL 语句。
例如：
- 创建
User(id=1, username="alice", email="alice@example.com")→ 对应INSERT INTO users (id, username, email) VALUES (1, 'alice', 'alice@example.com') - 查询
session.execute(select(User))→ 对应SELECT * FROM users
这样你就不需要直接写 SQL，而是通过操作熟悉的 Python 对象来与数据库交互。
注意，需要继承Base基类，这个Base基类就是之前继承了DeclarativeBase的类。
4.获取数据库 Session (Dependency)
利用 FastAPI 的 Depends 来注入 Session，确保每次请求都能正确关闭连接。
依赖一般像下面这样写：
asyncdefget_db():asyncwithAsyncSessionLocal()assession:try:yieldsessionawaitsession.commit()# 无异常时自动提交exceptException:awaitsession.rollback()# 出现异常自动回滚raise# 继续抛出异常，让 FastAPI 处理# finally 不需要显式 close，async with 退出时已经关闭 session，也可以写一下更显眼……用于在每个请求中获取一个数据库会话（AsyncSession），并在请求结束后自动处理事务的提交或回滚，同时确保会话被正确关闭。
这是一个异步生成器。AsyncSessionLocal()就是之前创建的异步session工厂。
路由通过yield session获取到一个session对象，然后就可以用这个对象来执行数据库操作。
此时生成器函数就会停在这里，等待路由函数执行完，如果没问题，就会执行后面的提交代码，如果有问题，就会回滚，然后抛出异常。
