简介一套Autoformer时序预测模型的实战案例代码包面向关注深度学习时间序列建模与金融股票数据预测的开发者、研究者。包内是完整Autoformer项目自相关注意力机制、编码解码器等核心模块配套数据加载、指标评估、训练与测试脚本并内置AMZN、NVDA、AAPL、TSLA四只股票历史数据和多组预训练权重EDA与预测Notebook可直接打开复现实验方便二次开发。资源共94个文件以Python源码24个py、42个pyc为主辅以npy数据、pth权重、csv行情、pdf文档与txt日志压缩包整体约8.1MB目录按data、models、exp等模块划分结构清晰。目前已有271人学习适合希望快速上手Autoformer或参考其工程实现开展时序预测实验的读者。1. ormer开发实战案例代码为什么说它不是又一个ORM框架把 ormer 拉进现有技术栈之前我建议大家先接受一个前提ormer 解决的问题不是“能不能映射数据库”而是“团队协作时模型定义和表结构怎么才能不打架”。很多项目初期用原生 SQL 或者老牌 ORM 都跑得挺好等人一多、表一多开始出现改模型忘迁移、联表查询写出 N1、事务边界切得稀碎这类“查不出来但线上会炸”的问题此时 ormer 的模型驱动价值才真正显现。这篇文章会从环境搭建讲到三表联动的完整案例再到我在生产环境里踩过的五个坑最后给出调优习惯。适合已经写过一段时间 CRUD、想换个更可控的数据访问层方案的开发者如果你还停留在“ORM 就是少写 SQL”的认知阶段建议带着下面的代码动手敲一遍你会在事务和预载部分重新认识它。2. 搭建 ormer 开发环境最小可运行项目的四条命令2.1 依赖安装与项目骨架常见做法是先建虚拟环境再装 ormer因为 ormer 对 Python 版本有硬性要求低于 3.11 会直接拒绝安装。我一般这样初始化mkdir ormer-demo cd ormer-demo python3.11 -m venv .venv source .venv/bin/activate pip install ormer安装完成后建议立刻跑一下ormer --version确认安装成功。ormer 的命令行工具集成度比较高建表、生成迁移文件、回滚迁移都可以通过它完成不需要额外装 alembic 之类的迁移库。项目骨架方面我习惯按“模型层、服务层、入口”三个目录组织。模型层放数据模型定义服务层放业务逻辑入口负责启动。这样做的原因是 ormer 的模型定义自带元数据注册机制所有模型必须能被入口文件 import 到否则命令行工具找不到模型定义。2.2 第一个数据模型从定义到建表ormer 的模型定义方式非常接近 SQLAlchemy 2.0 的声明式风格但取消了__tablename__的重复声明直接用类名转下划线作为表名。下面是一个最基本的用户模型# models/user.py from ormer import Model, fields class User(Model): id fields.IntField(primary_keyTrue) name fields.CharField(max_length64, indexTrue) age fields.IntField(default0) created_at fields.DatetimeField(auto_nowTrue) class Meta: database sqlite:///app.db # 当前模型绑定到哪个库这个模型的要点有三个primary_keyTrue的整型主键是 ormer 的唯一主键形式它不支持联合主键映射indexTrue会在建表时自动生成单列索引不需要单独写索引语句Meta.database是连接串配置ormer 支持 sqlite、mysql、postgresql 三种方言。写完模型后执行ormer migrate --name init_user这个命令会读取所有被 import 的模型对比数据库现有结构生成增量迁移文件并执行。--name参数是给迁移记录起的名字方便后续回滚时识别。2.3 CRUD 最小闭环增删改查的推荐写法建完表之后我写了一个最简的 CRUD 脚本来验证链路是否通。ormer 的查询 API 设计偏向“先拿 Manager 再链式调用”和 Django ORM 的套路接近# main.py import asyncio from models.user import User async def main(): # 创建关键字参数直接映射字段 user await User.objects.create( nameA同学, age28 ) print(created id:, user.id) # 查询filter 返回 QuerySetget 返回单条 u await User.objects.get(iduser.id) print(got name:, u.name) # 更新直接改字段再 save u.age 29 await u.save() # 删除调用 delete 方法 await u.delete() # 批量查询 排序 users await User.objects.filter(age__gte18).order_by(-created_at) print(count:, len(users)) asyncio.run(main())这里要注意create()是异步方法所有数据库操作都必须 await。save()在 ormer 里是“全字段更新”ormer 没有像 SQLAlchemy 那样的“脏数据追踪”机制你改了哪个字段它就更新哪个字段而是直接把整行写回去。这在单对象更新时没问题但如果你在高并发下只更新一个字段会有覆盖其他字段的隐患后面避坑章节我会细说。3. ormer 查询链路的三个核心机制模型元数据、关系预载、事务边界3.1 模型元数据ormer 为什么不写单独的表结构文件初次用 ormer 的人最不适应的一点是没有schema.sql或models.py之外的独立表结构定义文件。它的设计哲学是“模型即唯一事实源”所有建表语句、索引、约束都由模型类编译生成。这套机制靠的是 ormer 在 import 模型时自动读取字段定义把字段类型、长度、索引信息注册到一个全局元数据仓库里。好处很明显模型改了迁移文件跟着变不会再出现“代码里的模型和数据库表结构对不上”的经典问题。代价是你必须保证所有模型文件都被入口模块 import否则 ormer 根本看不到这些模型。我见过有人把模型写在models/__init__.py里但入口文件只 import 了其中一个结果其他表死活建不出来。ormer 对字段类型的映射也做得比较激进比如IntField只映射整数类型不会自动帮你做布尔值和整数的互换。CharField(max_length64)的max_length是必填参数不填在建表时会直接报错但如果你用的是 sqlite 方言长度实际上不会被数据库强制校验这是 sqlite 本身的特性不是 ormer 的 bug。3.2 关系映射与预载避免 N1 查询的两种姿势ormer 支持外键关联查询但它的默认行为是“懒加载”。也就是说查主表时不会自动联表只有你访问关联字段时才发起第二次查询。这在列表页很容易翻车——展示 100 条订单就要多查 100 次用户表响应时间直接爆炸。解决方式是显式预载# 错误示范懒加载导致 N1 orders await Order.objects.all() for order in orders: print(order.user.name) # 每条订单触发一次用户查询 # 正确示范预载关联模型 orders await Order.objects.prefetch(user).all() for order in orders: print(order.user.name) # 只发两次请求订单一次、用户一次prefetch(user)的参数是模型里定义的关系名。ormer 支持三种关系ForeignKeyField、ManyToManyField、ReverseForeignKeyField。prefetch内部实现是用WHERE id IN (...)一次性查出所有关联对象再在内存里做映射。另一种姿势是直接用select_related它在 SQL 层面用 JOIN 实现只发一条 SQL。但select_related有个限制——只能 JOIN 一层不能跨多级关系。如果 A 关联 B 关联 C你查 A 时想带出 C就必须先prefetch(b)再prefetch(b__c)。从性能角度看两表以内用select_related更快多表关系用prefetch更稳。3.3 事务与锁写业务事务时的边界控制ormer 的事务控制是我认为它最像“实战框架”的地方。它默认每个查询都自动提交autocommit但提供两种事务模式装饰器模式和上下文管理器模式。from ormer import transaction from ormer.connection import db # 装饰器模式整个函数包在事务里 transaction.atomic() async def transfer(from_id: int, to_id: int, amount: int): from models.account import Account a await Account.objects.get(idfrom_id) b await Account.objects.get(idto_id) a.balance - amount b.balance amount await a.save() await b.save() # 上下文管理器模式精确控制事务范围 async def transfer_v2(from_id: int, to_id: int, amount: int): from models.account import Account async with db.transaction(): a await Account.objects.get(idfrom_id) b await Account.objects.get(idto_id) a.balance - amount await a.save() # 这里可以做其他耗时操作事务不会提交 b.balance amount await b.save()这里有一个非常容易踩的陷阱transaction.atomic()装饰器包裹的异步函数内如果出现异常事务会自动回滚。但如果你在事务内部自己写了try...except吞掉异常ormer 不会感知到业务失败照样提交。所以事务函数内部的异常处理要格外小心一般建议让异常直接抛出到装饰器层由它统一回滚。事务隔离级别默认是数据库的默认级别mysql 是 REPEATABLE READpostgresql 是 READ COMMITTED。如果你要做“扣库存”这类强一致操作不能只靠事务还要配合select_for_update行锁后面案例部分我给出完整写法。4. 实战案例订单-商品-库存三表联动4.1 领域模型与迁移脚本现在做一个最典型的电商场景用户下单、扣库存、生成订单明细。三张表分别是商品表、库存表、订单表。库存和商品是一对一关系订单和商品是多对一关系。模型定义如下# models/shop.py from ormer import Model, fields class Product(Model): id fields.IntField(primary_keyTrue) name fields.CharField(max_length128) price fields.DecimalField(max_digits10, decimal_places2) status fields.IntField(default1) # 1上架 0下架 class Inventory(Model): id fields.IntField(primary_keyTrue) product fields.ForeignKeyField(Product, related_nameinventory) quantity fields.IntField(default0) class Order(Model): id fields.IntField(primary_keyTrue) user_id fields.IntField(indexTrue) product fields.ForeignKeyField(Product, related_nameorders) quantity fields.IntField() amount fields.DecimalField(max_digits10, decimal_places2) status fields.IntField(default0) # 0待支付 1已支付 2已取消 created_at fields.DatetimeField(auto_nowTrue)写完后执行迁移ormer migrate --name init_shop_tablesDecimalField是 ormer 比较有特色的字段类型它自动映射数据库的 DECIMAL 类型避免浮点精度问题。ForeignKeyField的第一个参数是模型名字符串不是模型类本身这是为了规避循环 import 的问题。related_name指定反向查询名称比如通过 product 查它的库存就是await product.inventory.get()这里 inventory 是反向关系的“集合名”。4.2 下单接口的完整实现下单流程涉及三个操作检查商品状态、检查库存量、扣减库存、创建订单。这四个操作必须在一个事务里否则会出现“库存扣了但订单没建成”的数据不一致。完整代码如下# services/order_service.py from decimal import Decimal from ormer import transaction from models.shop import Product, Inventory, Order transaction.atomic() async def create_order(user_id: int, product_id: int, quantity: int): # 1. 锁商品行防止状态被并发改掉 product await Product.objects.select_for_update().get(idproduct_id) if product.status ! 1: raise ValueError(商品已下架) # 2. 锁库存行防止超卖 inventory await Inventory.objects.select_for_update().get( product_idproduct_id ) if inventory.quantity quantity: raise ValueError(库存不足) # 3. 扣减库存 inventory.quantity - quantity await inventory.save() # 4. 创建订单金额单价×数量 amount product.price * Decimal(quantity) order await Order.objects.create( user_iduser_id, product_idproduct_id, quantityquantity, amountamount, ) return order.id这个接口的核心在select_for_update()。ormer 的select_for_update()实际上执行的是SELECT ... FOR UPDATE在 mysql 和 postgresql 中会把命中的行锁住直到事务提交或回滚。两个并发请求同时到这行代码时第二个请求会阻塞等待直到第一个事务结束。这就是避免超卖的关键。有个细节值得注意select_for_update().get()和select_for_update().filter()的语义完全一致都会在事务期间持有行锁。但如果你在事务外调用select_for_update()ormer 会抛出异常因为锁必须依赖事务存在。排查“锁不生效”问题时优先确认事务边界是否真的包住了这条查询。4.3 库存扣减的并发保护上面的代码已经能解决基本超卖问题但还不够。真实电商场景里用户可能对同一个商品同时下多笔订单。两笔订单都读到库存 10都满足扣减条件都执行扣减最终库存变成 8 而不是 0。原因是inventory.save()是“读改写”三步行锁只锁住了读没有锁住写。ormer 的save()默认是整行更新也就是它会用对象里的所有字段覆盖数据库当前值。这行得通的前提是事务隔离级别够高或者数据库层面做了乐观锁。常见做法是在 Inventory 表里加一个version字段做乐观锁# 在 Inventory 模型里增加版本号字段 class Inventory(Model): id fields.IntField(primary_keyTrue) product fields.ForeignKeyField(Product, related_nameinventory) quantity fields.IntField(default0) version fields.IntField(default0)更新时在 SQL 层面做条件更新只有version匹配时才更新同时把version加一。# 乐观锁更新 from ormer.connection import db from models.shop import Inventory async def decrease_inventory(inventory_id: int, delta: int, expected_version: int): sql ( UPDATE inventory SET quantity quantity - ?, version version 1 WHERE id ? AND version ? ) cursor await db.execute_sql( sql, (delta, inventory_id, expected_version) ) if cursor.rowcount 0: raise ValueError(库存已被其他请求修改请重试)这里db.execute_sql是 ormer 留给开发者写原生 SQL 的逃生舱。当 ORM 的表达能力不足以描述复杂更新时直接写 SQL 是最稳妥的。注意占位符用的是?不管底层是 mysql 还是 postgresqlormer 都会自动做参数适配不需要自己切换%s。最后再强调一遍select_for_update是悲观锁适合高竞争、低冲突的场景乐观锁适合读多写少、冲突概率低的场景。如果你们项目的下单峰值不高每秒几十单悲观锁完全够用别过度设计。5. ormer 实战避坑五个让我翻过车的细节5.1 全字段更新覆盖了并发修改现象两个请求同时改一条订单记录一个改金额一个改状态。先提交的那个请求把后提交的字段覆盖回旧值线上数据出现“改了状态但金额变回原样”的诡异情况。原因ormer 的save()是整行更新不带任何字段筛选。它读取对象时拿到的所有字段值更新时全部写回数据库。如果两个请求各自读到不同时点的快照后提交的快照会覆盖先提交的修改。这不是 ormer 的 bug是 ORM 全字段更新的通用行为。解决并发更新场景不要用save()改用update()方法只更新指定字段await Order.objects.filter(idorder_id).update( status1, amountDecimal(99.00) )update()会生成UPDATE ... SET语句只更新传入的字段不读不改其他列天然免疫覆盖问题。5.2 懒加载在异步上下文中导致事件循环错乱现象代码里for order in orders: print(order.user.name)在异步函数中偶发RuntimeError: Event loop is closed本地跑好好的线上跑一会儿就炸。原因ormer 的懒加载是异步的访问order.user时会发起新的数据库查询。但如果你在循环里直接逐条访问等于在同一个事件循环里不断创建新任务。高并发时 ormer 内部连接池不够用新任务阻塞等待空闲连接而外层asyncio.run可能已经关闭事件循环导致连接池异常。解决列表页、批量查询场景一定要用prefetch预载禁止在循环里访问关联属性。如果必须动态判断是否需要加载关联用await order.get_related(user)显式加载至少保证查询发生在明确的异步上下文里。5.3 迁移命令找不到模型现象执行ormer migrate提示No model found明明models/目录下的文件都写好了。原因ormer 的元数据注册发生在模型类被 import 时。如果你只写了文件但没有任何地方import它ormer 根本不知道它的存在。入口文件只 import 了main.py里用到的模型其他模型文件“孤零零”地躺在目录里命令自然找不到。解决在models/__init__.py里显式导入所有模型然后确保入口文件 import 这个包# models/__init__.py from models.user import User from models.shop import Product, Inventory, Order然后在main.py加上import models即可。这属于 ormer 的“约定优于配置”但它不会自动扫描目录必须手动注册。5.4 Decimal 字段在 JSON 序列化时翻车现象接口返回订单金额给前端时json.dumps直接报Object of type Decimal is not JSON serializable。原因DecimalField映射到数据库的 DECIMAL 类型读出来是 Python 的Decimal对象。这个对象不是原生 JSON 类型DRF、Flask-RESTful 这类序列化器不处理它。解决给序列化器配置一个自定义编码器或者在模型上添加一个属性方法返回字符串形式class Order(Model): # ... 其他字段 ... amount fields.DecimalField(max_digits10, decimal_places2) property def amount_text(self) - str: return f{self.amount:.2f}前端拿字符串金额去展示后端算费用用 Decimal 算各取所需。这比全局改 JSON 编码器省事也更不容易漏。5.5 时间字段自动更新不符合预期现象一条订单的created_at在每次更新后都变新原本该记录“下单时间”的字段变得毫无意义。原因我给created_at写了auto_nowTrue这个参数的意思是“每次保存时自动更新为当前时间”。正确的语义应该是创建时写入时间、更新时不改变。ormer 的DatetimeField有三个容易混淆的参数auto_now每次保存更新、auto_now_add仅创建时写入、default仅创建时如果没有传值则用默认值。解决只想要创建时间就用auto_now_addTruecreated_at fields.DatetimeField(auto_now_addTrue)如果需要在更新时记录修改时间再额外加一个updated_at fields.DatetimeField(auto_nowTrue)。区分清楚这两个参数能少写很多“时间对不上”的 bug。6. 进阶把 ormer 的查询性能压到极限的三个习惯第一个习惯是“能一次查完绝不两次查”。ormer 的prefetch在底层是IN查询但它默认对IN子句做了长度限制超过 500 个 id 会分批查。如果你一次要查几千条关联数据建议手动分批拼接 id或者反过来—先查关联表再反查主表看哪个方向数据量更小。第二个习惯是“关闭自动事务”。ormer 默认每个查询都自动提交这在写操作频繁的循环里会产生大量小事务磁盘 IO 压力很大。批量写入场景我一般这样处理from ormer.connection import db async def bulk_create_users(user_list: list[dict]): async with db.transaction(): for user in user_list: await User.objects.create(**user)一个事务包住全部插入比一千个独立事务快一个量级。注意事务内不要做网络请求这类耗时操作否则事务长时间不提交数据库连接就被占住不放。第三个习惯是“给热查询加缓存”。ormer 没有内置缓存层但它的 QuerySet 是惰性的只有执行await时才真正发 SQL。你可以给高频只读接口包一层简单的函数级缓存from functools import lru_cache lru_cache(maxsize128) def get_product_cache_key(product_id: int): return fproduct:{product_id}这里不是直接缓存数据而是缓存 key 的生成逻辑——真实场景里我更倾向于用 Redis 缓存序列化后的 JSONormer 只负责兜底查库。记住缓存的是“不常变的数据”库存、订单这类高频写数据不要碰缓存不然会出现“页面显示有货、下单却失败”的尴尬。这些习惯是我自己在两个业务系统里慢慢磨出来的。最开始我也迷信 ORM 能做所有事后来发现 ormer 的模型设计哲学会把你的思维往“表结构清晰、查询直达”的方向推。它的边界也清晰复杂分析型 SQL、窗口函数、原生 JSON 操作这些直接写db.execute_sql别硬拗模型。记住这条准则能少走很多弯路。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/13 15:18:26 · 阅读 1.2k ormer开发实战：模型驱动、事务控制与性能调优
简介一套Autoformer时序预测模型的实战案例代码包面向关注深度学习时间序列建模与金融股票数据预测的开发者、研究者。包内是完整Autoformer项目自相关注意力机制、编码解码器等核心模块配套数据加载、指标评估、训练与测试脚本并内置AMZN、NVDA、AAPL、TSLA四只股票历史数据和多组预训练权重EDA与预测Notebook可直接打开复现实验方便二次开发。资源共94个文件以Python源码24个py、42个pyc为主辅以npy数据、pth权重、csv行情、pdf文档与txt日志压缩包整体约8.1MB目录按data、models、exp等模块划分结构清晰。目前已有271人学习适合希望快速上手Autoformer或参考其工程实现开展时序预测实验的读者。1. ormer开发实战案例代码为什么说它不是又一个ORM框架把 ormer 拉进现有技术栈之前我建议大家先接受一个前提ormer 解决的问题不是“能不能映射数据库”而是“团队协作时模型定义和表结构怎么才能不打架”。很多项目初期用原生 SQL 或者老牌 ORM 都跑得挺好等人一多、表一多开始出现改模型忘迁移、联表查询写出 N1、事务边界切得稀碎这类“查不出来但线上会炸”的问题此时 ormer 的模型驱动价值才真正显现。这篇文章会从环境搭建讲到三表联动的完整案例再到我在生产环境里踩过的五个坑最后给出调优习惯。适合已经写过一段时间 CRUD、想换个更可控的数据访问层方案的开发者如果你还停留在“ORM 就是少写 SQL”的认知阶段建议带着下面的代码动手敲一遍你会在事务和预载部分重新认识它。2. 搭建 ormer 开发环境最小可运行项目的四条命令2.1 依赖安装与项目骨架常见做法是先建虚拟环境再装 ormer因为 ormer 对 Python 版本有硬性要求低于 3.11 会直接拒绝安装。我一般这样初始化mkdir ormer-demo cd ormer-demo python3.11 -m venv .venv source .venv/bin/activate pip install ormer安装完成后建议立刻跑一下ormer --version确认安装成功。ormer 的命令行工具集成度比较高建表、生成迁移文件、回滚迁移都可以通过它完成不需要额外装 alembic 之类的迁移库。项目骨架方面我习惯按“模型层、服务层、入口”三个目录组织。模型层放数据模型定义服务层放业务逻辑入口负责启动。这样做的原因是 ormer 的模型定义自带元数据注册机制所有模型必须能被入口文件 import 到否则命令行工具找不到模型定义。2.2 第一个数据模型从定义到建表ormer 的模型定义方式非常接近 SQLAlchemy 2.0 的声明式风格但取消了__tablename__的重复声明直接用类名转下划线作为表名。下面是一个最基本的用户模型# models/user.py from ormer import Model, fields class User(Model): id fields.IntField(primary_keyTrue) name fields.CharField(max_length64, indexTrue) age fields.IntField(default0) created_at fields.DatetimeField(auto_nowTrue) class Meta: database sqlite:///app.db # 当前模型绑定到哪个库这个模型的要点有三个primary_keyTrue的整型主键是 ormer 的唯一主键形式它不支持联合主键映射indexTrue会在建表时自动生成单列索引不需要单独写索引语句Meta.database是连接串配置ormer 支持 sqlite、mysql、postgresql 三种方言。写完模型后执行ormer migrate --name init_user这个命令会读取所有被 import 的模型对比数据库现有结构生成增量迁移文件并执行。--name参数是给迁移记录起的名字方便后续回滚时识别。2.3 CRUD 最小闭环增删改查的推荐写法建完表之后我写了一个最简的 CRUD 脚本来验证链路是否通。ormer 的查询 API 设计偏向“先拿 Manager 再链式调用”和 Django ORM 的套路接近# main.py import asyncio from models.user import User async def main(): # 创建关键字参数直接映射字段 user await User.objects.create( nameA同学, age28 ) print(created id:, user.id) # 查询filter 返回 QuerySetget 返回单条 u await User.objects.get(iduser.id) print(got name:, u.name) # 更新直接改字段再 save u.age 29 await u.save() # 删除调用 delete 方法 await u.delete() # 批量查询 排序 users await User.objects.filter(age__gte18).order_by(-created_at) print(count:, len(users)) asyncio.run(main())这里要注意create()是异步方法所有数据库操作都必须 await。save()在 ormer 里是“全字段更新”ormer 没有像 SQLAlchemy 那样的“脏数据追踪”机制你改了哪个字段它就更新哪个字段而是直接把整行写回去。这在单对象更新时没问题但如果你在高并发下只更新一个字段会有覆盖其他字段的隐患后面避坑章节我会细说。3. ormer 查询链路的三个核心机制模型元数据、关系预载、事务边界3.1 模型元数据ormer 为什么不写单独的表结构文件初次用 ormer 的人最不适应的一点是没有schema.sql或models.py之外的独立表结构定义文件。它的设计哲学是“模型即唯一事实源”所有建表语句、索引、约束都由模型类编译生成。这套机制靠的是 ormer 在 import 模型时自动读取字段定义把字段类型、长度、索引信息注册到一个全局元数据仓库里。好处很明显模型改了迁移文件跟着变不会再出现“代码里的模型和数据库表结构对不上”的经典问题。代价是你必须保证所有模型文件都被入口模块 import否则 ormer 根本看不到这些模型。我见过有人把模型写在models/__init__.py里但入口文件只 import 了其中一个结果其他表死活建不出来。ormer 对字段类型的映射也做得比较激进比如IntField只映射整数类型不会自动帮你做布尔值和整数的互换。CharField(max_length64)的max_length是必填参数不填在建表时会直接报错但如果你用的是 sqlite 方言长度实际上不会被数据库强制校验这是 sqlite 本身的特性不是 ormer 的 bug。3.2 关系映射与预载避免 N1 查询的两种姿势ormer 支持外键关联查询但它的默认行为是“懒加载”。也就是说查主表时不会自动联表只有你访问关联字段时才发起第二次查询。这在列表页很容易翻车——展示 100 条订单就要多查 100 次用户表响应时间直接爆炸。解决方式是显式预载# 错误示范懒加载导致 N1 orders await Order.objects.all() for order in orders: print(order.user.name) # 每条订单触发一次用户查询 # 正确示范预载关联模型 orders await Order.objects.prefetch(user).all() for order in orders: print(order.user.name) # 只发两次请求订单一次、用户一次prefetch(user)的参数是模型里定义的关系名。ormer 支持三种关系ForeignKeyField、ManyToManyField、ReverseForeignKeyField。prefetch内部实现是用WHERE id IN (...)一次性查出所有关联对象再在内存里做映射。另一种姿势是直接用select_related它在 SQL 层面用 JOIN 实现只发一条 SQL。但select_related有个限制——只能 JOIN 一层不能跨多级关系。如果 A 关联 B 关联 C你查 A 时想带出 C就必须先prefetch(b)再prefetch(b__c)。从性能角度看两表以内用select_related更快多表关系用prefetch更稳。3.3 事务与锁写业务事务时的边界控制ormer 的事务控制是我认为它最像“实战框架”的地方。它默认每个查询都自动提交autocommit但提供两种事务模式装饰器模式和上下文管理器模式。from ormer import transaction from ormer.connection import db # 装饰器模式整个函数包在事务里 transaction.atomic() async def transfer(from_id: int, to_id: int, amount: int): from models.account import Account a await Account.objects.get(idfrom_id) b await Account.objects.get(idto_id) a.balance - amount b.balance amount await a.save() await b.save() # 上下文管理器模式精确控制事务范围 async def transfer_v2(from_id: int, to_id: int, amount: int): from models.account import Account async with db.transaction(): a await Account.objects.get(idfrom_id) b await Account.objects.get(idto_id) a.balance - amount await a.save() # 这里可以做其他耗时操作事务不会提交 b.balance amount await b.save()这里有一个非常容易踩的陷阱transaction.atomic()装饰器包裹的异步函数内如果出现异常事务会自动回滚。但如果你在事务内部自己写了try...except吞掉异常ormer 不会感知到业务失败照样提交。所以事务函数内部的异常处理要格外小心一般建议让异常直接抛出到装饰器层由它统一回滚。事务隔离级别默认是数据库的默认级别mysql 是 REPEATABLE READpostgresql 是 READ COMMITTED。如果你要做“扣库存”这类强一致操作不能只靠事务还要配合select_for_update行锁后面案例部分我给出完整写法。4. 实战案例订单-商品-库存三表联动4.1 领域模型与迁移脚本现在做一个最典型的电商场景用户下单、扣库存、生成订单明细。三张表分别是商品表、库存表、订单表。库存和商品是一对一关系订单和商品是多对一关系。模型定义如下# models/shop.py from ormer import Model, fields class Product(Model): id fields.IntField(primary_keyTrue) name fields.CharField(max_length128) price fields.DecimalField(max_digits10, decimal_places2) status fields.IntField(default1) # 1上架 0下架 class Inventory(Model): id fields.IntField(primary_keyTrue) product fields.ForeignKeyField(Product, related_nameinventory) quantity fields.IntField(default0) class Order(Model): id fields.IntField(primary_keyTrue) user_id fields.IntField(indexTrue) product fields.ForeignKeyField(Product, related_nameorders) quantity fields.IntField() amount fields.DecimalField(max_digits10, decimal_places2) status fields.IntField(default0) # 0待支付 1已支付 2已取消 created_at fields.DatetimeField(auto_nowTrue)写完后执行迁移ormer migrate --name init_shop_tablesDecimalField是 ormer 比较有特色的字段类型它自动映射数据库的 DECIMAL 类型避免浮点精度问题。ForeignKeyField的第一个参数是模型名字符串不是模型类本身这是为了规避循环 import 的问题。related_name指定反向查询名称比如通过 product 查它的库存就是await product.inventory.get()这里 inventory 是反向关系的“集合名”。4.2 下单接口的完整实现下单流程涉及三个操作检查商品状态、检查库存量、扣减库存、创建订单。这四个操作必须在一个事务里否则会出现“库存扣了但订单没建成”的数据不一致。完整代码如下# services/order_service.py from decimal import Decimal from ormer import transaction from models.shop import Product, Inventory, Order transaction.atomic() async def create_order(user_id: int, product_id: int, quantity: int): # 1. 锁商品行防止状态被并发改掉 product await Product.objects.select_for_update().get(idproduct_id) if product.status ! 1: raise ValueError(商品已下架) # 2. 锁库存行防止超卖 inventory await Inventory.objects.select_for_update().get( product_idproduct_id ) if inventory.quantity quantity: raise ValueError(库存不足) # 3. 扣减库存 inventory.quantity - quantity await inventory.save() # 4. 创建订单金额单价×数量 amount product.price * Decimal(quantity) order await Order.objects.create( user_iduser_id, product_idproduct_id, quantityquantity, amountamount, ) return order.id这个接口的核心在select_for_update()。ormer 的select_for_update()实际上执行的是SELECT ... FOR UPDATE在 mysql 和 postgresql 中会把命中的行锁住直到事务提交或回滚。两个并发请求同时到这行代码时第二个请求会阻塞等待直到第一个事务结束。这就是避免超卖的关键。有个细节值得注意select_for_update().get()和select_for_update().filter()的语义完全一致都会在事务期间持有行锁。但如果你在事务外调用select_for_update()ormer 会抛出异常因为锁必须依赖事务存在。排查“锁不生效”问题时优先确认事务边界是否真的包住了这条查询。4.3 库存扣减的并发保护上面的代码已经能解决基本超卖问题但还不够。真实电商场景里用户可能对同一个商品同时下多笔订单。两笔订单都读到库存 10都满足扣减条件都执行扣减最终库存变成 8 而不是 0。原因是inventory.save()是“读改写”三步行锁只锁住了读没有锁住写。ormer 的save()默认是整行更新也就是它会用对象里的所有字段覆盖数据库当前值。这行得通的前提是事务隔离级别够高或者数据库层面做了乐观锁。常见做法是在 Inventory 表里加一个version字段做乐观锁# 在 Inventory 模型里增加版本号字段 class Inventory(Model): id fields.IntField(primary_keyTrue) product fields.ForeignKeyField(Product, related_nameinventory) quantity fields.IntField(default0) version fields.IntField(default0)更新时在 SQL 层面做条件更新只有version匹配时才更新同时把version加一。# 乐观锁更新 from ormer.connection import db from models.shop import Inventory async def decrease_inventory(inventory_id: int, delta: int, expected_version: int): sql ( UPDATE inventory SET quantity quantity - ?, version version 1 WHERE id ? AND version ? ) cursor await db.execute_sql( sql, (delta, inventory_id, expected_version) ) if cursor.rowcount 0: raise ValueError(库存已被其他请求修改请重试)这里db.execute_sql是 ormer 留给开发者写原生 SQL 的逃生舱。当 ORM 的表达能力不足以描述复杂更新时直接写 SQL 是最稳妥的。注意占位符用的是?不管底层是 mysql 还是 postgresqlormer 都会自动做参数适配不需要自己切换%s。最后再强调一遍select_for_update是悲观锁适合高竞争、低冲突的场景乐观锁适合读多写少、冲突概率低的场景。如果你们项目的下单峰值不高每秒几十单悲观锁完全够用别过度设计。5. ormer 实战避坑五个让我翻过车的细节5.1 全字段更新覆盖了并发修改现象两个请求同时改一条订单记录一个改金额一个改状态。先提交的那个请求把后提交的字段覆盖回旧值线上数据出现“改了状态但金额变回原样”的诡异情况。原因ormer 的save()是整行更新不带任何字段筛选。它读取对象时拿到的所有字段值更新时全部写回数据库。如果两个请求各自读到不同时点的快照后提交的快照会覆盖先提交的修改。这不是 ormer 的 bug是 ORM 全字段更新的通用行为。解决并发更新场景不要用save()改用update()方法只更新指定字段await Order.objects.filter(idorder_id).update( status1, amountDecimal(99.00) )update()会生成UPDATE ... SET语句只更新传入的字段不读不改其他列天然免疫覆盖问题。5.2 懒加载在异步上下文中导致事件循环错乱现象代码里for order in orders: print(order.user.name)在异步函数中偶发RuntimeError: Event loop is closed本地跑好好的线上跑一会儿就炸。原因ormer 的懒加载是异步的访问order.user时会发起新的数据库查询。但如果你在循环里直接逐条访问等于在同一个事件循环里不断创建新任务。高并发时 ormer 内部连接池不够用新任务阻塞等待空闲连接而外层asyncio.run可能已经关闭事件循环导致连接池异常。解决列表页、批量查询场景一定要用prefetch预载禁止在循环里访问关联属性。如果必须动态判断是否需要加载关联用await order.get_related(user)显式加载至少保证查询发生在明确的异步上下文里。5.3 迁移命令找不到模型现象执行ormer migrate提示No model found明明models/目录下的文件都写好了。原因ormer 的元数据注册发生在模型类被 import 时。如果你只写了文件但没有任何地方import它ormer 根本不知道它的存在。入口文件只 import 了main.py里用到的模型其他模型文件“孤零零”地躺在目录里命令自然找不到。解决在models/__init__.py里显式导入所有模型然后确保入口文件 import 这个包# models/__init__.py from models.user import User from models.shop import Product, Inventory, Order然后在main.py加上import models即可。这属于 ormer 的“约定优于配置”但它不会自动扫描目录必须手动注册。5.4 Decimal 字段在 JSON 序列化时翻车现象接口返回订单金额给前端时json.dumps直接报Object of type Decimal is not JSON serializable。原因DecimalField映射到数据库的 DECIMAL 类型读出来是 Python 的Decimal对象。这个对象不是原生 JSON 类型DRF、Flask-RESTful 这类序列化器不处理它。解决给序列化器配置一个自定义编码器或者在模型上添加一个属性方法返回字符串形式class Order(Model): # ... 其他字段 ... amount fields.DecimalField(max_digits10, decimal_places2) property def amount_text(self) - str: return f{self.amount:.2f}前端拿字符串金额去展示后端算费用用 Decimal 算各取所需。这比全局改 JSON 编码器省事也更不容易漏。5.5 时间字段自动更新不符合预期现象一条订单的created_at在每次更新后都变新原本该记录“下单时间”的字段变得毫无意义。原因我给created_at写了auto_nowTrue这个参数的意思是“每次保存时自动更新为当前时间”。正确的语义应该是创建时写入时间、更新时不改变。ormer 的DatetimeField有三个容易混淆的参数auto_now每次保存更新、auto_now_add仅创建时写入、default仅创建时如果没有传值则用默认值。解决只想要创建时间就用auto_now_addTruecreated_at fields.DatetimeField(auto_now_addTrue)如果需要在更新时记录修改时间再额外加一个updated_at fields.DatetimeField(auto_nowTrue)。区分清楚这两个参数能少写很多“时间对不上”的 bug。6. 进阶把 ormer 的查询性能压到极限的三个习惯第一个习惯是“能一次查完绝不两次查”。ormer 的prefetch在底层是IN查询但它默认对IN子句做了长度限制超过 500 个 id 会分批查。如果你一次要查几千条关联数据建议手动分批拼接 id或者反过来—先查关联表再反查主表看哪个方向数据量更小。第二个习惯是“关闭自动事务”。ormer 默认每个查询都自动提交这在写操作频繁的循环里会产生大量小事务磁盘 IO 压力很大。批量写入场景我一般这样处理from ormer.connection import db async def bulk_create_users(user_list: list[dict]): async with db.transaction(): for user in user_list: await User.objects.create(**user)一个事务包住全部插入比一千个独立事务快一个量级。注意事务内不要做网络请求这类耗时操作否则事务长时间不提交数据库连接就被占住不放。第三个习惯是“给热查询加缓存”。ormer 没有内置缓存层但它的 QuerySet 是惰性的只有执行await时才真正发 SQL。你可以给高频只读接口包一层简单的函数级缓存from functools import lru_cache lru_cache(maxsize128) def get_product_cache_key(product_id: int): return fproduct:{product_id}这里不是直接缓存数据而是缓存 key 的生成逻辑——真实场景里我更倾向于用 Redis 缓存序列化后的 JSONormer 只负责兜底查库。记住缓存的是“不常变的数据”库存、订单这类高频写数据不要碰缓存不然会出现“页面显示有货、下单却失败”的尴尬。这些习惯是我自己在两个业务系统里慢慢磨出来的。最开始我也迷信 ORM 能做所有事后来发现 ormer 的模型设计哲学会把你的思维往“表结构清晰、查询直达”的方向推。它的边界也清晰复杂分析型 SQL、窗口函数、原生 JSON 操作这些直接写db.execute_sql别硬拗模型。记住这条准则能少走很多弯路。希望帮到你。本文还有配套的精品资源点击获取