在Python+FastAPI项目中使用SqlAlchemy操作数据的几种常见方式

克里斯顿 發表於 2025-12-25 11:59:00

在Python+FastAPI项目中使用SqlAlchemy操作数据的几种常见方式

在Python+FastAPI的后端项目中，我们往往很多时候需要对数据进行相关的处理，本篇随笔介绍在Python+FastAPI项目中使用SqlAlchemy操作数据的几种常见方式。
使用 FastAPI, SQLAlchemy, Pydantic构建后端项目的时候，其中数据库访问采用SQLAlchemy 的异步方式处理。一般我们在操作数据库操作的时候，采用基类继承的方式减少重复代码，提高代码复用性。不过我们在分析SQLAlchemy的时候，我们可以简单的方式来剖析几种常见的数据库操作方式，来介绍SQLAlchemy的具体使用。
<h3>1、SQLAlchemy介绍</h3>
<div>SQLAlchemy 是一个功能强大且灵活的 Python SQL 工具包和对象关系映射（ORM）库。它被广泛用于在 Python 项目中处理关系型数据库的场景，既提供了高级的 ORM 功能，又保留了对底层 SQL 语句的强大控制力。<code>SQLAlchemy</code> 允许开发者通过 Python 代码与数据库进行交互，而无需直接编写 SQL 语句，同时也支持直接使用原生 SQL 进行复杂查询。下面是SQLAlchemy和我们常规数据库对象的对应关系说明。</div>
<div>Engine  　　连接对象         驱动引擎</div>
<div>Session 　　连接池           事务  由此开始查询</div>
<div>Model   　　表                   类定义</div>
<div>Column  　  列  </div>
<div>Query   　　若干行         可以链式添加多个条件</div>
<div> </div>
<div>在使用SQLAlchemy时，通常会将其与数据库对象对应起来。</div>
<div>SQLAlchemy: 使用 <code>Table</code> 对象或 <code>Declarative Base</code> 中的类来表示。</div>
<div>对应关系: 数据库中的每一个表对应于SQLAlchemy中的一个类，该类继承自 <code>declarative_base()</code>。</div>
<div>
<div class="cnblogs_code">
<pre>from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
__tablename__ = 'users'# 数据库表名
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)</pre>
</div>
数据库列 (Database Column)：使用 <code>Column</code> 对象来表示。每个数据库表中的列在SQLAlchemy中表示为 <code>Column</code> 对象，并作为类的属性定义。
<div>
<div class="cnblogs_code">
<pre>id = Column(Integer, primary_key=True)
name = Column(String(50))</pre>
</div>
数据库行 (Database Row)：每个数据库表的一个实例（对象）代表数据库表中的一行。在SQLAlchemy中，通过实例化模型类来表示数据库表中的一行。
<div>
<div class="cnblogs_code">
<pre>new_user = User(id=1, name='John Doe', email='john@example.com')</pre>
</div>
主键 (Primary Key)：使用 <code>primary_key=True</code> 参数定义主键。
</div>
</div>
<div>
<div class="cnblogs_code">
<pre>id = Column(Integer, primary_key=True)</pre>
</div>
外键 (Foreign Key): 使用 <code>ForeignKey</code> 对象来表示。
<div>
<div class="cnblogs_code">
<pre>from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
user = relationship('User')</pre>
</div>
关系 (Relationships): 使用 <code>relationship</code> 对象来表示。数据库中表与表之间的关系在SQLAlchemy中通过 <code>relationship</code> 来定义。
<div>
<div class="cnblogs_code">
<pre>addresses = relationship("Address", back_populates="user")</pre>
</div>
 
</div>
</div>
</div>
</div>
<h3>2、常规的单表处理</h3>
下面我们通过异步处理的方式，介绍如何在单表中操作相关的数据库数据。
<div class="cnblogs_code">
<pre>async def get(self, db: AsyncSession, id: Any) -> Any:
"""根据主键获取一个对象"""

if isinstance(id, str):
 query = select(self.model).filter(func.lower(self.model.id) == id.lower())
else:
 query = select(self.model).filter(self.model.id == id)

result = await db.execute(query)
item = result.scalars().first()

return item</pre>
</div>
如果我们需要强制对外键的类型进行匹配（如对于Postgresql的严格要求，数据比较的类型必须一致），那么我们需要在基类或者CRUD类初始化的时候，获得对应的主键类型。
<div class="cnblogs_code">
<pre>class BaseCrud(Generic):
"""
基础CRUD操作类，传入参数说明：
* `ModelType`: SQLAlchemy 模型类
* `PrimaryKeyType`: 限定主键的类型
* `PageDtoType`: 分页查询输入类
* `DtoType`: 数据传输对象类，如新增、更新的单个对象DTO
"""

def __init__(self, model: Type):
 """
 数据库访问操作的基类对象(CRUD).
 * `model`: A SQLAlchemy model class
 """

 self.model = model# 模型类型

 # 运行期获取主键字段类型
 pk_column = inspect(model).primary_key
 self._pk_type = pk_column.type.python_type# int / str</pre>
</div>
因此对于单表的Get方法，我们修改下，让他匹配主键的类型进行比较，这样过对于严格类型判断的Postgresql也正常匹配了。
<div class="cnblogs_code">
<pre> async def get(self, db: AsyncSession, id: PrimaryKeyType) -> Optional:
 """根据主键获取一个对象"""

 #对id的主键进行类型转换，self._pk_type在构造函数的初始化中获取
 try:
 id = self._pk_type(id)
 except Exception:
 raise ValueError(f"Invalid primary key type: {id}")

 if isinstance(id, str):
 query = select(self.model).filter(func.lower(self.model.id) == id.lower())
 else:
 query = select(self.model).filter(self.model.id == id)

 result = await db.execute(query)
 item = result.scalars().first()

 return item</pre>
</div>
对于删除的数据，我们也可以类似的处理对比进行了。
<div class="cnblogs_code">
<pre>from sqlalchemy.orm import Session, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import delete as sa_delete, update as sa_update

async def delete_byid(self, db: AsyncSession, id: PrimaryKeyType) -> bool:
"""根据主键删除一个对象

:param id: 主键值
"""

#对id的主键进行类型转换，self._pk_type在构造函数的初始化中获取
try:
 id = self._pk_type(id)
except Exception:
 raise ValueError(f"Invalid primary key type: {id}")

del_query: sa_delete
if isinstance(id, str):
 del_query = sa_delete(self.model).where(
 func.lower(self.model.id) == id.lower()
 )
else:
 del_query = sa_delete(self.model).where(self.model.id == id)

result = await db.execute(del_query)

await db.commit()
return result.rowcount > 0</pre>
</div>
对于提供多条件的查询或者过滤，我们可以使用where函数或者filter函数，在 SQLAlchemy 中，<code>select(...).where(...)</code> 和 <code>select(...).filter(...)</code> 都用于构造查询条件，如下所示等效。
<div class="cnblogs_code">
<pre>query = select(self.model).where(self.model.id == id)

query = select(self.model).filter(self.model.id == id)</pre>
</div>
我们可以通过sqlAlchemy的and_和or_函数来进行组合多个条件。
<div class="cnblogs_code">
<pre>from sqlalchemy import ( Table,Column,and_,or_,asc,desc,select,func,distinct,text, Integer)

....
match expression:
 case "and":
 query = await db.execute(
 select(self.model)
 .filter(and_(*where_list))
 .order_by(*order_by_list)
 )
 case "or":
 query = await db.execute(
 select(self.model).filter(or_(*where_list)).order_by(*order_by_list)
 )</pre>
</div>
Python的SqlAlchemy提供 InstrumentedAttribute 对象来操作多个条件，如我们对于一些多条件的处理，可以利用它来传递多个参数。
<div class="cnblogs_code">
<pre> async def get_all_by_attributes(
 self, db: AsyncSession, *attributes: InstrumentedAttribute, sorting: str = ""
) -> List | None:
 """根据列名称和值获取相关的对象列表

 :param sorting: 格式：name asc 或 name asc,age desc
 :param attributes: SQLAlchemy InstrumentedAttribute objects，可以输入多个条件
 例子：User.id != 1 或者 User.username == "JohnDoe"
 """

 order_by_list = parse_sort_string(sorting, self.model)
 query = select(self.model).filter(and_(*attributes)).order_by(*order_by_list)

 result = await db.execute(query)
 return result.scalars().all()</pre>
</div>
例如，对于模型 Material 对象，我们对它进行多个条件的查询处理，如下所示，红色部分为 *attributes: InstrumentedAttribute 参数。
<div class="cnblogs_code">
<pre>items = await super().get_all_by_attributes(
db,
 Material.id == vercol.id,
Material.vercol == vercol.vercol,
Material.ischecked == 0,
Material.status == 0,
)</pre>
</div>
同样我们可以利用它来获取数量，或者判断多条件的记录是否存在。
<img src="https://img2024.cnblogs.com/blog/8867/202512/8867-20251225103430193-1873231330.png" alt="image" width="944" height="532" loading="lazy">
 在数据插入或者更新的操作中，我们可以接受对象类型或者字典类型的参数对象，因此方法如下所示。
<div class="cnblogs_code">
<pre> async def update(self, db: AsyncSession, obj_in: DtoType | dict) -> bool:
 """更新对象

 :param obj_in: 对象输入数据,可以是 DTO 对象或字典
 """
 try:
 if isinstance(obj_in, dict):
 obj_id = obj_in.get("id")
 if obj_id is None:
 raise ValueError("id is required for update")
 update_data = obj_in
 else:
 obj_id = obj_in.id
 # update_data = vars(obj_in)
 update_data = obj_in.model_dump(exclude_unset=True)

 query = select(self.model).filter(self.model.id == obj_id)
 result = await db.execute(query)
 db_obj = result.scalars().first()

 if db_obj:
 # 更新对象字段
 for field, value in update_data.items():
 # 跳过以 "_" 开头的私有属性
 if field.startswith("_"):
 continue
 setattr(db_obj, field, value)

 # 处理更新前的回调处理
 self.on_before_update(update_data, db_obj)
 # 提交事务
 await db.commit()
 return True
 else:
 return False
 except SQLAlchemyError as e:
 self.logger.error(f"update 操作出现错误: {e}")
 await db.rollback()# 确保在出错时回滚事务
 return False</pre>
</div>
我们在插入或者更新数据的时候，一般会默认更新一些字段，如创建人，创建日期、编辑人，编辑日期等信息，我们可以把它单独作为一个可以给子类重写的函数，基类做一些默认的处理。
<div class="cnblogs_code">
<pre> def on_before_update(self, update_data: dict, db_obj: ModelType) -> None:
 """更新对象前的回调函数，子类可以重写此方法

 可通过 setattr(db_obj, field, value) 设置字段值
 """

 setattr(db_obj, "edittime", datetime.now())
 user :CurrentUserIns= get_current_user()
 if user:
 setattr(db_obj, "editor", user.fullname)
 setattr(db_obj, "editor_id", user.id)

 setattr(db_obj, "company_id", user.company_id)
 setattr(db_obj, "companyname", user.companyname)</pre>
</div>
有时候，如果我们需要获取某个字段非重复的列表，用来做为动态下拉列表的数据，那么我们可以通过下面函数封装下。
<div class="cnblogs_code">
<pre> async def get_field_list(self, db: AsyncSession, field_name: str) -> Iterable:
 """获取指定字段值的唯一列表

 :param field_name: 字段名称
 """

 field = getattr(self.model, field_name)
 query = select(distinct(field))
 result = await db.execute(query)
 return result.scalars().all()</pre>
</div>
 
<h3>3、多表联合的处理操作 </h3>
多表操作，也是我们经常碰到的处理方式，如对于字典类型和字典项目，他们是两个表，需要联合起来获取数据，那么就需要多表的联合操作。
<img src="https://img2024.cnblogs.com/blog/8867/202512/8867-20251225110211988-1085898697.png" alt="image" width="238" height="326" loading="lazy">
 如下是字典CRUD类中，联合字典类型获取数据的记录处理。
<div class="cnblogs_code">
<pre> async def get_dict_by_typename(self, db: AsyncSession, dicttype_name: str) -> dict:
 """根据字典类型名称获取所有该类型的字典列表集合"""
 result = await db.execute(
 select(self.model)
 .join(DictType, DictType.id == self.model.dicttype_id)# 关联字典类型表
 .filter(DictType.name == dicttype_name) # 过滤字典类型名称
 .order_by(DictData.seq)# 排序
 )
 items = result.scalars().all()

 dict = {}
 for info in items:
 if info.name not in dict:
 dict = info.value

 return dict</pre>
</div>
如果我们需要对某个表的递归获取树列表，可以如下处理
<div class="cnblogs_code">
<pre> async def get_tree(self, db: AsyncSession, pid: str) -> list:
 """获取字典类型一级列表及其下面的内容"""

 # 使用三元运算符将 pid 设为 "-1"（如果 pid 是 null 或空白）或保持原值
 pid = "-1" if not pid or pid.strip() == "" else pid
 result = await db.execute(
 select(self.model)
 .filter(self.model.pid == pid)
 .options(selectinload(DictType.children))
 )
 nodes = result.scalars().all()
 return nodes</pre>
</div>
我们来假设用户和文章的示例表结构（ORM 模型，如下所示。
<div class="cnblogs_code">
<pre>class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String)
email = Column(String)
 articles = relationship("Article", back_populates="author")

class Article(Base):
__tablename__ = "articles"
id = Column(Integer, primary_key=True, index=True)
title = Column(String)
content = Column(Text)
 user_id = Column(Integer, ForeignKey("users.id"))
author = relationship("User", back_populates="articles")</pre>
</div>
我们可以通过下面函数处理获得相关的记录集合。
<div class="cnblogs_code">
<pre>async def get_user_articles(db: AsyncSession):
stmt = (
 select(User, Article)
 .join(Article, Article.user_id == User.id)
)
result = await db.execute(stmt)
return result.all() # [(User(), Article()), ...]</pre>
</div>
如果需要可以使用outer_join函数处理
<div class="cnblogs_code">
<pre>async def get_users_with_articles(db: AsyncSession):
stmt = (
 select(User, Article)
 .outerjoin(Article, Article.user_id == User.id)
)
result = await db.execute(stmt)
return result.all()# 用户即便没有文章也会出现</pre>
</div>
 
如果我们需要获取有文章的所有用户，如下所示。
<div class="cnblogs_code">
<pre>async def get_users_with_articles(db: AsyncSession):
stmt = select(User).options(selectinload(User.articles))# 自动 load 关联
result = await db.execute(stmt)
return result.scalars().all()</pre>
</div>
<code data-start="1468" data-end="1482">selectinload</code> 会执行两次 SQL，但效率高，不会产生笛卡尔积，非常适合集合查询。
多表链式 Join的处理，可以获得两个表的不同信息进行组合。
<div class="cnblogs_code">
<pre>async def get_articles_with_author(db: AsyncSession):
stmt = (
 select(Article.title, User.name.label("author"))
 .join(User, Article.user_id == User.id)
)
rows = await db.execute(stmt)
return rows.mappings().all()# 以 dict 形式返回 [{'title':..., 'author':...}]</pre>
</div>
带筛选条件与分页的处理实现，如下所示
<div class="cnblogs_code">
<pre>async def search_articles(db: AsyncSession, keyword: str, page: int = 1, size: int = 10):
stmt = (
 select(Article, User.name.label("author"))
 .join(User)
 .filter(Article.title.contains(keyword))
 .offset((page - 1) * size)
 .limit(size)
)
result = await db.execute(stmt)
return result.mappings().all()</pre>
</div>
 
对于权限管理系统来说，一般有用户、角色，以及用户角色的中间表，一般来说他们的关系如下
<img src="https://img2024.cnblogs.com/blog/8867/202512/8867-20251225122040874-818263456.png" alt="image" width="729" height="363" loading="lazy">
 在界面中一般会提供选择用户的操作关联，如角色中维护用户列表。
<img src="https://img2024.cnblogs.com/blog/8867/202512/8867-20251225121122302-651245416.png" alt="image" width="481" height="426" loading="lazy">
 我们来看看这个在SQLAlchemy最佳实践是如何的操作。
<div class="cnblogs_code">
<pre>from sqlalchemy import Table, Column, Integer, ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column
from database import Base

# --- 中间表写法 ---
role_user = Table(
"role_user",
Base.metadata,
Column("user_id", ForeignKey("users.id"), primary_key=True),
Column("role_id", ForeignKey("roles.id"), primary_key=True)
)

class User(Base):
__tablename__ = "users"
id: Mapped = mapped_column(primary_key=True)
username: Mapped = mapped_column()

roles: Mapped] = relationship(
 secondary=role_user,
 back_populates="users",
 lazy="selectin"
)

class Role(Base):
__tablename__ = "roles"
id: Mapped = mapped_column(primary_key=True)
name: Mapped = mapped_column()

users: Mapped] = relationship(
 secondary=role_user,
 back_populates="roles",
 lazy="selectin"
)</pre>
</div>
在 SQLAlchemy 声明多对多关系时，<code data-start="22" data-end="33">secondary</code> 参数既可以填 字符串形式的表名，也可以填 已经定义好的中间表对象（Table 对象）。
① econdary="role_user" —— 使用字符串表名
<div class="cnblogs_code">
<pre>roles = relationship("Role", secondary="role_user", back_populates="users")</pre>
</div>
② secondary=role_user —— 传入中间表对象（推荐方式）
<div class="cnblogs_code">
<pre>role_user = Table(
"role_user",
Base.metadata,
Column("user_id", ForeignKey("users.id"), primary_key=True),
Column("role_id", ForeignKey("roles.id"), primary_key=True)
)

roles = relationship("Role", secondary=role_user, back_populates="users")</pre>
</div>
对于如果获取对应角色的用户记录，我们可以通过下面方式获取（通过连接中间表的方式）
<div class="cnblogs_code">
<pre>async def get_users_by_role(db: AsyncSession, role_id: int) -> list:
stmt = (
 select(User)
 .join(role_user, role_user.c.user_id == User.id)
 .where(role_user.c.role_id == role_id)
)
result = await db.execute(stmt)
return result.scalars().all()</pre>
</div>
也可以下面的方式进行处理（使用 relationship any()），效果是一样的。
<div class="cnblogs_code">
<pre>select(User).filter(User.roles.any(id=role_id))</pre>
</div>
如果需要写入用户、角色的关联关系，我们可以使用下面方法来通过中间表进行判断并写入记录。
<div class="cnblogs_code">
<pre>from sqlalchemy import select, insert

async def add_users_to_role(db: AsyncSession, role_id: int, user_ids: list):
# 1️⃣ 查询已有关联 user_id
stmt = select(role_user.c.user_id).where(role_user.c.role_id == role_id)
res = await db.execute(stmt)
existing_user_ids = {row for row in res.fetchall()}

# 2️⃣ 过滤出新的 user_id
new_user_ids =
if not new_user_ids:
 return 0# 没有新增

# 3️⃣ 批量插入
values = [{"user_id": uid, "role_id": role_id} for uid in new_user_ids]
stmt = insert(role_user).values(values)
await db.execute(stmt)
await db.commit()

return len(new_user_ids)</pre>
</div>
如果只是单个记录的插入，可以利用下面的方式处理。
<div class="cnblogs_code">
<pre>from sqlalchemy import select, insert

async def add_user_to_role(db: AsyncSession, role_id: int, user_id: int) -> bool:
# 1️⃣ 检查是否已存在关联
stmt = select(role_user).where(
 role_user.c.role_id == role_id,
 role_user.c.user_id == user_id
)
res = await db.execute(stmt)
exists = res.first()

if exists:
 return False# 已存在，不再插入

# 2️⃣ 插入记录
stmt = insert(role_user).values(user_id=user_id, role_id=role_id)
await db.execute(stmt)
await db.commit()

return True</pre>
</div>
删除用户角色的关联关系，通过下面函数进行处理即可
<div class="cnblogs_code">
<pre>from sqlalchemy import and_, or_, select
from sqlalchemy import delete as sa_delete, update as sa_update

async def remove_user(self, db: AsyncSession, role_id: int, user_id: int) -> bool:
 """移除角色-用户关联"""
 stmt = sa_delete(user_role).where(
 and_(
 user_role.c.role_id == role_id,
 user_role.c.user_id == user_id,
 )
 )
 result = await db.execute(stmt)
 await db.commit()

 # rowcount 返回删除行数
 return result.rowcount > 0</pre>
</div>
 
以上就是对于在Python+FastAPI的后端项目中使用SqlAlchemy操作数据的几种常见方式，包括单表处理，多表关联、中间表的数据维护和定义等内容，是我们在操作常规数据的时候，经常碰到的几种方式。
希望上文对你有所启发和帮助，感谢阅读。

</div>
<div id="MySignature" role="contentinfo">
<div style="border-right-color: #cccccc; border-right-width: 1px; border-right-style: solid; padding-right: 5px; border-top-color: #cccccc; border-top-width: 1px; border-top-style: solid; padding-left: 4px; font-size: 13px; padding-bottom: 4px; border-left-color: #cccccc; border-left-width: 1px; border-left-style: solid; width: 98%; padding-top: 4px; border-bottom-color: #cccccc; border-bottom-width: 1px; border-bottom-style: solid; background-color: #eeeeee;">
<img src="http://www.cnblogs.com/Images/OutliningIndicators/None.gif" align="top" alt>

专注于代码生成工具、.Net/Python 框架架构及软件开发，以及各种Vue.js的前端技术应用。著有Winform开发框架/混合式开发框架、微信开发框架、Bootstrap开发框架、ABP开发框架、SqlSugar开发框架、Python开发框架等框架产品。
 转载请注明出处：撰写人：伍华聪 http://www.iqidi.com </div> 
来源：https://www.cnblogs.com/wuhuacong/p/19396612

頁: [1]

琼殿技术论坛's Archiver

在Python+FastAPI项目中使用SqlAlchemy操作数据的几种常见方式