一、SQLAlchemy 引入

安装

pip install sqlalchemy

痛点：在 Windows 环境下经常会遇到 Microsoft Visual C++ 14.0 is required 的编译错误，解决办法是先安装 或者直接使用预编译的二进制轮子。

项目配置类

import os
from dotenv import load_dotenv
from sqlalchemy.ext.asyncio import async_sessionmaker,AsyncSession,create_async_engine
load_dotenv
# 数据库 URL
ASYNC_DATABASE_URL = os.getenv
POOL_SIZE = int)
MAX_OVERFLOW = int)
POOL_RECYCLE = int)
POOL_TIMEOUT = int)
SQL_ECHO = os.getenv.lower == 'true'
# 创建异步引擎
async_engine = create_async_engine(
ASYNC_DATABASE_URL,echo=SQL_ECHO,pool_size=POOL_SIZE,max_overflow=MAX_OVERFLOW,pool_timeout=POOL_TIMEOUT。pool_recycle=POOL_RECYCLE,)
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
bind=async_engine,class_=AsyncSession,expire_on_commit=False,)
# 依赖项，用于获取数据库会话
async def get_db:
async with AsyncSessionLocal as session:
try这方面，yield session
await session.commit
except Exception:
await session.rollback
raise
finally:
await session.close

痛点：如果在 Docker 容器中运行，ASYNC_DATABASE_URL 常常忘记加上前缀 postgresql+asyncpg://导致 “No async driver found” 错误。确保 URL 与驱动匹配。

二、实体类定义

下面以使用者表为例，演示如何使用 SQLAlchemy 的声明式基类。

from datetime import datetime
from typing import Optional
from sqlalchemy import Index,Integer,String,Enum,DateTime
from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column
class Base:
pass
class User:
"""使用者信息表 ORM 模型"""
__tablename__ = 'user'
# 创建唯一索引，防止重复插入
__table_args__ = (
Index,Index,)
id这方面，Mapped = mapped_column(Integer,primary_key=True,autoincrement=True。comment="使用者 ID")
username: Mapped = mapped_column,unique=True,nullable=False,comment="使用者名")
password: Mapped = mapped_column,nullable=False,comment="密码")
nickname: Mapped] = mapped_column,comment="昵称")
从avatar来看，Mapped] = mapped_column,default='https://fastly.jsdelivr.net/npm/@vant/assets/cat.jpeg',comment="头像 URL")
再看gender，Mapped] = mapped_column,default='unknown',comment="性别")
bio这方面，Mapped] = mapped_column,default='这个人很懒，什么都没留下！',comment="个人简介")
phone这方面，Mapped] = mapped_column,unique=True,comment="手机号")
created_at: Mapped = mapped_column(DateTime,default=datetime.now,comment="创建时间")
updated_at: Mapped = mapped_column(DateTime,default=datetime.now,onupdate=datetime.now。comment="更新时间")
def __repr__:
return f""

痛点：初学者经常把 MergedColumn 写成 Merged_Column 或者忘记导入 MergedColumn导致 “name 'Mapped' is not defined” 报错。务必检查导入方法，

三、CRUD 操作

通用注意事项

• 所有写操作都必须在 await session.commit 之后才能持久化到数据库。
• 异常处理是关键：在事务块里捕获异常后要调用 await session.rollback否则后续会话可能处于不一致状态。说起来，
• Pain Point：很多人忘记在异步函数里使用 await db.refresh。导致返回的对象缺少自增主键或默认字段。

创建

两种常见方式这方面，

1. Add 方法：适用于单条或少量数据。
2. Insert + execute：适用于批量写入，提高性能。

# 示例：使用 add 添加单个使用者
async def create_user -> User:
hashed_password = get_hash_password # 假设已实现的密码哈希函数
user = User
db.add
再看try。
await db.commit
await db.refresh # 获取自增 ID 等默认值
return user
except Exception as e:
await db.rollback
raise e

查询

SQAlchemy 的查询链式写法与原生 SQL 十分相似，下面给出一个分页查询的完整示例。

from sqlalchemy import select
async def get_users:
stmt = (
select
.order_by)
.offset
.limit
)
result = await db.execute
# result.scalars 返回的是模型实例列表；若只想要字典，可使用 result.mappings.all
return result.scalars.all

Pain Point：If you forget .scalars。you will get a list of Row objects and accessing attributes like ``user.username`` will raise ``AttributeError``.

更新

SQAlchemy 提供两种更新方式：直接修改对象属性后提交，或使用表达式式 UPDATE。

# 方式一：对象属性更新
async def update_user_profile:
stmt = select.where
result = await db.execute
user_obj = result.scalar_one_or_none
if not user_obj:
raise ValueError
for key,value in data.items:
setattr
try的观点是，await db.commit
await db.refresh
return user_obj
except Exception as e:
await db.rollback
raise e
# 方式二：表达式 UPDATE
from sqlalchemy import update
async def bulk_update_user:
stmt = (
update
.where
.values(**payload.model_dump(exclude_unset=True,exclude_none=True))
)
result = await db.execute
await db.commit
return result.rowcount # 返回受影响行数

Pain Point：If you pass a Pydantic model directly into ``values`` without converting it to a dict ``)。SQLAlchemy will raise “Invalid column name” because it receives an object instead of key‑value pairs.

删除

from sqlalchemy import delete
async def delete_user -> int:
stmt = delete.where
result = await db.execute
if result.rowcount == 0:
raise ValueError
await db.commit
return result.rowcount

Pain Point：The common mistake is forgetting to call ``await db.commit`` after ``execute``,which makes deletion appear successful in logs but nothing changes in database.

四、事务与原子性

SQAlchemy 支持显式事务控制，通过 ``async with session.begin:`` 可以确保一组操作要么全部成功，要么全部回滚。老实说，这在涉及多表同步更新时尤为关键，例如积分转移场景。

from sqlalchemy import update,func
from sqlalchemy.ext.asyncio import AsyncSession
class PointsLog:
__tablename__='points_log'
id这方面，Merged=mapped_column)
from_user_id:Merged=mapped_column
to_user_id:Merged=mapped_column
再看points，Merged=mapped_column
created_at:Merged=mapped_column))
async def transfer_points(
说到db，AsyncSession,from_user_id:int,to_user_id:int,points:int) -> None:
async with db.begin:
# 扣除转出方积分并做充足性校验
stmt_from = (
update
.where(User.id == from_user_id,User.points>= points) # 防止负数积分
.values
)
res_from = await db.execute
if res_from.rowcount == 0:
raise ValueError
# 增加转入方积分
stmt_to = (
update
.where
.values
)
res_to = await db.execute
if res_to.rowcount == 0:
raise ValueError
# 写入日志表
log_entry = PointsLog(
from_user_id=from_user_id。to_user_id=to_user_id,points=points,created_at=func.now,)
db.add
# 调用示例
try的观点是，await transfer_points
print
except ValueError as e:
print
# 此时事务已经自动回滚，无需额外处理

Pain Point：If you use ``await db.commit`` inside ``async with session.begin:`` block manually and also let context manager commit automatically later on a successful exit – this leads to “Transaction already closed” errors. Let context manager handle commit/rollback exclusively.

五、常见错误与调试技巧 🎯

• Error: "QueuePool limit of size X overflow Y reached". ⚠️ 原因通常是连接未及时释放。方法的观点是，在每个请求结束后确保调用 ``await session.close`` 或者使用依赖注入的 ``get_db`` 自动关闭。
• Error: "InvalidRequestError: Table 'user' is already defined". ⚠️ 多次执行模型导入导致 Base 被重复注册。推荐把模型统一放在一个模块并只导入一次或者在测试环境中使用 ``Base.metadata.clear`` 重置元数据。
• Error: "TypeError: Column type Enum has no attribute 'python_type'" . ⚠️ 使用 SQLite 时 Enum 不被原生支持，需要手动指定 ``native_enum=False`` 或者改用字符串类型。
• TIPS: 开启 ``SQL_ECHO=True`` 可以实时看到生成的 SQL，在定位“为什么没有更新”时非常有帮助。其实，
• TIPS: 利用 VS Code / PyCharm 的断点调试。在执行 ``await session.execute`` 前后观察 ``session.new``,``session.dirty``,``session.deleted`` 的内容，可以快速定位状态不一致的问题。

六、学习路线推荐 📚

通过以上结构化的章节和针对性的痛点提醒。你可以一步步消除学习过程中的卡点，从基础开始熟练掌握 SQLAlchemy ORM 的主要技巧。