Python异步ORM ormar实战：基于Pydantic与SQLAlchemy Core的高效开发-深圳市維司達科技有限公司

1. 项目概述：一个为异步而生、兼顾灵活与简洁的Python ORM

如果你正在用FastAPI、Starlette或者任何基于asyncio的Python异步框架开发项目，并且厌倦了在SQLAlchemy的同步核心上套用各种异步适配器的别扭感，或者觉得像tortoise-orm这样的全异步ORM学习曲线又有点陡峭，那么ormar这个项目很可能就是你在找的那把趁手的“瑞士军刀”。我是在去年一个高并发数据采集项目的技术选型中深度接触并最终选用它的，经过几个线上项目的锤炼，它已经成了我异步后端开发的默认ORM选择。

简单来说，ormar是一个构建在SQLAlchemy core和databases（一个支持asyncio的数据库驱动包装库）之上的异步ORM。它的设计哲学非常明确：为异步应用提供一流的开发体验，同时保持与Pydantic模型的高度一致性，让数据验证、序列化和数据库操作无缝衔接。这意味着你定义的模型既是Pydantic的BaseModel（用于请求/响应验证），也是ORM的模型（用于数据库CRUD），一份代码，两种用途，极大地减少了样板代码和心智负担。

它的核心吸引力在于，既继承了SQLAlchemy强大的SQL表达能力和数据库兼容性（通过其core层），又通过databases库实现了真正的异步I/O。同时，它提供了类似Django ORM或SQLAlchemy的声明式语法，但API设计更加现代和Pythonic。对于从FastAPI生态过来的开发者尤其友好，因为你的Pydantic模型几乎可以“无损”地转化为ORM模型。

2. 核心设计哲学与架构拆解

2.1 为什么是“Pydantic-First”？

ormar最颠覆性的设计就是将ORM模型完全构建在Pydantic之上。在传统ORM如SQLAlchemy中，模型类主要用于定义数据库表结构，如果你需要序列化（比如返回JSON给前端）或反序列化（比如接收前端请求数据），通常需要另外定义Pydantic模型或使用其他序列化库，这就导致了模型定义的重复。

ormar解决了这个问题。当你定义一个ormar.Model时，它同时也是一个pydantic.BaseModel。这意味着：

自动验证：所有通过ORM模型赋值的数据都会经过Pydantic的严格类型和验证器检查。
无缝序列化：你可以直接使用.dict()或.json()方法将模型实例转换为字典或JSON字符串，非常适合FastAPI的响应模型。
单一数据源：从API接口接收到数据验证，再到数据库持久化，全程使用同一个模型类，保证了数据一致性，杜绝了因模型定义不一致导致的隐蔽bug。

这种设计特别适合前后端分离、API驱动的现代Web开发。你的业务逻辑层可以完全基于这些强类型的模型对象进行操作，安全性大大提升。

2.2 异步引擎的基石：SQLAlchemy Core + Databases

ormar没有重复造轮子去实现一个完整的SQL解析和执行引擎，而是巧妙地站在了两位“巨人”的肩膀上：

SQLAlchemy Core：负责生成SQL语句。SQLAlchemy Core提供了强大的、可组合的SQL表达式语言，ormar利用它来构建SELECT，INSERT，UPDATE，DELETE等语句。这保证了生成的SQL是高效且符合数据库方言的。你甚至可以通过ormar的底层接口接触到这些Select对象，进行更复杂的操作。
Databases：负责异步执行SQL。这是一个轻量级的库，为asyncpg，aiomysql，aiosqlite等异步数据库驱动提供了统一的异步接口。ormar通过databases执行SQLAlchemy Core生成的SQL语句，从而实现真正的非阻塞I/O。

这种架构带来了巨大优势：性能与控制的平衡。你既享受了ORM的便捷，又因为使用轻量的Core而非全功能的ORM（指SQLAlchemy ORM层）以及异步驱动，获得了接近原生异步驱动的性能。同时，由于依赖的是成熟的SQLAlchemy和databases，其数据库兼容性和稳定性非常可靠。

2.3 声明式语法与关系映射

ormar的模型定义语法非常清晰。它使用元类（ormar.ModelMeta）来配置表的元数据（如表名、数据库连接），而字段则直接作为类属性，用ormar提供的String，Integer，ForeignKey等来声明。这种声明式风格对开发者非常友好。

在关系处理上，它支持一对一、一对多、多对多等所有常见关系类型。定义关系的方式直观，通过ForeignKey指向目标模型，并在反向查询端使用ormar.ManyToMany或反向引用名称。查询关联数据时，它提供了类似select_related和prefetch_related的机制（通过load_all()方法或查询时的select_related参数）来避免N+1查询问题，这对于异步环境下的性能至关重要。

3. 从零开始：模型定义与数据库连接

3.1 初始化配置与模型定义实战

让我们从一个完整的例子开始。假设我们要构建一个简单的博客系统，有User和Post两个模型。

首先，你需要配置一个全局的元数据（Meta），它包含了数据库连接信息和表的基础命名规则。我习惯在一个单独的配置文件（如database.py）中做这件事：

# database.py import databases import sqlalchemy import ormar DATABASE_URL = "sqlite:///./test.db" # 也可以是 postgresql://, mysql:// 等 database = databases.Database(DATABASE_URL) metadata = sqlalchemy.MetaData() # 这个BaseMeta会被所有模型继承 class BaseMeta(ormar.ModelMeta): metadata = metadata database = database

接下来，定义User模型：

# models.py import ormar import pydantic from typing import Optional, List from .database import BaseMeta class User(ormar.Model): class Meta(BaseMeta): tablename = "users" id: int = ormar.Integer(primary_key=True, autoincrement=True) username: str = ormar.String(max_length=100, unique=True, nullable=False) email: str = ormar.String(max_length=255, unique=True, nullable=False) hashed_password: str = ormar.String(max_length=255, nullable=False) is_active: bool = ormar.Boolean(default=True) # 注意：密码等敏感字段不应直接返回，我们可以利用Pydantic的特性 # 通过配置 `ormar.Model` 的 `exclude` 或 `include` 来控制序列化字段

这里可以看到，字段定义就像写Pydantic模型一样自然。ormar.String等字段类型不仅定义了数据库列的类型，也内置了Pydantic的验证规则。

然后，定义与User关联的Post模型：

class Post(ormar.Model): class Meta(BaseMeta): tablename = "posts" id: int = ormar.Integer(primary_key=True, autoincrement=True) title: str = ormar.String(max_length=200, nullable=False) content: str = ormar.Text(nullable=False) published: bool = ormar.Boolean(default=False) # 定义外键关系 owner_id: int = ormar.ForeignKey(User, related_name="posts") # 这不是一个数据库字段，而是用于反向查询的关系属性 # `related_name="posts"` 允许我们通过 `user.posts` 获取该用户的所有文章

实操心得：关于related_name一定要显式地设置related_name，尤其是在模型关系复杂时。如果不设置，ormar会自动生成一个（如post_set），但这会使代码意图不清晰，且在模型类名更改时可能导致混淆。我习惯使用复数形式，如posts，comments，一目了然。

3.2 数据库连接与表创建

在应用启动时（例如FastAPI的lifespan事件处理器中），你需要连接数据库并创建表。

# main.py (FastAPI示例) from fastapi import FastAPI from contextlib import asynccontextmanager from .database import database, engine, metadata from . import models @asynccontextmanager async def lifespan(app: FastAPI): # 启动时连接数据库 await database.connect() # 创建所有表。在生产环境，更推荐使用Alembic进行迁移。 async with database: await database.run_sync(metadata.create_all, checkfirst=True) yield # 关闭时断开数据库连接 await database.disconnect() app = FastAPI(lifespan=lifespan) @app.get("/") async def root(): return {"message": "Hello World"}

注意事项：表创建与迁移metadata.create_all()在开发初期很方便，但绝不能用于生产环境。生产环境的数据库结构变更必须通过迁移工具（如Alembic）来管理。ormar与Alembic兼容良好，因为底层是标准的SQLAlchemyMetaData。你需要编写Alembic迁移脚本来处理ALTER TABLE等操作。

4. 核心CRUD操作详解

4.1 创建（Create）与事务处理

创建记录非常简单，直接使用模型的create方法或先实例化再保存。

# 方法1：使用create类方法（推荐，更简洁） new_user = await User.objects.create( username="johndoe", email="john@example.com", hashed_password="hashed_secret" ) # 方法2：实例化后保存 new_user = User( username="janedoe", email="jane@example.com", hashed_password="hashed_secret2" ) await new_user.save()

事务处理是数据库操作的关键。ormar通过database.transaction()上下文管理器来支持：

async with database.transaction(): user = await User.objects.create(username="alice", email="alice@example.com", hashed_password="...") # 如果这里创建Post失败，上面的User创建也会回滚 post = await Post.objects.create(title="Hello", content="World", owner=user) # 也可以使用模型实例的`save()`或`update()`，它们都会在事务内执行

踩坑记录：await是必须的！所有ormar的数据库操作方法都是异步的，必须使用await。忘记写await是一个常见错误，它不会立即报错，但会返回一个协程对象，导致后续代码逻辑混乱。我养成的习惯是，看到create，get，update，delete，all，filter等操作，手指下意识地先敲await。

4.2 查询（Read）：从简单到复杂

查询是ORM的核心。ormar提供了链式调用和类Django的查询API，非常直观。

1. 获取单个对象：

# 通过主键获取 user = await User.objects.get(id=1) # 通过其他唯一字段获取 user = await User.objects.get(username="johndoe") # get方法如果没找到会抛出`ormar.NoMatch`异常，通常需要捕获处理 from ormar.exceptions import NoMatch try: user = await User.objects.get(username="nonexistent") except NoMatch: user = None

2. 过滤与获取列表：

# 获取所有活跃用户 active_users = await User.objects.filter(is_active=True).all() # 多条件过滤 users = await User.objects.filter(is_active=True, username__icontains="john").all() # 查询操作符非常丰富： # __exact, __iexact, __contains, __icontains, __in, __gt, __gte, __lt, __lte, __startswith, __istartswith 等 posts = await Post.objects.filter(published=True, id__in=[1, 2, 3], title__icontains="tutorial").all()

3. 关联查询与数据预加载：

这是体现ORM价值的地方。避免N+1查询至关重要。

# 方式1：使用select_related加载外键关联的对象（针对ForeignKey，即多对一或一对一） post = await Post.objects.select_related("owner").get(id=1) print(post.title, post.owner.username) # 这里访问owner不会触发额外查询 # 方式2：使用load_all()在获取实例后加载所有关联（包括反向关系） user = await User.objects.get(id=1) await user.load_all() # 这会加载该用户所有的`posts`（通过related_name） for post in user.posts: print(post.title) # 方式3：在查询集上使用select_related（适用于一对多关系的“一”方查询） # 注意：对于反向的一对多关系，`select_related`不能直接用在查询集上，通常用`prefetch_related`或`load_all` # ormar的`prefetch_related`是通过`select_related`和后续处理模拟的，对于简单场景，在查询后调用`load_all()`更直接。

4. 排序、分页与数量统计：

# 排序 posts = await Post.objects.filter(published=True).order_by("-created_at").all() # 负号表示降序 # 分页 (limit/offset) page_size = 10 page_number = 2 posts = await Post.objects.filter(published=True).offset((page_number-1)*page_size).limit(page_size).all() # 数量统计 count = await Post.objects.filter(published=True).count() # 是否存在 exists = await Post.objects.filter(title="My Title").exists()

4.3 更新（Update）与删除（Delete）

更新操作可以直接在查询集上执行，也可以先获取对象再修改属性。

# 方式1：直接更新查询集（批量更新） await Post.objects.filter(published=False, created_at__lt=some_old_date).update(published=True) # 方式2：获取对象后更新（更常用，能触发Pydantic验证） post = await Post.objects.get(id=1) post.title = "Updated Title" post.content = "Updated content" await post.update() # 注意：这里更新的是所有字段。ormar也支持`update(_columns=["title"])`只更新特定列。 # 部分更新（PATCH风格） post = await Post.objects.get(id=1) # 假设我们从一个字典`data`中接收更新 data = {"title": "New Title"} for key, value in data.items(): if hasattr(post, key): setattr(post, key, value) await post.update()

删除操作类似：

# 删除单个对象 post = await Post.objects.get(id=1) await post.delete() # 批量删除 await Post.objects.filter(published=False, owner_id=some_user_id).delete()

重要提示：更新/删除的返回值update()和delete()方法在查询集上调用时，返回的是受影响的行数（整数）。在单个模型实例上调用update()返回None，调用delete()也返回None。这一点和Django ORM不同，需要注意。

5. 高级特性与性能优化

5.1 利用Pydantic高级特性

由于模型就是Pydantic模型，你可以充分利用Pydantic的所有功能：

自定义验证器：在字段上使用@pydantic.validator。
字段别名：通过ormar.String(alias="userName")定义，方便处理JSON字段名与Python属性名不一致的情况。
序列化控制：使用ormar.Model的Config类（继承自Pydantic）来设置exclude，include，exclude_none等，控制API响应中包含哪些字段。这是处理敏感信息（如密码哈希）的关键。

class User(ormar.Model): class Meta(BaseMeta): tablename = "users" # 数据库字段定义... hashed_password: str = ormar.String(max_length=255, nullable=False) class Config: # 当调用`.dict()`或`.json()`时，默认排除`hashed_password`字段 exclude = {"hashed_password"} # 或者使用属性getter进行更复杂的控制

嵌套模型的序列化：当查询包含select_related时，关联的对象也会被自动序列化为嵌套字典/JSON，完全符合API响应需求。

5.2 复杂查询与原生SQL逃生舱

虽然ORM能处理大部分查询，但总有需要复杂JOIN或窗口函数的时候。ormar提供了逃生舱口。

1. 使用queryset属性获取底层SQLAlchemy查询对象：

from sqlalchemy import func # 获取ormar的查询集 qs = Post.objects.filter(published=True) # 获取底层的SQLAlchemy Select对象 sa_query = qs.queryset # 可以对其进行修改，例如添加GROUP BY, HAVING等 sa_query = sa_query.group_by(Post.owner_id).having(func.count(Post.id) > 5) # 通过database执行这个修改后的查询 results = await database.fetch_all(sa_query) # results是记录列表，不是ormar模型实例

2. 直接执行原生SQL：

对于极其复杂的查询，直接使用databases执行原生SQL是最直接的。

query = "SELECT username, COUNT(*) as post_count FROM users u JOIN posts p ON u.id = p.owner_id WHERE p.published = :published GROUP BY u.id" values = {"published": True} rows = await database.fetch_all(query=query, values=values)

5.3 性能考量与最佳实践

明智地使用select_related和load_all：只加载当前业务逻辑需要的关联数据。过度加载会浪费内存和数据库资源。对于列表页，可能只需要外键对象的ID和名称；对于详情页，才需要加载所有关联详情。
警惕N+1查询：在循环中访问未加载的关联属性是性能杀手。务必在循环外部使用select_related或提前load_all。
索引是数据库的朋友：ormar帮你生成SQL，但索引需要你自己在数据库层面规划。对于频繁用于filter()，order_by()和关联查询的字段，务必考虑添加数据库索引。可以通过Alembic迁移来管理索引创建。
分页：对于可能返回大量数据的查询，必须使用.limit()和.offset()进行分页。一次性加载百万条数据会拖垮你的应用和数据库。
连接池：确保你的异步数据库驱动（如asyncpg）配置了合适的连接池大小，以应对高并发。

6. 常见问题与排查实录

6.1 模型定义与迁移问题

问题：修改模型字段（如增加字段、修改类型）后，运行应用时报错，提示表结构不匹配。
排查：metadata.create_all()只在表不存在时创建。修改已存在的表需要使用数据库迁移工具。立即停止使用create_all进行结构变更，转而使用Alembic。
解决：
1. 初始化Alembic：alembic init alembic
2. 修改alembic/env.py，将target_metadata设置为你的metadata（即from yourapp.database import metadata）。
3. 生成迁移脚本：alembic revision --autogenerate -m "Add new field"
4. 审查生成的脚本（alembic/versions/下的文件），确保无误。
5. 应用迁移：alembic upgrade head

6.2 查询结果为空或异常

问题：await Model.objects.get(...)抛出NoMatch异常，但你觉得数据应该存在。
排查：
1. 检查过滤条件：确认字段名和操作符（__exact，__contains）是否正确。字符串匹配是大小写敏感的，除非使用__iexact或__icontains。
2. 检查数据库连接和事务：确认你的查询是否在一个未提交的事务内？其他连接可能还看不到这些数据。
3. 打印生成SQL（调试利器）：
```
qs = User.objects.filter(username="test") print(qs.queryset.compile(compile_kwargs={"literal_binds": True})) # 这会打印出带参数的SQL，可以复制到数据库客户端执行，看是否返回预期结果。
```
解决：根据SQL调试结果修正查询条件或检查数据库实际数据。

6.3 循环导入（Circular Imports）

问题：在定义有相互外键引用的模型时（如User和Post），可能会遇到因模型类尚未完全定义而导致的导入错误。
解决：使用字符串形式的模型引用。
```
# 在Post模型中 owner_id: int = ormar.ForeignKey("User", related_name="posts") # 使用字符串 "User"
```
同样，如果User模型需要反向引用Post模型中的某个字段类型，也可能需要延迟导入或使用ForwardRef（Pydantic特性）。ormar很好地集成了Pydantic的ForwardRef来处理这种循环依赖。

6.4 异步上下文管理

问题：在非异步环境（如普通的同步脚本）中调用await，或者在异步函数外使用ormar的异步方法。
解决：确保你的整个调用链是异步的。对于脚本，可以使用asyncio.run()来启动异步主函数。
```
import asyncio async def main(): await do_something_with_ormar() if __name__ == "__main__": asyncio.run(main())
```
在Web框架（如FastAPI）中，路由函数本身就是异步的，所以没有问题。

6.5 序列化时的递归错误

问题：两个模型互相引用（例如User有posts，Post有owner），当序列化一个User并加载了所有posts，而每个post又试图序列化其owner时，会导致无限递归。
解决：这是API设计层面的问题。你需要决定序列化的深度。Pydantic提供了exclude，include等工具。通常的实践是：
1. 在列表接口中，只序列化核心字段，关联对象只提供ID或简单信息。
2. 在详情接口中，才进行深层序列化。
3. 使用Pydantic的exclude或自定义响应模型来精确控制输出。例如，为User定义一个不包含posts的Pydantic模型用于某些接口。

经过几个项目的深度使用，我个人体会是，ormar在异步Python生态中找到了一个非常舒适的平衡点。它没有SQLAlchemy ORM那么重，但提供了足够强大的功能；它比一些更简单的异步ORM更严谨，得益于Pydantic的加持。对于FastAPI项目而言，其“Pydantic-First”的设计简直是天作之合，能让你从重复的模型定义和序列化工作中彻底解放出来，更专注于业务逻辑本身。当然，它也不是银弹，在处理极其复杂的查询或需要高度定制化SQL时，你还是需要了解其底层的SQLAlchemy Core和databases库，或者直接使用原生SQL。但就覆盖日常90%的数据库操作而言，ormar的表现堪称优秀。