简介：资源是一份ORMER开发实战的案例代码合集，面向后端开发者与初中级全栈工程师，帮助读者用对象映射方式替代手写SQL，在Web项目（如博客、电商、用户系统）中快速完成数据模型定义、增删改查及一对多关联管理。RAR压缩包共94个文件、约8.1MB，以Python源码和pyc编译文件为主体，配有多份CSV样例数据、npy分析结果、pth模型权重、pdf说明文档、ipynb交互式笔记与txt运行日志，覆盖数据准备、模型构建、实验运行到结果评估的完整链路，已有271人浏览学习。其中既有基础CRUD与关系映射的完整示例，也包含Autoformer自动化模块及多种模型实现，读者可参考其目录组织、模块划分、参数配置与调用方式，在真实项目中灵活扩展数据访问层。
1. ormer 到底是什么：一个让异步服务不再被数据库拖垮的 ORM
做异步服务最怕的不是并发上不去，而是数据库操作把事件循环卡死。ormer 是 Python 异步生态里一个轻量级 ORM，核心基于 SQLAlchemy Core 做 SQL 构建，但不像 SQLAlchemy ORM 那样依赖同步 Session，而是自己实现了一套贴近 Django 风格的 Query 构建器，配合 databases 库统一处理 PostgreSQL、MySQL、SQLite 的连接。它解决的是真实场景里的一个痛点：在 FastAPI、异步爬虫、异步任务队列里写数据层时，不用再把查询丢进线程池，可以一路 async/await 到底。适合那些已经受够了「同步 ORM 在 async 接口里偷偷阻塞事件循环」的开发者，也适合想从零搭一个干净数据层的小团队。这篇实战笔记就围绕 ormer 的模型定义、增删改查、查询构建和事务处理展开，把能直接抄的代码和踩过的坑一起给你。
2. ormer 的模型与连接：先从一张表开始
2.1 连接配置：三类数据库的差异点
ormer 本身不管理连接池，它把连接层交给 databases 库来做。所以第一步不是配置 ormer，而是先创建一个 databases 的 Database 实例，再把它挂到 ormer 上。
import ormer import databases # 本地开发用 SQLite，最省事 database = databases.Database("sqlite:///./dev.db") # 接入 MySQL 时改成这样，注意驱动要装对 # database = databases.Database("mysql://root:password@127.0.0.1:3306/app_db") # 接入 PostgreSQL 时改成这样 # database = databases.Database("postgresql://user:password@127.0.0.1:5432/app_db") ormer.database = database # 在应用启动阶段连接，关闭阶段断开 async def startup(): await database.connect() async def shutdown(): await database.disconnect()这里最值得留意的是连接串的驱动选择。SQLite 不需要额外驱动，MySQL 默认走 aiomysql，PostgreSQL 默认走 asyncpg，如果生产环境用 PostgreSQL，建议显式装上 asyncpg 并确认版本兼容。连接池参数也由 databases 控制，比如min_size、max_size，在构造 Database 时直接传进去，ormer 不干预。
我一般会把 Database 实例单独放在一个db.py里，而不是散落在业务代码中，这样 ormer 和后续的迁移脚本、测试替换数据库都能共用同一份配置。另一个容易忽略的点是：ormer 的ormer.database是全局的，一旦赋值后所有模型共享这一个连接，如果项目里需要连接多个库，常见做法是手写一个多数据库分发层，或者直接用原生 SQL 操作另一个库。
2.2 定义第一个模型：字段类型与主键的默认行为
ormer 的模型定义方式和 Django ORM 很像，用类型注解声明字段，用 Field 控制主键、默认值、索引等属性。
from datetime import datetime, timezone import ormer from ormer import Model, Field class Album(Model): id: int = Field(primary_key=True) title: str = Field(max_length=200) release_date: datetime = Field(default=datetime.now(timezone.utc)) rating: float = Field(default=0.0, index=True) class Meta: tablename = "album"每个字段对应数据库里的一列。id用了Field(primary_key=True)，ormer 会把它当成自增主键处理；title限定了最大长度，在 PostgreSQL 和 MySQL 里会映射成 varchar(200)；rating加了索引，后续按评分排序或过滤时会快很多。release_date的 default 是一个可调用对象，建模时不要写成default=datetime.now(...)这种已经执行过的值，否则每个新对象都会共用首次导入时的时间。
Meta 里的 tablename 如果不写，ormer 默认会用类名转小写。这个默认行为在生成表名时很容易踩坑：类名为AlbumOrder时，有的版本生成albumorder，有的生成album_order。我习惯所有模型都显式指定 tablename，避免不同环境之间表名不一致。
2.3 建表与迁移：ormer 不自带 migrate
ormer 模型定义好之后，并不会在 import 阶段自动建表。SQLite 场景下可以直接用Model.objects.create触发建表，但生产环境最好用统一方式管理表结构。
import sqlalchemy # 从 ormer 内置的 metadata 拿到所有模型定义 metadata = ormer.Model.metadata # 方式一：开发环境直接建表 engine = sqlalchemy.create_engine("sqlite:///./dev.db") metadata.create_all(engine)这里metadata.create_all(engine)只会创建不存在的表，不会修改已有表。一旦某张表已经在库里，后期给模型加字段，create_all 不会自动补列。
生产环境我推荐用 Alembic 做迁移。做法是：把 ormer 的 metadata 暴露给 Alembic 的 target_metadata，然后按正常的 autogenerate 流程生成迁移脚本。需要注意，Alembic 默认会把全部表都纳入管理，如果项目里同时存在非 ormer 管理的表，要在 env.py 里配置 include_object 过滤。
3. 增删改查的四个常用写法
3.1 插入：单条 create 与批量写入的取舍
ormer 的单条插入很直观，调用objects.create并带上模型字段值即可。
album = await Album.objects.create( title="夜航西飞", release_date=datetime.now(timezone.utc), rating=8.7, ) print(album.id) # 返回自增主键create 方法返回的是完整模型实例，主键、默认值都已经回填。这一点在写接口时很省心，创建完可以直接把带 id 的对象序列化返回给前端。
但批量插入就不太适合用循环 + create。每条 create 独立发一条 INSERT，几百条数据能让数据库连接忙死。我在实际项目里通常分两种情况：数据量在几十条量级，用 asyncio.gather 并发 create；数据量上百甚至上千条，我宁愿直接走原生 SQL 多行插入。
import databases rows = [ {"title": f"专辑-{i}", "release_date": datetime.now(timezone.utc), "rating": float(i % 5)} for i in range(100) ] # 手写多行 INSERT，避免逐条 create query = """ INSERT INTO album (title, release_date, rating) VALUES (:title, :release_date, :rating) """ await database.execute_many(query, rows)execute_many是 databases 库提供的方法，一条语句批量执行，性能比循环 create 高一个量级。代价是绕过 ormer 的模型层，拿不到回填的自增 id，需要再查一次或提前生成业务主键。这个取舍要看业务是否需要立即拿到每个新记录的 id。
3.2 查询：get/all/filter 的返回类型
查询是 ormer 里最常用的能力，先说三个最基础的方法。
# 取一条，查不到会抛 DoesNotExist 异常 album = await Album.objects.get(id=1) # 取全部，返回列表 albums = await Album.objects.all() # 条件过滤 recent_albums = await Album.objects.filter(rating__gte=8.0).all()get用于按唯一条件取单条，在 FastAPI 里写GET /albums/{id}接口最合适。查不到时抛出的异常类型是 ormer 自己的DoesNotExist，接口层要对它做统一捕获，否则会直接 500。
all()返回的是列表，不是懒加载的 queryset。这一点和 Django 不同，ormer 的查询在 await 时就真正执行了，所以写分页时一定要先 limit/offset 再 all()，绝不要把全表捞回来再在 Python 里切片。
filter的条件字段用双下划线区分操作符，rating__gte=8.0表示 rating >= 8.0。常用的还有title__icontains做模糊匹配、id__in做 IN 查询。刚开始用 ormer 的人容易下意识写filter(rating >= 8.0)，这是语法错误，ormer 不支持这种运算符重载式的写法。
3.3 更新与删除：影响行数怎么判断
更新和删除在 ormer 里都区分「按条件更新」和「按实例更新」两种用法。
# 用法一：按条件更新 updated_count = await Album.objects.filter(id=1).update(rating=9.0) # 用法二：先取实例再改，最后 save album = await Album.objects.get(id=1) album.rating = 9.0 await album.save() # 用法三：删除 deleted_count = await Album.objects.filter(id=1).delete()按条件更新的写法最常用，返回的是影响行数。如果传入的 id 不存在，返回 0，不会抛异常。这个特性在业务里要特别注意：更新接口如果传了不存在的 id，客户端不该收到 200，应该根据updated_count == 0返回 404。
先取实例再 save 的方式适合需要修改多个字段并触发其他逻辑的场景，比如修改后要比较旧值。但多了一步查询，性能上有损耗，高频更新路径不建议这么写。删除同理，按条件删除返回影响行数，判断是否存在要依赖返回值，而不是依赖异常。
3.4 事务：谁该包在 async with 里
ormer 的事务直接复用 databases 的 transaction API，把写操作包进async with即可。
from databases import Database async def transfer(album_id: int, target_rating: float): async with database.transaction(): album = await Album.objects.get(id=album_id) await Album.objects.filter(id=album_id).update(rating=target_rating) # 这里如果抛异常，上面的 update 会自动回滚事务的粒度我有一条经验：只包必要的写操作，不在事务里做耗时计算，也不在事务里调外部 HTTP 接口。事务持有连接锁的时间越长，并发高峰时死锁和连接池耗尽的概率越大。另外，事务内跨库操作是不行的，同一个事务只作用于同一个连接，如果第二个库不在事务范围内，出现跨库数据不一致时没有后悔药可吃。
4. 查询构建进阶：过滤、排序、分页、聚合
4.1 条件组合与排序分页的写法
真实业务里很少有单一条件的查询，组合条件、排序、分页几乎每个接口都会遇到。ormer 的条件组合方式我拆成三种：链式 filter、or 条件、排序分页。
# 链式过滤：多个 filter 是 AND 关系 albums = await ( Album.objects .filter(rating__gte=8.0) .filter(title__icontains="夜") .order_by("-release_date") .limit(20) .offset(0) .all() )链式 filter 每调一次就叠加一个 AND 条件。order_by("-release_date")表示按时间倒序，字段名前的负号是倒序标记，不写负号默认升序。limit(20)和offset(0)组成经典分页。
ORM 的分页还有一个容易忽略的坑：count()的统计要和分页条件保持一致。我见过不少人在filter(...).all()之后才发现要算总数，又单独写一个不带分页的 count 查询，两个查询条件没有复用，结果总数对不上。正确做法是先把条件链保存成变量，再分别调用 count 和分页。
base_query = Album.objects.filter(rating__gte=8.0) total = await base_query.count() albums = await base_query.order_by("-release_date").limit(20).offset(0).all()4.2 聚合查询：分组统计的实现
聚合不是 ormer 的主打场景，但接口里做榜单、做统计报表时绕不开。ormer 的模型层对聚合支持比较基础，我通常用两种方式混着写。
# 方式一：用 ormer 的 QuerySet 完成简单计数 count_by_rating = await Album.objects.filter(rating__gte=6.0).count() # 方式二：执行原生聚合 SQL query = """ SELECT rating, COUNT(*) AS cnt FROM album GROUP BY rating ORDER BY cnt DESC """ rows = await database.fetch_all(query)方式一只能处理单个表的行数统计，能力边界很明显。方式二则能处理 GROUP BY、HAVING、SUM、AVG 这类复杂聚合。在 ormer 项目里写原生 SQL 不丢人，聚合查询用 SQL 表达反而更清晰。
需要说明的是，fetch_all返回的是 Record 对象列表，取值用row["cnt"]或row.cnt都行。如果聚合结果是单个值，用fetch_val更直接。
4.3 关联数据怎么拿：避免 N+1 的两种做法
模型关联是各种 ORM 宣传的重点，但 ormer 的关联能力相对薄，没有完整的 select_related，盲目链式访问关联属性很容易出现 N+1 查询。背后的现象是：先查出一批专辑，再在循环里逐个查歌手或订单，查询次数变成了 1+N。我在项目里一般用两种方式规避。
第一种是手动 JOIN + 一次查询，适合固定维度的关联数据。
query = """ SELECT a.*, s.name AS singer_name FROM album a LEFT JOIN singer s ON a.singer_id = s.id WHERE a.rating >= :min_rating """ rows = await database.fetch_all(query, {"min_rating": 8.0})第二种是拆成两次查询，用id__in做第二轮批量加载。
albums = await Album.objects.filter(rating__gte=8.0).all() singer_ids = {album.singer_id for album in albums if album.singer_id} singers = await Singer.objects.filter(id__in=list(singer_ids)).all() singer_map = {s.id: s for s in singers}第二种方式比第一种更通用，适合关联维度会变化的场景。核心思路就是不要在循环里逐条查关联表，而是先收集 id 列表，再一次查出来组装成 dict。这个习惯能解决大部分关联查询的性能问题，比依赖 ORM 的魔法方法更可控。
5. ormer 实战避坑：连接池、时区与事务的踩坑记录
5.1 现象：连接被多个协程复用，报错 "connection is closed"
在测试环境并发调接口时，某个请求偶尔报连接已关闭，错误信息指向底层驱动，但代码本身看起来没问题。
原因：ormer 使用的 databases 连接池有上限，默认情况下连接被一个事务或查询用完后归还，但归还时序在极端并发下不可控。当服务里同时存在 fastapi 的 BackgroundTasks、定时任务、WebSocket 长连接时，多个协程可能拿到同一个连接实例，其中一个协程关闭了连接，另一个协程继续用它执行查询，就会触发这个报错。
解决：不要在业务代码里手动调用disconnect()，连接生命周期只归应用启动/关闭管理；给 Database 显式设置合理的max_size，比如原来默认 10 改成 20 或更大；长事务里绝对不能混入 await 外部请求，否则连接持有时间过长，池子很快空掉。
5.2 现象：存进去的时间，查出来少了 8 小时
PostgreSQL 场景下，用datetime.now(timezone.utc)插入数据，查询返回的却是不带时区信息的 naive datetime，和本地时间一对比，凭空差了几小时。
原因：数据库列类型是 timestamp without time zone，驱动在取出 datetime 时丢掉了时区标记，Python 这边拿到的是 naive datetime，后端再序列化成 ISO 字符串时被误当成 UTC，前端显示就偏了。
解决：建表时把时间列用原生 SQL 改成 timestamptz；或者在模型层统一约定——所有时间字段读写都走 UTC naive，展示时由序列化层手动转本地时区。我倾向后者，因为改列类型要动迁移，而且 timestamptz 在某些数据库驱动下仍需额外处理。重要的是全项目统一，别一半字段用带时区时间，一半用 naive 时间。
5.3 现象：事务内部混了同步代码，接口直接卡死
事务里调用了一个同步的第三方 SDK，这个 SDK 内部用了线程锁等待，事务所在的协程被阻塞，连接一直不释放，整个接口响应时间从几十毫秒涨到几十秒。
原因：async with database.transaction()会占用一个连接直到事务结束，在这个期间如果执行阻塞式 IO，事件循环被卡住，事务自然无法提交或回滚。本质是把事务连接当成了普通工作线程来用。
解决：事务代码块里只放数据库操作；需要调外部服务时，先从数据库把数据查出来，退出事务后再调外部服务，最后用单独事务更新状态；实在绕不开，就把阻塞调用丢给asyncio.to_thread执行，但绝不能让它出现在事务中间，否则连接池照样被拖垮。
5.4 现象：SQLite 下 JSON 字段整体变成字符串
开发环境用 SQLite 调试时，模型里定义 JSON 字段，存入 dict 正常，查出来却是一整个字符串，需要 json.loads 才能用，但同一套代码在 PostgreSQL 下没问题。
原因：SQLite 没有原生 JSON 类型，底层存储用的是 TEXT，ormer 和 databases 不会自动为 SQLite 做 JSON 编解码，写入时 dict 被 str() 转成了字符串，读取时自然原样返回。
解决：开发环境能忍就先忍，毕竟只是调试；需要严格模拟线上行为时，给字段加一个自定义类型注解，或者在查询后统一走json.loads。另一个更省事的方案是：所有环境都用一个统一的 JSON 处理工具函数包一层读写，代码里不直接暴露 JSON 字段的原始值。这样环境差异只影响工具函数内部，业务层不用感知。
5.5 现象：表结构加了新列，ormer 查询还是报字段不存在
改了模型，加了字段，也确认数据库表里已经通过迁移加上对应列，但查询时 ormer 仍然报 AttributeError，说这个字段不存在。
原因：ormer 的模型类在首次导入后，字段缓存不会自动感知数据库表结构的变化。更常见的是，项目用了热重载调试，模型类被重新加载的时候，旧类对象还残留在模块引用里，新字段没生效。
解决：改完模型先重启服务进程，不要依赖热重载；如果确认重启无效，检查是否在模型定义外还用setattr动态加过字段，这类动态加的字段 ormer 的 metadata 不一定认；最彻底的做法是删掉相关__pycache__再重启，避免旧字节码里的模型定义被复用。
6. 进阶：把 ormer 封装成仓库层
6.1 一个可复用的 Repository 基类
项目里几十张表，每张表都写一遍 get/create/update/delete 会非常啰嗦。我会抽一个 Repository 基类，把 ormer 的模型和通用的增删改查绑在一起。
from typing import TypeVar, Generic import ormer from ormer import Model T = TypeVar("T", bound=Model) class BaseRepository(Generic[T]): model: type[T] async def get(self, pk: int) -> T: return await self.model.objects.get(id=pk) async def create(self, **kwargs) -> T: return await self.model.objects.create(**kwargs) async def update(self, pk: int, **kwargs) -> bool: affected = await self.model.objects.filter(id=pk).update(**kwargs) return affected > 0 async def delete(self, pk: int) -> bool: affected = await self.model.objects.filter(id=pk).delete() return affected > 0 async def list_page(self, page: int, page_size: int, **filters): query = self.model.objects.filter(**filters) total = await query.count() items = await query.limit(page_size).offset((page - 1) * page_size).all() return items, total这个基类把所有调用方统一成一套接口。get负责按主键取单条；create直接透传字段；update和delete都返回 bool，避免上层到处比较影响行数；list_page把分页和总数封装在一起，上层接口不需要关心 ormer 的分页语法。每个具体表建一个子类，指定 model 即可：
class AlbumRepository(BaseRepository[Album]): model = Album为什么要做这层封装？核心原因不是省代码，而是把 ormer 的 API 隔离在数据层内部。将来哪天要换掉 ormer、或者某张表要走缓存，上层业务代码不需要跟着改。这个封装让我在项目里少改了很多冤枉代码。
6.2 在接口层注入仓库对象
FastAPI 里可以把仓库对象直接挂到依赖注入上，接口函数不直接碰 ormer。
from fastapi import FastAPI, Depends app = FastAPI() repo = AlbumRepository() @app.get("/albums/{album_id}") async def get_album(album_id: int, r: AlbumRepository = Depends(lambda: repo)): album = await r.get(album_id) return {"id": album.id, "title": album.title}依赖注入的好处是测试时可以替换仓库实例。比如单测里用一个内存版仓库，或者给仓库加缓存、加埋点拦截，业务函数完全无感。我这边的做法是：接口层永远不直接出现 ormer 的查询语句，所有数据访问都走 Repository。切成这样之后，新增一张表平均只需写 5 行子类代码，查询逻辑统一在基类维护。
这个习惯是从一次大改中攒下来的教训：当初某个模拟项目X里所有接口直接调 ormer，后期要给所有热点查询加 Redis 缓存，翻了几十个接口才改完，从那以后新项目一律先建 Repository 层再写业务。希望这个习惯和这篇 ormer 实战笔记一起，能帮你少走一段弯路。
本文还有配套的精品资源，点击获取
