python3 sqlalchemy2.0+ async_sessionmaker orm select where 查询对象转字典

<think>我们使用SQLAlchemy 2.0+的异步ORM，通过`async_sessionmaker`创建异步会话，然后执行查询并将结果转换为字典。 根据引用[3]中的示例，我们使用`select`和`filter`进行查询，然后处理结果。 有几种方法可以将ORM对象转换为字典： 1. 使用`mappings()`方法：在查询时使用，返回的是包含列名和值的字典形式。 2. 为模型定义`to_dict`方法：在模型类中添加一个方法，将自身转换为字典。 3. 使用工具函数：编写一个通用函数，将任何ORM对象转换为字典。 考虑到异步环境和SQLAlchemy 2.0，我们分别展示这些方法。 方法一：使用`mappings()`（推荐，因为不需要修改模型，且直接由SQLAlchemy处理） 方法二：在模型类中定义`to_dict`方法（可定制，但需要修改模型） 方法三：使用通用函数（不修改模型，但需要额外函数） 我们以方法一为主，因为引用[3]中使用了`select`和`execute`，我们可以直接使用`mappings()`来获取字典。 步骤： 1. 创建异步引擎和`async_sessionmaker`。 2. 使用`async_sessionmaker`创建会话。 3. 构建查询语句（使用`select`）。 4. 执行查询，并使用`mappings()`方法，这样返回的每一行都是`RowMapping`对象，可以转换为字典。 5. 将结果转换为字典列表。 示例代码： 假设我们有一个模型`User`，定义如下： ```python from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import Mapped, mapped_column class Base(DeclarativeBase): pass class User(Base): __tablename__ = "user_account" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(30)) fullname: Mapped[Optional[str]] ``` 使用`mappings()`进行查询并转换为字典： ```python from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from sqlalchemy import select # 创建异步引擎 engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname") # 创建异步会话工厂 async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async def get_users(): async with async_session() as session: # 使用mappings()方法，返回的结果每一行是一个字典结构 stmt = select(User).where(User.name == 'some_name') # 示例条件 result = await session.execute(stmt) # 将每一行转换为字典 # 注意：因为select(User)返回的是User对象，但使用mappings()后，每一行是一个包含User所有列的字典 # 但是，这里每一行实际上是一个RowMapping对象，它有一个._mapping属性，但也可以直接转换为dict # 方法1: 使用row._asdict()（但RowMapping没有_asdict方法，所以不用）; 方法2: 直接使用dict(row) # 因为RowMapping实现了__getitem__和keys()，所以可以用dict()转换 users_dict_list = [dict(row) for row in result.mappings()] return users_dict_list ``` 注意：`result.mappings()`返回的是可迭代的`RowMapping`对象，每个`RowMapping`对象代表一行，并且可以像字典一样访问。使用`dict(row)`即可转换为普通字典。 方法二：使用模型上的`to_dict`方法 在模型类中添加方法： ```python class User(Base): # ... 同上 def to_dict(self): return {c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs} ``` 然后查询并转换： ```python async def get_users(): async with async_session() as session: stmt = select(User).where(User.name == 'some_name') result = await session.execute(stmt) # 获取所有User对象 users = result.scalars().all() # 使用每个对象的to_dict方法 return [user.to_dict() for user in users] ``` 方法三：使用通用转换函数 ```python def orm_to_dict(obj): # 使用inspect获取映射的列属性 from sqlalchemy import inspect return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs} async def get_users(): async with async_session() as session: stmt = select(User).where(User.name == 'some_name') result = await session.execute(stmt) users = result.scalars().all() return [orm_to_dict(user) for user in users] ``` 总结： 推荐使用方法一（`mappings()`），因为它更直接，且不需要加载完整的ORM对象（实际上，它仍然会构建ORM对象吗？注意：在2.0中，`mappings()`方法返回的是包含列名和值的行，但不会构建ORM对象，因此性能可能更好？）。但是，根据SQLAlchemy文档，`mappings()`方法返回的是与列名对应的值，而不是构建ORM对象，所以效率较高。 注意：在方法一中，我们使用`select(User)`，然后使用`mappings()`，实际上它返回的是每一行作为一个包含User所有列的字典。但是，如果查询中包含多个实体，那么每个行字典会包含多个实体的键，键为属性名或给定的标签。因此，在只查询一个实体时，每个行字典的键就是该实体（User）的列名。 另外，如果查询中指定了特定的列（例如`select(User.name, User.fullname)`），那么字典中就只有这两个键。 因此，方法一更灵活，且不需要修改模型，推荐使用。 引用[4]中提到SQLAlchemy的文档混乱，但2.0有所改善，建议参考官方文档[^4]。 根据引用[2]，我们使用了`select`和`Session`（这里是异步的`AsyncSession`）[^2]。 根据引用[3]，我们使用了`filter`（在示例中为`where`）和`execute`[^3]。 因此，以上方法都是可行的。 最后，注意在异步环境中，所有操作都要在异步上下文中，并且使用`await`。 相关问题： 1. 在SQLAlchemy异步查询中，如何处理关联对象的加载并转换为字典？ 2. 使用`mappings()`方法时，如何给列取别名？ 3. 在转换字典时，如何处理日期时间等特殊类型的序列化？ 4. 如何排除某些字段（如密码）不被包含在返回的字典中？ 5. SQLAlchemy异步查询中，如何分页并转换为字典？</think>在 Python 3 中使用 SQLAlchemy 2.0+ 和 `async_sessionmaker` 将 ORM 查询结果转换为字典，主要有以下几种推荐方法： ### 方法 1：使用 `mappings()` 直接获取字典（推荐） 这是最高效且简洁的方式，直接通过 SQLAlchemy 的 `mappings()` 方法获取字典结果： ```python from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker async def query_users(session: AsyncSession) -> list[dict]: # 构建查询（带 WHERE 条件） stmt = select(User).where(User.is_active == True).limit(10) # 执行查询并返回字典结果 result = await session.execute(stmt) return [dict(row) for row in result.mappings()] # 使用示例 async with async_session() as session: users = await query_users(session) # users 格式: [{"id": 1, "name": "Alice", ...}, ...] ``` **优点**： - 无需修改模型定义 - 性能最佳（避免额外对象转换） - 直接使用列名作为字典键 [^1] - 支持所有 SQLAlchemy 查询特性（包括 JOIN、子查询等） ### 方法 2：为模型添加 `to_dict()` 方法 在 ORM 模型类中定义转换方法，适合需要自定义字段的场景： ```python from sqlalchemy import inspect from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] email: Mapped[str] def to_dict(self) -> dict: """将 ORM 对象转为字典""" return {c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs} # 查询使用 async def get_user(user_id: int, session: AsyncSession) -> dict: stmt = select(User).where(User.id == user_id) result = await session.execute(stmt) user = result.scalars().first() return user.to_dict() if user else {} ``` **优点**： - 可自定义输出（如排除敏感字段） - 支持复杂对象关系处理 - 代码可读性高 [^2] ### 方法 3：通用转换函数 创建独立于模型的转换工具函数： ```python from sqlalchemy import inspect def orm_to_dict(obj) -> dict: """通用 ORM 对象转字典函数""" return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs} # 在查询中使用 async def get_active_users(session: AsyncSession) -> list[dict]: stmt = select(User).where(User.is_active == True) result = await session.execute(stmt) users = result.scalars().all() return [orm_to_dict(user) for user in users] ``` ### 关键注意事项 1. **异步会话管理**： ```python # 正确使用 async_sessionmaker from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") async_session = async_sessionmaker(engine, expire_on_commit=False) ``` 2. **日期时间处理**： ```python # 自定义处理日期格式 def orm_to_dict(obj): return { **{c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs}, "created_at": obj.created_at.isoformat() # 转换为 ISO 格式 } ``` 3. **关系对象处理**： ```python # 使用 selectinload 预加载关系 from sqlalchemy.orm import selectinload stmt = select(User).options(selectinload(User.addresses)).where(...) result = await session.execute(stmt) user_dict = dict(result.mappings().first()) # 包含 addresses 关系 ``` ### 性能对比 | 方法 | 执行速度 | 内存占用 | 适用场景 | |--------------|----------|----------|------------------------| | `mappings()` | ★★★ | ★ | 简单查询，无需自定义 | | `to_dict()` | ★★ | ★★ | 需要字段控制或复杂处理 | | 通用函数 | ★★ | ★★ | 跨模型统一处理 | > 对于大多数场景，**推荐使用 `mappings()` 方法**，它直接利用 SQLAlchemy 的底层优化，避免了不必要的对象转换开销[^1][^4]。