简介这是一套面向中高级Python开发者的对象关系映射实战案例代码包聚焦如何在真实项目中运用数据建模思想减少手动编写SQL语句提升业务代码的可维护性与复用性。压缩包内提供完整工程目录和可运行脚本覆盖数据模型定义、查询构造、关联关系配置、增删改查等核心环节并附带实验环境配置、运行结果与文档说明便于按模块对照学习和二次开发。资源共94个文件以Python源码为主辅以数据表、数值数组、模型权重、PDF文档、Jupyter笔记本及文本说明分别承担业务逻辑、数据存储、数值计算、模型保存、资料查阅与交互式分析等职责整个压缩包约8.1MB目录结构清晰可快速定位关键模块。目前已有271人学习下载适合已掌握Python基础、希望借助完整案例巩固数据层开发能力并熟悉工程项目结构的开发者可应用于博客系统、电商平台或用户管理等典型后端场景。1. ormer开发实战案例代码把反复写SQL的力气省下来手写 SQL 写久了谁都有过这样的时刻明明只是换一个查询条件却要把同一段WHERE拼上三四遍明明只需要一张表却因为关联关系没设计好散落出七八个查询函数。我第一次看到 ormer 开发实战案例代码时以为它又是一套“增删改查 Demo”翻完目录才发现它把所有容易被忽略的配套逻辑都放进了代码里事务边界、软删除、连接池耗尽、多表聚合查询以及一个可以插进现有软件项目的插件扩展点。这套代码典型场景是给内部管理系统做数据访问层适合正在做后端业务、每天处理重复 SQL 的开发者也适合想把数据层做成可插拔软件插件的二次开发人群。如果你也经历过“改一个字段要改三个接口”的尴尬这套代码能直接拿来当脚手架。2. 架构设计一张表映射到一个对象的插件式数据层2.1 目录结构先看清 ormer 的六个目录拆项目第一步我是从目录开始的。ormer 没有把代码塞进一个包而是按数据访问链路拆成了六个模块前两个是核心后四个是业务扩展。ormer/ ├── ormer_core/ # 核心引擎连接管理、模型注册、SQL 构造 │ ├── connection.py │ ├── fields.py │ ├── model.py │ ├── session.py │ └── transaction.py ├── models/ # 业务表映射对象 │ ├── order.py │ ├── customer.py │ └── product.py ├── repositories/ # 数据访问层封装 CRUD │ ├── order_repo.py │ └── customer_repo.py ├── services/ # 业务逻辑层调用 repo │ └── order_service.py ├── migrations/ # 表结构变更脚本 │ ├── 001_init.py │ └── 002_add_status_index.py └── plugins/ # 插件扩展点 └── tenancy.py这里的ormer_core是整个软件数据层的灵魂它只做三件事把模型注册到数据库元数据、把查询条件转换成安全的 SQL、把事务状态管好。models、repositories、services是标准的软件分层互不越界。plugins目录是我最关注的位置它让 ormer 能以插件方式嵌进现有项目而不是强制你重写整个业务系统。初次接触时最容易忽略的是migrations。很多个人项目图省事直接让代码建表一到上生产就变成“表结构到底谁改的”黑盒子。ormer 把它单独拎出来每次表变更都保留一个版本脚本这个设计思路值得直接抄走。2.2 配置与连接池用单例管理数据库会话跑通 ormer 之前先要看懂它的配置入口。项目里没有把数据库连接写死而是用环境变量加载配置方便不同环境直接切换。# config.py import os from ormer_core import connection def load_config(): return { host: os.getenv(DB_HOST, 127.0.0.1), port: int(os.getenv(DB_PORT, 5432)), user: os.getenv(DB_USER, blog), password: os.getenv(DB_PASS, dev_password), database: os.getenv(DB_NAME, ormer_demo), pool_size: int(os.getenv(DB_POOL_SIZE, 10)), max_overflow: int(os.getenv(DB_POOL_MAX, 20)), pool_recycle: int(os.getenv(DB_POOL_RECYCLE, 3600)), } # 模块级单例整个进程只初始化一次 db connection.Database(load_config())这段配置里最值得留意的是pool_size、max_overflow、pool_recycle三个参数。pool_size是连接池里常驻连接数max_overflow表示连接不够时最多额外扩多少个pool_recycle是连接最长存活时间超过这个秒数会被强制回收。我之前遇到过数据库偶尔报“too many connections”就是因为pool_recycle没配置底层连接被网络设备空闲断开后应用层不知道还在继续往外发。连接池还有一个常见误区不要把Database实例每次调用都 new 一遍。ormer 这里在模块顶层初始化一次整个进程共享单例我一般也这么用。如果项目里用了多进程模式再把连接池分配给每个 worker效果等同。2.3 插件扩展点为什么数据访问层适合做软件插件很多人一听到“软件/插件”第一反应是前端里的菜单扩展。实际上 ormer 的插件机制是把“全局行为”做成可插拔策略。比如多租户隔离有了插件后可以不改repositories里任何 SQL只挂一个全局过滤器就能让每张表都自动带上租户编号条件。插件点面向网络请求上下文获取租户信息核心引擎只负责执行已构造好的条件。我在 ormer 的源码里看到插件通过装饰器注册业务代码完全不知道插件存在。这样切环境、切租户、切换逻辑只需要调一个加载函数不会碰业务代码。对二次开发来说这个扩展点是最值得研究的后文我会专门讲如何加一个自己的插件。3. 核心模块拆解一个请求从对象到 SQL 的完整链路3.1 模型声明用声明式风格定义一张订单表ormer 的模型声明比常见的 ORM 方案更接近字段先行的思路。下面这段是models/order.py的核心定义# models/order.py from ormer_core import fields, Model class Order(Model): __tablename__ orders id fields.Integer(primary_keyTrue) order_no fields.String(length64, uniqueTrue, indexTrue) customer_id fields.Integer(nullableFalse) amount fields.Decimal(max_digits10, decimal_places2) status fields.String(length20, defaultpending) created_at fields.Datetime(auto_now_addTrue) updated_at fields.Datetime(auto_nowTrue) # 逻辑删除标记下文软删除实战中会用到 deleted_at fields.Datetime(nullableTrue, hiddenTrue)关键点有四个primary_key声明主键uniqueTrue会生成唯一约束indexTrue自动建索引hiddenTrue表示默认查询不带这个字段。第一次看 ormer 代码时我被hidden骗了以为它是过滤掉字段实际上它只控制字段是否出现在自动生成的查询列中并不影响 SQL 里的WHERE条件。写模型最常见的问题是“表和模型字段对不上不改迁移”。ormer 的模型定义里没有绝对标准答案但应该尽量让default、nullable与数据库约束保持一致。否则代码层觉得能插入空值数据库直接报异常两边来回扯皮。3.2 数据仓库把 CRUD 收敛成业务可调用的方法repositories/order_repo.py承担所有订单表的原子操作。业务层不直接拼 SQL而是调用仓库对象里的方法。我把核心代码精简成下面这段# repositories/order_repo.py from ormer_core import session from models.order import Order class OrderRepo: def create_order(self, payload: dict) - Order: entity Order(**payload) session.add(entity) session.commit() session.refresh(entity) return entity def get_by_id(self, order_id: int) - Order | None: return session.query(Order).get(idorder_id) def update_status(self, order_id: int, status: str): order self.get_by_id(order_id) if not order: return None order.status status session.commit() return order这里我特别想解释refresh的作用。创建订单后数据库里的默认时间、自增主键是在插入那一刻生成的如果不refresh内存里的对象可能还是None后续业务取order.id会拿到空值。这是很多初学者第一次跑 ormer 时翻车的点。另外注意session是全局导入的单例还是每次请求新建ormer 给了两种模式请求级会话和全局会话。上面对简单脚本用全局会话没问题但 Web 项目里建议每个请求结束后主动关闭会话避免连接被占用不归还。3.3 查询条件不靠拼字符串的过滤器手写 SQL 最怕条件拼接漏掉空格、引号导致注入。ormer 用过滤器对象表达查询条件这段代码是从复杂筛选接口里抽出来的# services/order_service.py from ormer_core import session from models.order import Order def search_orders(customer_id: int None, status: str None, page: int 1, page_size: int 20): query session.query(Order) if customer_id is not None: query query.filter(Order.customer_id customer_id) if status is not None: query query.filter(Order.status status) orders query.order_by(Order.created_at.desc()) \ .limit(page_size) \ .offset((page - 1) * page_size) \ .all() return ordersquery.filter返回的是一个新查询对象原对象不变所以可以无脑链式调用。这也是排查问题时最有用的特性你能在任意位置拿query打印出待执行的 SQL不用去猜数据库到底收到什么。limit和offset适合小数据量分页数据体量过了百万级再换游标分页。业务代码里常见的一种错误是“先取所有数据再在 Python 里筛选”导致内存被非业务字段撑爆。改成过滤器下推到 SQL传输量和内存占用立刻下降这个案例我反复看了好几遍。4. 业务场景实战事务、多表关联与软删除的工程写法4.1 事务边界一个创建订单扣库存的原子操作单表 CRUD 跑通不算实战真实软件里最怕的是“一条数据成功、另一条失败”。ormer 的事务代码在ormer_core/transaction.py里封装成了上下文管理器业务层使用起来很接近直觉# services/order_service.py from ormer_core import transaction from repositories.order_repo import OrderRepo from repositories.inventory_repo import InventoryRepo def create_order_with_inventory(payload: dict): with transaction(): order OrderRepo().create_order(payload) InventoryRepo().deduct(product_idpayload[product_id], quantitypayload[quantity]) return order这段代码里只要deduct抛出异常整个数据块都会回滚order的插入操作不会保留。初学者容易误以为“在 Python 里 try 一下就好了”但事务的关键在于提交窗口只有把所有修改放在同一个事务上下文里数据库才能保证原子性。使用事务时还应留意transaction()内部是启动新事务还是沿用当前事务。ormer 的做法是调用session.begin_nested()做保存点内层异常只回滚到保存点外层成功再把全部数据提交。这样做的好处是支持业务层嵌套调用不会因为一个内层方法失败导致整个请求全灭。4.2 多表关联不写 JOIN 也能完成一对多查询关联查询是或门项目里写得最有价值的部分。models/customer.py定义了客户models/order.py通过customer_id关联。ormer 让模型直接声明关系拿到客户对象后可以像访问属性一样拿到订单列表# models/customer.py from ormer_core import fields, Model, relation class Customer(Model): __tablename__ customers id fields.Integer(primary_keyTrue) name fields.String(length64) orders relation(Order, backrefcustomer, lazyselectin)lazyselectin表示查询客户时用第二条SELECT ... WHERE customer_id IN (...)把订单一次性带出来。这个参数很重要如果默认用延迟加载业务侧循环访问customer.orders就可能导致发几十条 SQL也就是常说的 N1 查询。我在实际开发里吃过这个亏列表页展示一百个客户的订单数量结果原来 1 条查询变成了 101 条数据库响应直接糊掉。改成selectin后ORM 先查出客户列表再根据客户主键批量查出订单全程两条 SQL。如果你遇到接口变慢第一反应先去看日志里有没有 N1。4.3 软删除与全局过滤每张表都带deleted_atormer 处理逻辑删除的思路很彻底它会把“未删除”做成全局默认过滤器让业务代码无感。定义如下# ormer_core/soft_delete.py from ormer_core import hooks hooks.register(before_select) def apply_deleted_filter(query, model_class): if hasattr(model_class, deleted_at): query query.filter(model_class.deleted_at.is_(None)) return query这段代码的意思是任何经过 ormer 查询引擎的 SELECT 语句如果目标模型有deleted_at字段就自动追加WHERE deleted_at IS NULL。业务层完全不需要手动写“未删除”条件二次开发时也不会漏掉删数据校验。软删除本身不复杂复杂的是唯一约束。比如订单表中order_no字段设置了uniqueTrue软删除后再次插入相同订单号会撞唯一约束。常见做法是把唯一约束改成组合唯一索引(order_no, deleted_at)这样不同删除时间的数据彼此隔离。这个坑我最早是在生产环境里踩到的恢复一条误删数据时插入同样订单号直接失败。另一个容易忽略的是软删除字段只对查询生效对UPDATE条件未必生效。如果你使用了全局过滤器一定要确认它也覆盖了更新操作的WHERE条件不然会出现“能查出来但更新报错”的奇怪状态。5. 排查手册跑 ormer 前后最容易翻车的五个位置5.1 循环导入导致的模型注册失败现象启动项目时报ImportError: cannot import name Order或者模型注册顺序不对数据库表始终少几张。原因模型 A 和模型 B 存在外键关系两个模块在models/__init__.py里互相导入Python 解释器处理到一半时有一方还没加载完。ormer 的模型注册依赖元数据导入顺序错了后续表映射就乱了。解决不要在models/__init__.py里直接from order import Order、from customer import Customer这种显式互引。改成在ormer_core的注册入口统一加载或者把关联关系写成一个单独模块例如models/relations.py。我习惯的做法是让模块内部不引用其他模块只声明本地字段关系定义集中到外层注册时补充这样既避开循环导入也方便查看整体结构。5.2 连接池耗尽数据库连接被占满现象接口偶发超时日志里出现cant connect to database重试一两秒又恢复数据库服务端排查时连接数处于高位。原因连接池配置不合适或者业务代码打开会话后没有释放。如果每个请求都调用session.add()却不close()连接不会被归还到池子。解决先看配置里的pool_size是否过小常见个人电脑设置 5 就行了生产环境 1020 起步还要配max_overflow。更关键的是用完关闭会话在 Web 框架拦截器或异步任务的 finally 块里统一调用session.close()。我一般会在请求入口处用中间件包一层try/finally保证异常时也会正常归还连接。5.3 事务回滚后 session 还残留着旧数据现象事务里先插入一条数据随后业务逻辑判断失败主动raise但在外层except里重新查询这张表发现脏数据还能查到。原因事务回滚只撤销了数据库端的变更当前 session 对象里还缓存着未提交的实体。ORM 的会话缓存没有跟着回滚刷新再次查询优先命中缓存。解决事务回滚后调用session.expire_all()或session.clear()强制下次查询重新从数据库读取。如果你的业务代码在事务外还要用同一个 session这点必须处理。我在 ormer 的源码里看到事务上下文管理器在__exit__里做了回滚后的缓存清理但是二次开发时如果自己写事务块很容易漏掉这个操作。5.4 迁移一周后表结构跟模型代码对不上现象本地跑得好好的测试环境突然报no such column: deleted_at检查数据库发现表结构还是旧版本。原因ormer 默认不会自动把模型字段映射成 DDL除非你调用了迁移引擎。很多人图省事直接让create_all建表字段变更后create_all又不会更新已有表。解决严格走migrations目录的变更脚本。每新增一个字段或者改约束就加一个版本脚本例如003_add_delivery_type.py。如果团队规模小也可以接受手动用外接工具改表但一定要把变更记录同步到代码库不然项目换个人就全凭经验猜了。5.5 时区问题本地保存created_at跟生产查出来差 8 小时现象数据库里存的时间是东八区接口返回时客户端看到的是世界协调时或者查询当天订单时边界少算一个小时。原因数据库引擎、连接参数和应用服务器三处时区不一致。auto_now_add生成的是 Python 侧的当前时间如果连接串里没指定时区数据库写入时会按它自己的 session 时区转换。解决在 ormer 的配置模块里统一设置时区例如timezone: Asia/Shanghai同时让应用层只存取时间戳展示层的时区转换交给前端。另外存储时间类型尽量用带时区的时间类型不要用字符串拼的本地时间字符串在排序和跨时区计算时就是个灾难。6. 二次开发改造思路给 ormer 加一个多租户插件如果要在这个项目上做插件化改造我建议从plugins目录切入。整个改造思路是不改动ormer_core的查询逻辑只注册一个全局过滤器。下面是一个多租户插件的最小实现# plugins/tenancy.py from ormer_core import hooks from contextvars import ContextVar current_tenant ContextVar(current_tenant, defaultNone) hooks.register(before_select) def add_tenant_filter(query, model_class): tenant current_tenant.get() if tenant and hasattr(model_class, tenant_id): query query.filter(model_class.tenant_id tenant) return query这个插件利用 Python 的ContextVar保存当前请求的租户 ID每个 Web 请求进来时调用current_tenant.set(tenant_id)退出时重置。之后所有模型查询都会自动带上租户条件。这里要注意两点一是必须让筛选字段名在不同模型中统一为tenant_id二是如果模型不想参与租户隔离可以约定一个内部标记跳过。我实际改造过类似系统发现最有价值的不是上面这些代码而是先给项目补一套“查询测试”。每次新增插件我就用一段固定数据跑一遍单测确认哪些查询被全局过滤器影响、哪些没有。这样能让插件行为透明化不至于上线后才发现某个统计报表把其他租户的数据也带了出来。老实说ormer 不是那种打开就能跑的现成框架它更像一份能让你学会“怎么把数据访问层写好”的源码笔记。我拆完这个项目后最深的体会是数据层的坑不会消失只会转移。从那以后我每次改数据访问层都会强制把“事务边界、软删除、N1 查询、连接池释放”过一遍清单这套习惯直接来自拆 ormer 的过程。希望你也能从这个项目里找到适合自己业务的改造起点。本文还有配套的精品资源点击获取 
长河君 · 2026/10/24 0:04:36 · 阅读 1.2k ormer开发实战：从SQL到数据访问层插件化的工程实践
简介这是一套面向中高级Python开发者的对象关系映射实战案例代码包聚焦如何在真实项目中运用数据建模思想减少手动编写SQL语句提升业务代码的可维护性与复用性。压缩包内提供完整工程目录和可运行脚本覆盖数据模型定义、查询构造、关联关系配置、增删改查等核心环节并附带实验环境配置、运行结果与文档说明便于按模块对照学习和二次开发。资源共94个文件以Python源码为主辅以数据表、数值数组、模型权重、PDF文档、Jupyter笔记本及文本说明分别承担业务逻辑、数据存储、数值计算、模型保存、资料查阅与交互式分析等职责整个压缩包约8.1MB目录结构清晰可快速定位关键模块。目前已有271人学习下载适合已掌握Python基础、希望借助完整案例巩固数据层开发能力并熟悉工程项目结构的开发者可应用于博客系统、电商平台或用户管理等典型后端场景。1. ormer开发实战案例代码把反复写SQL的力气省下来手写 SQL 写久了谁都有过这样的时刻明明只是换一个查询条件却要把同一段WHERE拼上三四遍明明只需要一张表却因为关联关系没设计好散落出七八个查询函数。我第一次看到 ormer 开发实战案例代码时以为它又是一套“增删改查 Demo”翻完目录才发现它把所有容易被忽略的配套逻辑都放进了代码里事务边界、软删除、连接池耗尽、多表聚合查询以及一个可以插进现有软件项目的插件扩展点。这套代码典型场景是给内部管理系统做数据访问层适合正在做后端业务、每天处理重复 SQL 的开发者也适合想把数据层做成可插拔软件插件的二次开发人群。如果你也经历过“改一个字段要改三个接口”的尴尬这套代码能直接拿来当脚手架。2. 架构设计一张表映射到一个对象的插件式数据层2.1 目录结构先看清 ormer 的六个目录拆项目第一步我是从目录开始的。ormer 没有把代码塞进一个包而是按数据访问链路拆成了六个模块前两个是核心后四个是业务扩展。ormer/ ├── ormer_core/ # 核心引擎连接管理、模型注册、SQL 构造 │ ├── connection.py │ ├── fields.py │ ├── model.py │ ├── session.py │ └── transaction.py ├── models/ # 业务表映射对象 │ ├── order.py │ ├── customer.py │ └── product.py ├── repositories/ # 数据访问层封装 CRUD │ ├── order_repo.py │ └── customer_repo.py ├── services/ # 业务逻辑层调用 repo │ └── order_service.py ├── migrations/ # 表结构变更脚本 │ ├── 001_init.py │ └── 002_add_status_index.py └── plugins/ # 插件扩展点 └── tenancy.py这里的ormer_core是整个软件数据层的灵魂它只做三件事把模型注册到数据库元数据、把查询条件转换成安全的 SQL、把事务状态管好。models、repositories、services是标准的软件分层互不越界。plugins目录是我最关注的位置它让 ormer 能以插件方式嵌进现有项目而不是强制你重写整个业务系统。初次接触时最容易忽略的是migrations。很多个人项目图省事直接让代码建表一到上生产就变成“表结构到底谁改的”黑盒子。ormer 把它单独拎出来每次表变更都保留一个版本脚本这个设计思路值得直接抄走。2.2 配置与连接池用单例管理数据库会话跑通 ormer 之前先要看懂它的配置入口。项目里没有把数据库连接写死而是用环境变量加载配置方便不同环境直接切换。# config.py import os from ormer_core import connection def load_config(): return { host: os.getenv(DB_HOST, 127.0.0.1), port: int(os.getenv(DB_PORT, 5432)), user: os.getenv(DB_USER, blog), password: os.getenv(DB_PASS, dev_password), database: os.getenv(DB_NAME, ormer_demo), pool_size: int(os.getenv(DB_POOL_SIZE, 10)), max_overflow: int(os.getenv(DB_POOL_MAX, 20)), pool_recycle: int(os.getenv(DB_POOL_RECYCLE, 3600)), } # 模块级单例整个进程只初始化一次 db connection.Database(load_config())这段配置里最值得留意的是pool_size、max_overflow、pool_recycle三个参数。pool_size是连接池里常驻连接数max_overflow表示连接不够时最多额外扩多少个pool_recycle是连接最长存活时间超过这个秒数会被强制回收。我之前遇到过数据库偶尔报“too many connections”就是因为pool_recycle没配置底层连接被网络设备空闲断开后应用层不知道还在继续往外发。连接池还有一个常见误区不要把Database实例每次调用都 new 一遍。ormer 这里在模块顶层初始化一次整个进程共享单例我一般也这么用。如果项目里用了多进程模式再把连接池分配给每个 worker效果等同。2.3 插件扩展点为什么数据访问层适合做软件插件很多人一听到“软件/插件”第一反应是前端里的菜单扩展。实际上 ormer 的插件机制是把“全局行为”做成可插拔策略。比如多租户隔离有了插件后可以不改repositories里任何 SQL只挂一个全局过滤器就能让每张表都自动带上租户编号条件。插件点面向网络请求上下文获取租户信息核心引擎只负责执行已构造好的条件。我在 ormer 的源码里看到插件通过装饰器注册业务代码完全不知道插件存在。这样切环境、切租户、切换逻辑只需要调一个加载函数不会碰业务代码。对二次开发来说这个扩展点是最值得研究的后文我会专门讲如何加一个自己的插件。3. 核心模块拆解一个请求从对象到 SQL 的完整链路3.1 模型声明用声明式风格定义一张订单表ormer 的模型声明比常见的 ORM 方案更接近字段先行的思路。下面这段是models/order.py的核心定义# models/order.py from ormer_core import fields, Model class Order(Model): __tablename__ orders id fields.Integer(primary_keyTrue) order_no fields.String(length64, uniqueTrue, indexTrue) customer_id fields.Integer(nullableFalse) amount fields.Decimal(max_digits10, decimal_places2) status fields.String(length20, defaultpending) created_at fields.Datetime(auto_now_addTrue) updated_at fields.Datetime(auto_nowTrue) # 逻辑删除标记下文软删除实战中会用到 deleted_at fields.Datetime(nullableTrue, hiddenTrue)关键点有四个primary_key声明主键uniqueTrue会生成唯一约束indexTrue自动建索引hiddenTrue表示默认查询不带这个字段。第一次看 ormer 代码时我被hidden骗了以为它是过滤掉字段实际上它只控制字段是否出现在自动生成的查询列中并不影响 SQL 里的WHERE条件。写模型最常见的问题是“表和模型字段对不上不改迁移”。ormer 的模型定义里没有绝对标准答案但应该尽量让default、nullable与数据库约束保持一致。否则代码层觉得能插入空值数据库直接报异常两边来回扯皮。3.2 数据仓库把 CRUD 收敛成业务可调用的方法repositories/order_repo.py承担所有订单表的原子操作。业务层不直接拼 SQL而是调用仓库对象里的方法。我把核心代码精简成下面这段# repositories/order_repo.py from ormer_core import session from models.order import Order class OrderRepo: def create_order(self, payload: dict) - Order: entity Order(**payload) session.add(entity) session.commit() session.refresh(entity) return entity def get_by_id(self, order_id: int) - Order | None: return session.query(Order).get(idorder_id) def update_status(self, order_id: int, status: str): order self.get_by_id(order_id) if not order: return None order.status status session.commit() return order这里我特别想解释refresh的作用。创建订单后数据库里的默认时间、自增主键是在插入那一刻生成的如果不refresh内存里的对象可能还是None后续业务取order.id会拿到空值。这是很多初学者第一次跑 ormer 时翻车的点。另外注意session是全局导入的单例还是每次请求新建ormer 给了两种模式请求级会话和全局会话。上面对简单脚本用全局会话没问题但 Web 项目里建议每个请求结束后主动关闭会话避免连接被占用不归还。3.3 查询条件不靠拼字符串的过滤器手写 SQL 最怕条件拼接漏掉空格、引号导致注入。ormer 用过滤器对象表达查询条件这段代码是从复杂筛选接口里抽出来的# services/order_service.py from ormer_core import session from models.order import Order def search_orders(customer_id: int None, status: str None, page: int 1, page_size: int 20): query session.query(Order) if customer_id is not None: query query.filter(Order.customer_id customer_id) if status is not None: query query.filter(Order.status status) orders query.order_by(Order.created_at.desc()) \ .limit(page_size) \ .offset((page - 1) * page_size) \ .all() return ordersquery.filter返回的是一个新查询对象原对象不变所以可以无脑链式调用。这也是排查问题时最有用的特性你能在任意位置拿query打印出待执行的 SQL不用去猜数据库到底收到什么。limit和offset适合小数据量分页数据体量过了百万级再换游标分页。业务代码里常见的一种错误是“先取所有数据再在 Python 里筛选”导致内存被非业务字段撑爆。改成过滤器下推到 SQL传输量和内存占用立刻下降这个案例我反复看了好几遍。4. 业务场景实战事务、多表关联与软删除的工程写法4.1 事务边界一个创建订单扣库存的原子操作单表 CRUD 跑通不算实战真实软件里最怕的是“一条数据成功、另一条失败”。ormer 的事务代码在ormer_core/transaction.py里封装成了上下文管理器业务层使用起来很接近直觉# services/order_service.py from ormer_core import transaction from repositories.order_repo import OrderRepo from repositories.inventory_repo import InventoryRepo def create_order_with_inventory(payload: dict): with transaction(): order OrderRepo().create_order(payload) InventoryRepo().deduct(product_idpayload[product_id], quantitypayload[quantity]) return order这段代码里只要deduct抛出异常整个数据块都会回滚order的插入操作不会保留。初学者容易误以为“在 Python 里 try 一下就好了”但事务的关键在于提交窗口只有把所有修改放在同一个事务上下文里数据库才能保证原子性。使用事务时还应留意transaction()内部是启动新事务还是沿用当前事务。ormer 的做法是调用session.begin_nested()做保存点内层异常只回滚到保存点外层成功再把全部数据提交。这样做的好处是支持业务层嵌套调用不会因为一个内层方法失败导致整个请求全灭。4.2 多表关联不写 JOIN 也能完成一对多查询关联查询是或门项目里写得最有价值的部分。models/customer.py定义了客户models/order.py通过customer_id关联。ormer 让模型直接声明关系拿到客户对象后可以像访问属性一样拿到订单列表# models/customer.py from ormer_core import fields, Model, relation class Customer(Model): __tablename__ customers id fields.Integer(primary_keyTrue) name fields.String(length64) orders relation(Order, backrefcustomer, lazyselectin)lazyselectin表示查询客户时用第二条SELECT ... WHERE customer_id IN (...)把订单一次性带出来。这个参数很重要如果默认用延迟加载业务侧循环访问customer.orders就可能导致发几十条 SQL也就是常说的 N1 查询。我在实际开发里吃过这个亏列表页展示一百个客户的订单数量结果原来 1 条查询变成了 101 条数据库响应直接糊掉。改成selectin后ORM 先查出客户列表再根据客户主键批量查出订单全程两条 SQL。如果你遇到接口变慢第一反应先去看日志里有没有 N1。4.3 软删除与全局过滤每张表都带deleted_atormer 处理逻辑删除的思路很彻底它会把“未删除”做成全局默认过滤器让业务代码无感。定义如下# ormer_core/soft_delete.py from ormer_core import hooks hooks.register(before_select) def apply_deleted_filter(query, model_class): if hasattr(model_class, deleted_at): query query.filter(model_class.deleted_at.is_(None)) return query这段代码的意思是任何经过 ormer 查询引擎的 SELECT 语句如果目标模型有deleted_at字段就自动追加WHERE deleted_at IS NULL。业务层完全不需要手动写“未删除”条件二次开发时也不会漏掉删数据校验。软删除本身不复杂复杂的是唯一约束。比如订单表中order_no字段设置了uniqueTrue软删除后再次插入相同订单号会撞唯一约束。常见做法是把唯一约束改成组合唯一索引(order_no, deleted_at)这样不同删除时间的数据彼此隔离。这个坑我最早是在生产环境里踩到的恢复一条误删数据时插入同样订单号直接失败。另一个容易忽略的是软删除字段只对查询生效对UPDATE条件未必生效。如果你使用了全局过滤器一定要确认它也覆盖了更新操作的WHERE条件不然会出现“能查出来但更新报错”的奇怪状态。5. 排查手册跑 ormer 前后最容易翻车的五个位置5.1 循环导入导致的模型注册失败现象启动项目时报ImportError: cannot import name Order或者模型注册顺序不对数据库表始终少几张。原因模型 A 和模型 B 存在外键关系两个模块在models/__init__.py里互相导入Python 解释器处理到一半时有一方还没加载完。ormer 的模型注册依赖元数据导入顺序错了后续表映射就乱了。解决不要在models/__init__.py里直接from order import Order、from customer import Customer这种显式互引。改成在ormer_core的注册入口统一加载或者把关联关系写成一个单独模块例如models/relations.py。我习惯的做法是让模块内部不引用其他模块只声明本地字段关系定义集中到外层注册时补充这样既避开循环导入也方便查看整体结构。5.2 连接池耗尽数据库连接被占满现象接口偶发超时日志里出现cant connect to database重试一两秒又恢复数据库服务端排查时连接数处于高位。原因连接池配置不合适或者业务代码打开会话后没有释放。如果每个请求都调用session.add()却不close()连接不会被归还到池子。解决先看配置里的pool_size是否过小常见个人电脑设置 5 就行了生产环境 1020 起步还要配max_overflow。更关键的是用完关闭会话在 Web 框架拦截器或异步任务的 finally 块里统一调用session.close()。我一般会在请求入口处用中间件包一层try/finally保证异常时也会正常归还连接。5.3 事务回滚后 session 还残留着旧数据现象事务里先插入一条数据随后业务逻辑判断失败主动raise但在外层except里重新查询这张表发现脏数据还能查到。原因事务回滚只撤销了数据库端的变更当前 session 对象里还缓存着未提交的实体。ORM 的会话缓存没有跟着回滚刷新再次查询优先命中缓存。解决事务回滚后调用session.expire_all()或session.clear()强制下次查询重新从数据库读取。如果你的业务代码在事务外还要用同一个 session这点必须处理。我在 ormer 的源码里看到事务上下文管理器在__exit__里做了回滚后的缓存清理但是二次开发时如果自己写事务块很容易漏掉这个操作。5.4 迁移一周后表结构跟模型代码对不上现象本地跑得好好的测试环境突然报no such column: deleted_at检查数据库发现表结构还是旧版本。原因ormer 默认不会自动把模型字段映射成 DDL除非你调用了迁移引擎。很多人图省事直接让create_all建表字段变更后create_all又不会更新已有表。解决严格走migrations目录的变更脚本。每新增一个字段或者改约束就加一个版本脚本例如003_add_delivery_type.py。如果团队规模小也可以接受手动用外接工具改表但一定要把变更记录同步到代码库不然项目换个人就全凭经验猜了。5.5 时区问题本地保存created_at跟生产查出来差 8 小时现象数据库里存的时间是东八区接口返回时客户端看到的是世界协调时或者查询当天订单时边界少算一个小时。原因数据库引擎、连接参数和应用服务器三处时区不一致。auto_now_add生成的是 Python 侧的当前时间如果连接串里没指定时区数据库写入时会按它自己的 session 时区转换。解决在 ormer 的配置模块里统一设置时区例如timezone: Asia/Shanghai同时让应用层只存取时间戳展示层的时区转换交给前端。另外存储时间类型尽量用带时区的时间类型不要用字符串拼的本地时间字符串在排序和跨时区计算时就是个灾难。6. 二次开发改造思路给 ormer 加一个多租户插件如果要在这个项目上做插件化改造我建议从plugins目录切入。整个改造思路是不改动ormer_core的查询逻辑只注册一个全局过滤器。下面是一个多租户插件的最小实现# plugins/tenancy.py from ormer_core import hooks from contextvars import ContextVar current_tenant ContextVar(current_tenant, defaultNone) hooks.register(before_select) def add_tenant_filter(query, model_class): tenant current_tenant.get() if tenant and hasattr(model_class, tenant_id): query query.filter(model_class.tenant_id tenant) return query这个插件利用 Python 的ContextVar保存当前请求的租户 ID每个 Web 请求进来时调用current_tenant.set(tenant_id)退出时重置。之后所有模型查询都会自动带上租户条件。这里要注意两点一是必须让筛选字段名在不同模型中统一为tenant_id二是如果模型不想参与租户隔离可以约定一个内部标记跳过。我实际改造过类似系统发现最有价值的不是上面这些代码而是先给项目补一套“查询测试”。每次新增插件我就用一段固定数据跑一遍单测确认哪些查询被全局过滤器影响、哪些没有。这样能让插件行为透明化不至于上线后才发现某个统计报表把其他租户的数据也带了出来。老实说ormer 不是那种打开就能跑的现成框架它更像一份能让你学会“怎么把数据访问层写好”的源码笔记。我拆完这个项目后最深的体会是数据层的坑不会消失只会转移。从那以后我每次改数据访问层都会强制把“事务边界、软删除、N1 查询、连接池释放”过一遍清单这套习惯直接来自拆 ormer 的过程。希望你也能从这个项目里找到适合自己业务的改造起点。本文还有配套的精品资源点击获取