简介本资源是一个面向Python后端开发者与数据库应用工程师的轻量级PostgreSQL操作框架源码旨在解决原生 psycopg2 使用繁琐、事务管理分散、SQL与代码耦合度高等问题特别适用于中小型Web服务、数据工具开发及教学实践场景。压缩包共28个文件55KB含17个Python核心模块如db.py、sql_mapper.py、dbx.py等实现连接池、SQL映射、声明式事务、2个Markdown/RST文档提供快速上手指南与设计说明、2个批处理脚本支持一键部署与测试、1个XMLDTD组合定义SQL映射结构实现MyBatis风格的SQL外部化管理以及许可证、Git忽略规则等工程必需文件。已有270人学习下载读者可直接复用该框架的模块化设计——包括SQL模板解析、参数绑定、自动事务提交/回滚、日志增强支持及单元测试用例快速构建可维护、易扩展的数据库交互层显著降低重复编码成本并提升开发规范性。 以前自己写项目、带团队的时候最烦的一件事就是业务代码里到处是散落的数据库操作。今天psycopg2.connect写下连接明天换个业务再cursor.execute一遍后天处理异常时又在每个函数里重复写rollback。代码一旦多起来这种重复劳动不仅累人还特别容易出问题——连接忘关、事务没提交、参数拼接出错每一个都是线上事故的种子。后来我把这套基于Python的PostgreSQL操作封装抽出来写成了一套简单但不简陋的数据库操作框架用了一年多维护成本明显降了下来。这篇文章就把这套框架的设计思路、核心源码和踩坑经验完整分享出来希望能给正在被数据库操作代码折磨的人一个参考。整个框架的设计目标就四个字简单、实用。不需要像大型ORM那样搞一堆复杂的模型映射和Query Builder也不要像裸写psycopg2那样把每个细节都暴露给业务方。它要解决的是业务方最频繁的增删改查、事务管理、连接复用这些日常问题同时把最容易出错的连接生命周期和事务边界管好。这套封装我自己在多个项目里反复打磨过现在把完整的思路和源码级别的实现细节拿出来讲清楚。1. 为什么非要自己做一套数据库操作封装很多人会问直接用psycopg2不好吗官方文档写得很清楚网上教程也一堆为什么要再包一层我先说说我在实际项目里看到的痛点你就明白了。1.1 裸用psycopg2最让人难受的五个场景第一连接管理靠自觉。每次psycopg2.connect()之后用完必须close()但业务逻辑一复杂尤其是遇到异常分支connection和cursor经常漏关。在长驻进程里这种泄漏是温水煮青蛙等连接数到达上限整个服务瞬间雪崩。第二事务边界全靠手动。PostgreSQL的事务很强大但裸写时你要自己记得commit()还是rollback()。我见过太多人写完execute直接返回结果连接释放时自动回滚了都没发现数据根本没写进去。第三参数化查询容易翻车。虽然psycopg2支持%s占位符但业务场景一多构造SQL时拼接字段、拼表名、拼排序条件非常容易绕过参数化退化成字符串拼接这是SQL注入或者说反斜杠地狱的温床。第四返回结果处理不统一。同一套代码里有时候要字典有时候要元组有时候只要第一行。每个开发者按自己的习惯写代码风格五花八门。第五异常处理支离破碎。每个函数都写一遍try...except来捕获数据库异常吗写完之后日志格式还不一样问题定位要靠grep各种奇怪的错误信息。1.2 我期望的封装到底是什么样我想要的封装不是把psycopg2的功能藏起来而是在它上面加一层面向业务的上下文。理想状态下业务代码只需要做三件事告诉框架“我要连哪个库”配置告诉框架“我要执行什么”SQL参数拿到结果或者异常其他的事情框架来管简单来说就是从“面向数据库API编程”变成“面向业务意图编程”。这也是我设计这个框架的核心思路把复杂留给框架把简单还给业务。1.3 明确边界这个框架不做什么虽然叫“框架”但它的野心没那么大。我不打算做一个完整的ORM。为什么不因为ORM的学习成本和隐式行为太多团队协作时会形成“黑魔法依赖”。这个封装只解决“数据库操作本身”不涉及数据模型定义、关联关系映射、迁移管理这些重活。你依然要自己写SQL这是刻意为之——SQL本身就是一种表达力很强的查询语言没必要为了“不用写SQL”而不写SQL。注意这里说的封装是介于psycopg2和完整ORM比如SQLAlchemy之间的一个轻量层。如果你需要复杂的模型关系映射和自动建表迁移那还是老老实实上SQLAlchemy吧。2. 框架的整体设计与目录划分动手写代码之前先把设计意图理清楚。我见过很多人一上来就写DBHelper类一个类几百行塞了几十个方法看着很唬人用起来简直灾难。我的设计原则很简单职责单一每个模块只干一件事。2.1 核心设计原则连接、执行、配置三者分离我把框架拆成三个核心模块配置模块负责加载和校验数据库连接信息不涉及任何数据库操作。连接模块负责创建、获取、释放连接管理连接池不涉及具体SQL。执行模块负责SQL执行、参数绑定、结果处理、事务控制不关心连接怎么来。它们之间的依赖方向是单向的执行模块依赖连接模块连接模块依赖配置模块。这样任何一个模块都可以单独替换和测试。2.2 目录结构小而美的包设计pg_simple/ ├── __init__.py # 对外导出主类 ├── config.py # 配置加载与校验 ├── connection.py # 连接管理与连接池封装 ├── executor.py # SQL执行器查询、写入、事务 ├── errors.py # 自定义异常体系 └── utils.py # 参数格式化、结果集处理等工具函数对着这个目录说下我的设计考量。config.py我用了dataclass相比字典的好处是有类型提示和默认值IDE能补全配置错了直接报错而不是运行时才炸。connection.py不对外暴露原始connection统一返回一个线程安全的包装对象。executor.py是业务方打交道最多的模块所有方法都围绕“SQL参数”展开。errors.py单独拆出来是因为业务方需要捕获特定异常比如唯一键冲突、死锁做重试或返回友好提示自定义异常体系能避免业务代码去认psycopg2的底层异常码。2.3 配置加载环境变量优先文件兜底配置模块看起来简单但有几个细节值得说。我要支持两种配置来源环境变量和配置文件。环境变量在容器化部署里是标准做法配置文件适合本地开发。# pg_simple/config.py from dataclasses import dataclass, field import os from typing import Dict, Any dataclass class DatabaseConfig: host: str 127.0.0.1 port: int 5432 database: str postgres user: str postgres password: str min_conn: int 1 max_conn: int 10 connect_timeout: int 5 pool_timeout: int 5 application_name: str pg_simple classmethod def from_env(cls) - DatabaseConfig: 从环境变量加载配置变量前缀为 PG_ return cls( hostos.getenv(PG_HOST, cls.host), portint(os.getenv(PG_PORT, cls.port)), databaseos.getenv(PG_DATABASE, cls.database), useros.getenv(PG_USER, cls.user), passwordos.getenv(PG_PASSWORD, cls.password), min_connint(os.getenv(PG_MIN_CONN, cls.min_conn)), max_connint(os.getenv(PG_MAX_CONN, cls.max_conn)), connect_timeoutint(os.getenv(PG_CONNECT_TIMEOUT, cls.connect_timeout)), pool_timeoutint(os.getenv(PG_POOL_TIMEOUT, cls.pool_timeout)), ) def to_dsn(self) - str: 转换为psycopg2需要的DSN字符串 return ( fhost{self.host} port{self.port} dbname{self.database} fuser{self.user} password{self.password} fconnect_timeout{self.connect_timeout} application_name{self.application_name} )这里有个小技巧connect_timeout5。默认情况下psycopg2建立连接时如果数据库不可达会一直等TCP超时有时候隔着防火墙能卡几分钟。设置了这个参数5秒连不上直接抛异常服务启动时就能快速失败而不是挂在那里“半死不活”。3. 核心模块实现从连接管理到CRUD这一部分是框架的心脏也是代码量最集中的地方。我把整个实现拆成三层讲连接层怎么管理连接和连接池执行层怎么封装SQL操作以及结果集处理怎么做。3.1 连接管理连接池的核心实现PostgreSQL的连接建立代价不小每次新建连接都要经过TCP握手、认证、内存分配高频业务下性能损耗明显。所以连接池是必须的。这里我没有引入SQLAlchemy的池用的是psycopg2.pool.ThreadedConnectionPool它足够轻量但在使用上有个要注意的点它不是绝对线程安全的获取连接时需要加锁。# pg_simple/connection.py import threading from contextlib import contextmanager from typing import Generator, Optional import psycopg2 from psycopg2.pool import ThreadedConnectionPool from psycopg2.extensions import connection as pg_connection from .config import DatabaseConfig from .errors import PoolExhaustedError, DBConnectionError class ConnectionManager: def __init__(self, config: DatabaseConfig): self._config config self._pool: Optional[ThreadedConnectionPool] None self._lock threading.Lock() self._local threading.local() def _ensure_pool(self) - ThreadedConnectionPool: if self._pool is None: with self._lock: if self._pool is None: try: self._pool ThreadedConnectionPool( self._config.min_conn, self._config.max_conn, self._config.to_dsn() ) except psycopg2.Error as e: raise DBConnectionError(f无法创建连接池: {e}) from e return self._pool def get_connection(self) - pg_connection: 从连接池获取一个连接并放到线程局部变量中 pool self._ensure_pool() try: conn pool.getconn() except Exception as e: raise PoolExhaustedError(f获取连接超时或失败: {e}) from e self._local.connection conn return conn def release_connection(self, conn: Optional[pg_connection] None) - None: 释放连接回连接池 pool self._ensure_pool() if conn is None: conn getattr(self._local, connection, None) if conn is not None: pool.putconn(conn) self._local.connection None contextmanager def connection_scope(self) - Generator[pg_connection, None, None]: 上下文管理器自动获取和释放连接 conn self.get_connection() try: yield conn finally: self.release_connection(conn)这段代码里有两个细节我要重点强调。第一个是线程局部变量。很多人在一个线程里同时开启多个事务如果共用全局连接事务会互相污染。我用threading.local()保证每个线程拿到的连接是独立的同一个线程内多次调用connection_scope()拿到的是同一个连接配合事务使用非常自然。第二个是释放连接时的异常处理。正常情况下putconn会把连接放回池子但如果连接已经断了比如数据库重启putconn可能抛异常。这里我虽然没展开写但你在实际使用中应该在finally里加一层try-except如果放回失败就销毁这个坏连接。我在后面的踩坑章节再细说。3.2 SQL执行器的骨架把参数安全地绑进去执行器是框架最重要的对外接口。它的核心能力有两个安全地绑定参数、灵活地获取结果。参数绑定用的是psycopg2提供的execute和mogrify永远不要自己拼接SQL值这是安全的底线。# pg_simple/executor.py from typing import Any, Dict, List, Optional, Sequence, Union from psycopg2.extras import RealDictCursor from .connection import ConnectionManager from .errors import DatabaseExecutionError, DataConflictError class Executor: def __init__(self, conn_manager: ConnectionManager): self._cm conn_manager def _get_cursor_factory(self, as_dict: bool): return RealDictCursor if as_dict else None def query(self, sql: str, params: Optional[Sequence[Any]] None, as_dict: bool True) - List[Any]: 执行查询返回结果列表 with self._cm.connection_scope() as conn: cursor_factory self._get_cursor_factory(as_dict) with conn.cursor(cursor_factorycursor_factory) as cur: try: cur.execute(sql, params) rows cur.fetchall() return rows except Exception as e: self._handle_error(e, sql, params) raise这里的_handle_error不是一个简单的日志而是要翻译数据库异常。比如psycopg2.errors.UniqueViolation要转成DataConflictError业务方才能精准处理。这个逻辑我单独抽出来了def _handle_error(self, e: Exception, sql: str, params: Any) - None: import psycopg2.errors as pg_errors if isinstance(e, pg_errors.UniqueViolation): raise DataConflictError(f数据冲突(唯一键): {e}) from e if isinstance(e, pg_errors.DeadlockDetected): raise DatabaseExecutionError(f检测到死锁请重试: {e}) from e raise DatabaseExecutionError(fSQL执行失败: {e}\nSQL: {sql}\n参数: {params}) from e3.3 写入操作与事务边界一个上下文管理器搞定写入操作最怕什么事务没有提交业务方不知道提交了但中途出错没有回滚。所以我做了一个事务上下文管理器让事务边界跟代码块走contextmanager def transaction(self): 事务上下文管理器正常结束自动commit异常自动rollback with self._cm.connection_scope() as conn: try: # 确保事务开启前不是 autocommit 状态 conn.autocommit False yield conn conn.commit() except Exception: conn.rollback() raise有了这个业务写起来就非常舒服with db.transaction(): db.update(UPDATE users SET balance balance - %s WHERE id %s, [100, 1]) db.update(UPDATE users SET balance balance %s WHERE id %s, [100, 2])这两条更新要么一起成功要么一起失败。不用担心手动commit忘了写也不用担心异常时忘了rollback。这个设计我认为是整个框架里价值最高的部分。3.4 查询结果集字典、元组和单值自由切换很多人纠结返回结果到底用什么类型。我观察下来业务方80%的场景需要字典因为字段名可读性高20%需要元组比如只需要一个计数、一个最大值。还有一小部分场景只需要单值比如SELECT count(*)。所以我提供三个层次的方法query()返回列表元素是字典或元组适合多行结果。query_one()返回一行或者None适合单行查询。query_value()返回第一个字段的值适合聚合结果。def query_one(self, sql: str, params: Optional[Sequence[Any]] None, as_dict: bool True) - Optional[Any]: rows self.query(sql, params, as_dictas_dict) if not rows: return None return rows[0] def query_value(self, sql: str, params: Optional[Sequence[Any]] None, default: Any None) - Any: rows self.query(sql, params, as_dictFalse) if not rows: return default row rows[0] if isinstance(row, tuple) and row: return row[0] return default这几个方法都做了None处理避免业务方每次都要判断if len(rows) 0。查询不到数据时返回None而不是抛异常这样行为更贴近业务直觉——查一个人查不到就是没有这个人不是系统出错了。4. 连接池的取舍线程安全与性能实测连接池是整个框架里最容易出问题、又最容易被忽视的部分。很多应用的性能瓶颈不是SQL慢而是连接池被耗尽新请求排队等连接。这一章我专门说说连接池的设计取舍和参数调优。4.1 为什么一定需要连接池PostgreSQL每建立一个新连接平均耗时在几十毫秒到几百毫秒之间取决于认证方式和网络环境。对于高并发接口如果每次请求都新建连接那数据库连接建立本身就会成为性能瓶颈。连接池是把空闲连接缓存起来复用能显著降低时延和数据库负担。另一个隐藏问题是连接数量的控制。没有连接池时如果应用疯狂并发连接数据库端会出现大量idle in transaction或active连接最终超出max_connections限制数据库直接拒绝服务。连接池相当于一个限流器把并发连接数框在一个可控范围内。4.2 线程安全使用连接池前必须知道的真相psycopg2.pool.ThreadedConnectionPool之所以叫“Threaded”是因为它的getconn和putconn内部有锁保护但连接对象拿到手之后不是线程安全的。也就是说你拿到了连接在同一个时刻只能被一个线程使用。如果你在多个线程同时用一个连接执行SQL会出各种奇怪的错误——比如“commands out of sync”。我用threading.local()来保证每个线程有且只有一个活动连接。这样设计的另外一个好处是同一个事务内多次有execute操作可以共用同一个连接因为事务和连接是绑定的。如果每次execute都从连接池拿新连接事务就没法跨多条SQL了。def execute(self, sql: str, params: Optional[Sequence[Any]] None) - int: 执行写操作返回受影响的行数 with self._cm.connection_scope() as conn: with conn.cursor() as cur: try: cur.execute(sql, params) return cur.rowcount except Exception as e: self._handle_error(e, sql, params) raise注意这里没有conn.commit()。为什么因为execute本身不决定事务边界。如果业务方在一个事务上下文中调用execute提交应该由事务块统一处理如果业务方没有开事务那每次execute就是一个隐式事务需要框架层面决定是否自动提交。这一点很容易踩坑我后面详述。4.3 连接池参数不是越大越好连接池参数主要有min_conn、max_conn、pool_timeout。很多人的第一反应是把max_conn调大觉得越大并发能力越强。但在PostgreSQL里连接是一个非常重的资源每个连接能占十几MB内存取决于work_mem等参数而且很多CPU密集查询会用满单核。我实测过一个场景应用实例4个每个实例连接池max_conn50总共有200个潜在连接。但数据库实例配置只有4核8GBmax_connections100。在压测时数据库直接内存爆掉。后来我把每个实例的max_conn降到20总并发连接数控制在80以内加上查询本身的优化整体吞吐量反而提升了因为数据库不需要花大量CPU去管理连接上下文切换。这里给出一个经验公式不是绝对的但可以当起点max_conn ≈ 数据库CPU核数 × 2 持久后台任务数如果单条查询很重比如大批量聚合max_conn还要再降。min_conn设置为2~3就够不用一开始就建立一堆空闲连接。5. 实测踩坑这些坑不踩一遍真不知道框架写出来只是第一步能不能在真实环境里扛住才是关键。我把自己在项目里踩过、排查过、解决的几个问题拿出来说说每一个都花了不少时间。5.1 事务与autocommit的坑隐式提交导致的迷惑行为psycopg2默认情况下连接是autocommitFalse的。也就是说你执行了一条INSERT它处于一个“未提交的事务”里当连接关闭时如果没提交数据就没了。这个现象有时候很隐蔽你在命令行工具里执行INSERT立刻能查到但在Python代码里同一条SQL执行完马上再SELECT却没有——因为连接释放时事务回滚了。我的解决方式是在框架层统一约束如果业务方使用transaction()上下文管理器那么进管理器时autocommitFalse退出时提交或回滚。如果业务方直接使用execute()而没有开启事务我内部在连接获取时保持数据库默认的autocommit行为让每条SQL自动提交这样业务方觉得“执行了就生效了”。这个设计需要实现为连接获取时做一次状态标记def _configure_connection(self, conn: pg_connection, autocommit: bool) - None: try: conn.autocommit autocommit except Exception: pass在execute这样的单条写操作中我显式设置autocommitTrue在transaction()上下文中设置autocommitFalse。这样两条路径互不干扰行为可预期。5.2 参数化查询的坑数字参数和JSON参数的处理psycopg2的参数化查询用%s占位符传入的参数会被转义成对应的类型。但有几个边界场景要小心场景一数字参数必须传Python数字类型不能传字符串。如果你传100字符串PostgreSQL会尝试把varchar类型转换成integer大多数时候能成功但在某些涉及类型推断的复杂SQL里会引发“operator does not exist: integer character varying”错误。所以框架层我在参数绑定前不做强制转换但在文档里明确要求业务方传正确的Python类型。场景二JSON字段参数。PostgreSQL的JSONB类型需要传入特殊对象如果直接传字符串有些驱动版本会把它当成varchar传给JSONB字段导致类型不匹配。psycopg2有一个Json适配器from psycopg2.extras import Json db.execute( INSERT INTO events (data) VALUES (%s), [Json({action: login, user_id: 123})] )我在框架的工具模块里封装了to_json辅助函数统一做转换这样业务方不用关心驱动层面的类型适配问题。5.3 大查询与游标的坑fetchall拉爆内存这个坑最隐蔽因为开发环境的测试数据量小永远暴露不出来。系统上线跑了一个月某个统计接口突然OOM排查发现是一条SELECT查了上百万行然后框架用fetchall()一次性拉回内存。百万行、每行几十个字段内存占用轻松上GB。后来我在框架里给query()方法加了一个可选参数fetch_sizedef query_iter(self, sql: str, params: Optional[Sequence[Any]] None, as_dict: bool True, fetch_size: int 5000): 分批获取结果适合大查询 with self._cm.connection_scope() as conn: cursor_factory self._get_cursor_factory(as_dict) with conn.cursor(nameserver_side_cursor, cursor_factorycursor_factory) as cur: cur.itersize fetch_size cur.execute(sql, params) for row in cur: yield row注意这里用了命名游标nameserver_side_cursorPostgreSQL会开启服务端游标让数据库分批次返回数据而不是一次性把所有结果发到客户端。改用生成器模式后内存占用从GB级别降到几十MB。5.4 连接泄漏最容易被忽视的隐性问题我见过不少团队使用psycopg2时连接泄漏表现是服务跑几天后数据库连接数莫名飙高最终数据库拒绝连接。最典型的原因就是代码里写了conn psycopg2.connect(...)但异常路径里没有close()。这个坑其实很好规避前提是使用框架的connection_scope()上下文管理器。它会保证在finally里释放连接。但如果你在业务代码里不小心调用了get_connection()拿走了连接而没有用上下文管理器那泄漏还是会发生的。所以我做了一个保护性的设计连接归还时检查并发数。如果并发数异常比如连续50次putconn时连接还处于事务未提交状态就在日志里打一个醒目的警告。def release_connection(self, conn: Optional[pg_connection] None) - None: pool self._ensure_pool() if conn is None: conn getattr(self._local, connection, None) if conn is not None: try: if conn.status ! psycopg2.extensions.STATUS_READY: # 连接不是空闲状态说明事务可能没结束 logger.warning(连接归还时状态不是READY可能存在未提交事务执行回滚) conn.rollback() pool.putconn(conn) except Exception: # 连接已损坏直接丢弃 pool.connection_collapse(0) self._local.connection None raise finally: self._local.connection None这个归还前的检查虽然简单但能拦住一大半的“拿连接不办事”问题。6. 日志与监控没有可观测性一切都是黑盒框架能跑通不叫完成能观测才叫可用。数据库操作如果出了异常你连是哪条SQL出的错、花了多久都不知道那排查问题就得靠猜。所以我把日志和监控也做进了框架。6.1 每条SQL的执行耗时与参数日志我给执行器的query、execute、transaction都加上了耗时统计和结构化日志。这样在排查慢查询时直接看日志就能知道哪条SQL慢慢在哪里。import time import logging logger logging.getLogger(pg_simple) def _log_sql(action: str, sql: str, params: Any, duration_ms: float) - None: # 参数可能很长默认截断前200个字符 param_preview repr(params)[:200] logger.info(SQL %s | duration%.1fms | sql%s | params%s, action, duration_ms, sql, param_preview)日志输出效果类似SQL query | duration3.2ms | sqlSELECT * FROM users WHERE id %s | params(1,) SQL transaction | duration12.8ms | sqlUPDATE users SET balance balance - %s WHERE id %s | params(100, 1)这个日志在生产环境里就是排查问题的第一手线索。每次看到某个接口耗时飙升去日志里搜对应的SQL基本能定位个八九不离十。6.2 慢查询告警约定超过阈值的SQL自动上报我还在框架里加了一个简单的慢查询检测。默认阈值是500毫秒超过这个时间会在日志里打WARNING级别。如果有监控系统可以直接接日志流如果只是个人项目至少在日志里能一眼看到。SLOW_QUERY_THRESHOLD_MS 500 def _check_slow(duration_ms: float, sql: str) - None: if duration_ms SLOW_QUERY_THRESHOLD_MS: logger.warning(SLOW SQL | duration%.1fms | sql%s | threshold%dms, duration_ms, sql, SLOW_QUERY_THRESHOLD_MS)在实际使用中这个阈值需要根据你的业务动态调整。一个在线交易接口的查询可能一秒执行几百次每次2毫秒那阈值设500ms就合适一个后台报表任务可能一条查询跑十几秒那阈值应设成5秒。不要把慢查询阈值设成一个全局常量最好在初始化框架时可以配置。6.3 连接池状态的监控指标连接池的健康度直接影响整个应用。我在ConnectionManager里暴露了三个指标pool_size当前池内总连接数。pool_used当前被业务占用的连接数。pool_available当前空闲连接数。这些指标配合日志输出能在连接池耗尽前提前预警。举个真实例子有一次我发现pool_available长期为0但pool_used经常打满也就是说所有连接都在被占用。排查后发现是一个后台线程持有了连接没有释放因为那个线程用了裸的get_connection()没走connection_scope()。如果监控能早几天发现这个问题就能少掉一次线上故障。7. 扩展与集成的思路让框架适应更多场景基础框架写完剩下的是让它更贴合复杂业务场景。这一章讲几个我已经用上的扩展点以及一些合理的后续方向。7.1 与FastAPI的集成依赖注入实现请求级数据库会话在Web框架里最常见的是把数据库会话绑定到请求生命周期。在FastAPI里我封装成一个依赖# webapp/dependencies.py from fastapi import Request from pg_simple.executor import Executor def get_db(request: Request) - Executor: executor: Executor request.app.state.db return executor这样每个路由都可以这样用app.get(/users/{user_id}) def get_user(user_id: int, db: Executor Depends(get_db)): return db.query_one(SELECT * FROM users WHERE id %s, [user_id])得益于前面的事务上下文管理器如果后续在路由里要组合多个写操作只需要在路由函数里使用db.transaction()即可事务边界和请求生命周期天然契合。7.2 读写分离的接入思路PostgreSQL的高可用部署里经常有主从架构读流量可以走从库。这个框架改动起来也不复杂。我设想是给Executor增加一个_read_conn_manager和_write_conn_manager在query()/query_one()/query_iter()这些只读操作里走读库连接池在execute()和transaction()里走写库连接池。要注意的是读写分离最大的痛点是主从延迟。如果你的业务刚写完数据立刻要读取比如创建订单后展示订单详情走从库可能因为延迟读不到。所以我在计划里保留了一个force_write参数在强一致场景下强制走主库。def query_one(self, sql, paramsNone, as_dictTrue, force_writeFalse): cm self._write_cm if force_write else self._read_cm with cm.connection_scope() as conn: ...这个改动本身不大框架的分层设计让它很容易扩展。7.3 多数据库实例的配置管理在微服务里你可能需要同时连接多个PostgreSQL实例比如一个业务库一个统计库。我的设计是让每个Executor绑定独立的ConnectionManager再在应用层维护一个实例注册表# pg_simple/registry.py from typing import Dict from .config import DatabaseConfig from .connection import ConnectionManager from .executor import Executor class DatabaseRegistry: def __init__(self): self._executors: Dict[str, Executor] {} def register(self, name: str, config: DatabaseConfig, **kwargs) - Executor: cm ConnectionManager(config) executor Executor(cm, **kwargs) self._executors[name] executor return executor def get(self, name: str) - Executor: return self._executors[name]这样业务代码统一用registry.get(primary)或registry.get(analytics)来获取执行器每个执行器独立管理自己的连接池互相不干扰。8. 把这次封装的得失做个复盘到这里整个基于Python的PostgreSQL数据库操作框架从设计、实现到踩坑、扩展都讲完了。最后说点我自己的体会。如果让我重新做一次这个封装我依然会选择“轻量”而不是“重量”。很多团队的数据库访问层复杂度是被撑大的——一开始想支持所有数据库结果每个数据库的方言差异都要适配一开始想做全自动的模型同步结果表结构变更频繁时反而成了枷锁。我的经验是先把手上的PostgreSQL用好性能调优做好比什么都强。框架本身的代码量不多但通过它约束了团队成员的代码习惯让项目的数据库代码风格统一了这才是最大的收益。有几个参数和设计点我建议你根据自己的场景调整连接池的max_conn不是越大越好事务的autocommit状态必须明确慢查询阈值要结合业务实际设置。另外永远不要相信“这个查询在测试环境很快”这个说法生产环境的数据量会放大所有你忽视的细节。这套框架的源码我已经在几个项目里跑过稳定性和性能都经受住了检验。如果你打算在你的项目里用我建议先跑一遍基础的单元测试把连接池大小和事务行为调成适合你业务的参数再逐步替换旧的数据库代码。如果有更好的设计思路或踩到了我没提到的坑欢迎交流。本文还有配套的精品资源点击获取