简介本资源是一个面向Python后端开发者与数据库应用工程师的轻量级PostgreSQL操作框架源码旨在解决原生psycopg2使用繁琐、事务管理分散、SQL与代码耦合度高等问题特别适用于中小型Web服务、数据工具开发及教学实践场景。压缩包共28个文件55KB含17个Python核心模块如db.py、sql_mapper.py、dbx.py等实现连接池、SQL映射、声明式事务、2个Markdown/RST文档含README与generator说明、2个批处理脚本deploy_g.bat等支持快速部署、1个XMLDTD组合person_mapper.xml与mapper.dtd构成MyBatis风格SQL外部化配置、以及LICENSE、.gitignore等工程必备文件。已有270人学习下载读者可直接获得一套结构清晰、遵循Python编码规范、支持SQL与逻辑分离、具备完整单元测试test/目录下含dbx_test.py等和日志增强log_support.py的可运行框架开箱即用便于二次开发与教学演示。 如果你写过一段时间 Python 操作 PostgreSQL 的代码大概率会有这种感觉psycopg2 功能确实强但每写一个模块连接、游标、提交、关闭这四件套就要重复一遍代码里到处是 try/except/finally事务边界全靠个人自觉。我自己的项目里曾经有个数据库操作模块累积到 600 多行真正跟业务相关的 SQL 可能不到一半剩下全是在处理连接和样板代码。这个项目就是针对这个痛点做的一次轻量封装。不是造一个完整的 ORM而是把日常操作 PostgreSQL 最频繁的那几个动作——连接管理、CRUD、事务、结果集映射——收敛成一套简洁的接口。基于纯 psycopg2 实现不引入额外重型依赖适合中小型项目、脚本工具和教学场景。文章会完整拆解框架的设计思路、核心模块的源码实现、踩坑记录和实操演示无论你是刚开始用 Python 连接 PostgreSQL还是想优化自己的 DAO 层都有可以直接抄走的代码。1. 为什么要自己封装设计思路和选型拆解1.1 原生 psycopg2 在实际项目中的痛点先还原一个最典型的场景。假设你要写一个查询用户信息的方法用原生 psycopg2 大概要这样import psycopg2 def get_user_by_id(user_id): conn psycopg2.connect( host127.0.0.1, port5432, userpostgres, passwordpostgres, dbnameapp_db ) try: cur conn.cursor() cur.execute(SELECT id, name, email FROM users WHERE id %s, (user_id,)) row cur.fetchone() if row: return {id: row[0], name: row[1], email: row[2]} return None except Exception: conn.rollback() raise finally: cur.close() conn.close()这段代码看起来还行一个方法也就十来行。但当你需要写十几个类似的方法问题就出来了连接信息在每个方法里重复写异常处理逻辑各写各的拿到的还是元组得手动按下标取字段一旦表结构字段调整这里要跟着改。更重要的是SQL 注入已经使用了参数占位符这值得表扬但如果团队里有人图省事直接拼接字符串风险立刻拉满。我做过一次统计在真实业务代码里纯连接和资源释放的样板代码能占到整个数据访问层的 30% 到 40%。这部分代码不产生业务价值但出了事背锅的往往是它——连接没关连接池耗尽事务没回滚全是从这堆样板代码里漏出去的。1.2 生态对比为什么没有直接用 SQLAlchemy有人会问SQLAlchemy 不是现成的轮子吗何必自己封装这里要说明一下我的选型逻辑。SQLAlchemy 核心非常强大但它的学习曲线不低。一个新手上来就要面对 Engine、Session、ORM 映射、declarative_base 这一堆概念稍微配置复杂一点还要理解 lazy load、eager load、session 的生命周期。如果项目只是简单的增删改查引入 SQLAlchemy 有点像用集装箱卡车去拉一箱牛奶能干但笨重。异步方案 asyncpg 也考虑过性能确实比 psycopg2 更好但异步代码的传染性很强——一旦数据库层用了 asyncpg整个调用链都得改成 async/await。项目里如果只是一个数据处理脚本引入异步反而增加了复杂度。所以最终选择在 psycopg2 之上做封装。psycopg2 是 Python 连接 PostgreSQL 的事实标准生态成熟、文档齐全、坑都被人踩过了。我要做的不是替换它而是把重复的部分抽出来把容易出错的部分标准化。框架的定位是“轻量封装层”不是新 ORM。1.3 框架设计的三条核心原则动手写之前我给自己定了三条原则后面所有代码都是围绕这三条来写的。第一零重型依赖。整个框架只依赖 psycopg2 和 Python 标准库保证在任何环境里都能快速跑起来。第二默认安全。SQL 语句里的值必须走参数化表名和字段名必须走白名单校验宁可代码多两行也不留注入的口子。第三事务显式化。业务代码里最容易出问题的就是对提交时机理解不一致封装之后的接口必须让开发者明确知道“这里是一个事务边界”而不是隐式地自动提交不然问题排查起来非常头疼。有了这三条后面的所有设计决策都变得很清晰。下面进入核心模块的实现思路。2. 核心模块从零实现连接管理到 CRUD 泛化2.1 配置模块单例模式与应用配置解耦配置写死在每个方法的连接串里是第一个要解决的问题。我实现了一个 Config 类负责从环境变量或配置文件加载数据库连接参数并且用单例模式保证整个进程里只有一份配置。import os from functools import lru_cache class DatabaseConfig: def __init__(self, **_overrides): self.host _overrides.get(host, os.getenv(DB_HOST, 127.0.0.1)) self.port int(_overrides.get(port, os.getenv(DB_PORT, 5432))) self.user _overrides.get(user, os.getenv(DB_USER, postgres)) self.password _overrides.get(password, os.getenv(DB_PASSWORD, postgres)) self.dbname _overrides.get(dbname, os.getenv(DB_NAME, postgres)) self.minconn int(_overrides.get(minconn, os.getenv(DB_MIN_CONN, 1))) self.maxconn int(_overrides.get(maxconn, os.getenv(DB_MAX_CONN, 10))) self.connect_timeout int(_overrides.get(connect_timeout, os.getenv(DB_CONNECT_TIMEOUT, 10))) classmethod lru_cache(maxsize1) def load(cls): return cls()lru_cache 在这里除了做单例还附带了一个好处只有传入新的配置对象时才会重新实例化测试代码里可以很方便地注入自定义配置不影响全局。配置里我特别加了 connect_timeout 这个参数。默认 10 秒意思是一旦数据库连不上最多等 10 秒就报错不会让进程无限阻塞。这个问题在真实生产环境里遇到过数据库负载高的时候连接请求排到几十秒业务线程全部卡住加了超时之后至少能快速失败让上层感知到问题。2.2 连接池封装参数选择和对 psycopg2 的连接管理每一次 connect 到 PostgreSQL都要经历 TCP 握手、认证协商这些过程开销远高于执行一条简单 SQL。如果业务请求比较密集连接池就是必需品。我选择了 ThreadedConnectionPool而不是 SimpleConnectionPool。区别在于线程安全。SimpleConnectionPool 是线程不安全的如果在 Flask、FastAPI 这类多线程环境下使用多个线程同时 getconn 可能拿到同一个连接很容易出现“连接被另一个线程占用”的诡异报错。ThreadedConnectionPool 内部加了锁每次 getconn 都保证是当前线程独占的连接。import threading import psycopg2 import psycopg2.pool from psycopg2.extras import RealDictCursor class Database: _instance None _lock threading.Lock() def __new__(cls, *args, **kwargs): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance super().__new__(cls) return cls._instance def __init__(self, configNone): self.config config or DatabaseConfig.load() self._pool psycopg2.pool.ThreadedConnectionPool( self.config.minconn, self.config.maxconn, hostself.config.host, portself.config.port, userself.config.user, passwordself.config.password, dbnameself.config.dbname, connect_timeoutself.config.connect_timeout, keepalives1, keepalives_idle30, keepalives_interval5, keepalives_count3, ) def get_conn(self): return self._pool.getconn() def return_conn(self, conn): self._pool.putconn(conn) def close_all(self): self._pool.closeall()注意这里的双重检查锁写法。在new里做线程安全的单例是为了保证多个业务线程共用同一个连接池实例不然每个线程各搞一个连接池maxconn 就形同虚设了。关于连接池大小的参数我的建议是不要盲目调大。maxconn 设置成 10 到 20 在绝大多数中小型应用里都够用。数据库连接是很贵的资源PostgreSQL 默认的 max_connections 通常也只有 100如果应用层每个实例都开 50 个连接两三个实例就把数据库的连接数吃满了。至于 keepalives 相关的参数是为了避免网络空闲时连接被中间设备断开我所在的网络环境里 MySQL 还好PostgreSQL 的长连接如果不配 keepalive经常隔几个小时就会出现 server closed the connection unexpectedly配上之后这个报错基本消失了。2.3 Repository 基类泛型 CRUD 的骨架设计配置和连接池解决了连接问题接下来是 CRUD 的泛化。我设计了一个 Repository 基类子类只需要指定表名和主键字段就能获得一套基础的增删改查能力。import psycopg2 from psycopg2.extras import RealDictCursor, execute_values class Repository: table_name primary_key id # 可选的字段白名单默认 None 表示全部信任 fields_whitelist None def __init__(self, db: Database None): self.db db or Database() classmethod def _validate_table_name(cls): if not cls.table_name: raise ValueError(table_name must be set) if not cls.table_name.replace(_, ).isalnum(): raise ValueError(invalid table name) def find_by_id(self, record_id): self._validate_table_name() conn self.db.get_conn() try: cur conn.cursor(cursor_factoryRealDictCursor) cur.execute( fSELECT * FROM {self.table_name} WHERE {self.primary_key} %s, (record_id,) ) return cur.fetchone() finally: self.db.return_conn(conn)这里的业务逻辑是查询结果用 RealDictCursor 返回每一行是一个字典字段名可以直接用不用再按下标取值。表名字段在类级别定义调用前先做校验确保不会拼入非法字符。值部分一律用 %s 占位符即便用户传入带引号的内容也只会被当作值处理不会被解释成 SQL 片段。为什么不支持动态表名参数因为动态表名在绝大多数场景下都是反模式——表结构是静态的查询模板也应该是静态的。硬要支持动态表名就意味着连接池里的每条 SQL 都不能复用执行计划还得分心处理表名注入的风险。用类继承的方式定义表名既清晰又安全。2.4 事务上下文管理器业务代码不再被事务吃掉数据库操作里最隐蔽的坑就是事务边界混乱。我见过不少代码在同一个连接上执行了多条 SQL有的提交了有的没提交最后数据处于一个难以解释的状态。所以我实现了一个事务上下文管理器进入 with 块时开启事务块内正常结束就提交抛异常就回滚无论什么情况连接最终都会归还到连接池。from contextlib import contextmanager class Database: # 前面已有代码...... contextmanager def transaction(self): conn self.get_conn() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: self.return_conn(conn)用法长这样with db.transaction() as conn: cur conn.cursor() cur.execute(UPDATE accounts SET balance balance - %s WHERE id %s, (100, 1)) cur.execute(UPDATE accounts SET balance balance %s WHERE id %s, (100, 2))这段业务代码保证了两点第一两个账户的资金变动在同一事务里要么都成功要么都回滚第二就算执行过程抛了异常连接也会在 finally 里归还不会出现连接池被占满的问题。这里需要提醒一个关键点事务期间拿到的连接一定不能在这个连接上再调用 commit 或 rollback否则会打乱事务上下文管理器的判断。简单说事务里只管执行 SQL把提交流程交给 with 块结束时的统一处理。这是我早期踩过的坑后面会详细展开。3. 完整源码解读关键代码逐段拆解3.1 项目结构总览封装完成后项目结构是这样的pg_framework/ ├── __init__.py # 对外暴露 Database, Repository ├── config.py # 配置加载单例 ├── db.py # 连接池 查询方法 事务管理 ├── repository.py # Repository 基类泛 CRUD └── utils.py # 分页、批量操作等辅助函数db.py 是整个框架的核心除了连接池管理我把查询方法也收敛到了这里。这样 Repository 基类可以更薄只需要拼 SQL 和调用查询方法。3.2 连接池与查询方法的实现细节db.py 里除了 get_conn、return_conn还封装了几个常用的执行方法。我逐个说明。def execute(self, sql: str, params: tuple ()): conn self.get_conn() try: cur conn.cursor() cur.execute(sql, params) conn.commit() except Exception: conn.rollback() raise finally: self.return_conn(conn) def query(self, sql: str, params: tuple ()): conn self.get_conn() try: cur conn.cursor(cursor_factoryRealDictCursor) cur.execute(sql, params) rows cur.fetchall() return rows finally: self.return_conn(conn)这里有三个坑值得单独说。第一个是 execute 里的自动 commit。我默认把单条 SQL 的执行当作独立事务处理执行完就提交这样符合大多数场景的直觉。但这里埋了一个伏笔如果是多条 SQL 需要原子性就必须用前面那个 transaction 方法不能用 execute 一条条调用否则第一条成功第二条失败数据就残缺了。第二个是 query 方法里没有 commit。SELECT 语句不需要提交而且如果连接是刚从连接池里取出来的前一个事务可能还没提交这时候 commit 反而会破坏状态。我当时就因为这个理解偏差写了一版“完美”的 query结果在事务里调用 query 时把外层事务的提交时机搞乱了。第三个是 get_conn 和 return_conn 的顺序。连接从连接池取出来后一定要保证在 finally 里归还。如果业务函数中间 return 了或者抛异常了finally 都会执行。代码里不能提前在 try 分支里 return否则 finally 虽然是执行的但如果你用的是“return 前手工归还”这种写法很容易漏掉异常路径。3.3 Repository 基类与分页查询Repository 基类的完整 CRUD 骨架可以进一步拆成几个方法来看。新增和更新我用了字段白名单机制避免把未知字段拼进 SQL。class Repository: def _check_fields(self, data: dict): if self.fields_whitelist is None: return data invalid set(data.keys()) - set(self.fields_whitelist) if invalid: raise ValueError(ffields not allowed: {invalid}) return data def insert(self, data: dict): self._validate_table_name() data self._check_fields(data) columns list(data.keys()) placeholders , .join([%s] * len(columns)) sql ( fINSERT INTO {self.table_name} f({, .join(columns)}) VALUES ({placeholders}) fRETURNING * ) rows self.db.query(sql, tuple(data.values())) return rows[0] if rows else None def update_by_id(self, record_id, data: dict): self._validate_table_name() data self._check_fields(data) assignments , .join([f{col} %s for col in data.keys()]) sql fUPDATE {self.table_name} SET {assignments} WHERE {self.primary_key} %s params tuple(data.values()) (record_id,) self.db.execute(sql, params) def delete_by_id(self, record_id): self._validate_table_name() sql fDELETE FROM {self.table_name} WHERE {self.primary_key} %s self.db.execute(sql, (record_id,)) def find_by_id(self, record_id): self._validate_table_name() sql fSELECT * FROM {self.table_name} WHERE {self.primary_key} %s rows self.db.query(sql, (record_id,)) return rows[0] if rows else None分页查询是我单独抽出来的方法因为分页太常用了而且两个数字一组很容易写错。我的实现是传入 page 和 page_size返回数据和总条数。def paginate(self, page: int 1, page_size: int 20, where: str , params: tuple ()): self._validate_table_name() where_sql f WHERE {where} if where else count_sql fSELECT COUNT(*) AS total FROM {self.table_name}{where_sql} data_sql ( fSELECT * FROM {self.table_name}{where_sql} fORDER BY {self.primary_key} DESC fLIMIT %s OFFSET %s ) offset (page - 1) * page_size total self.db.query(count_sql, params)[0][total] rows self.db.query(data_sql, params (page_size, offset)) return rows, total分页这里有个容易踩的坑OFFSET 和 LIMIT 必须放在 SQL 末尾而且 LIMIT 的占位符要在最后否则参数顺序容易错。另外这里的 where 参数是直接拼接进去的所以调用方必须传入已经用 %s 占位好的条件片段不能传用户输入的原样字符串。这个约束我在方法的 docstring 里写得很清楚但实际使用中还是有人踩雷所以后续版本我改成了完全基于字典构造条件的形式这里暂不展开。3.4 结果集的三种映射模式psycopg2 默认返回的是元组按下标取值可读性太差。我在框架里用 RealDictCursor 作为默认返回字典。除了字典模式其实还有 NamedTupleCursor返回的是命名元组字段访问用 row.name比 row[name] 更轻量。我测试过NamedTupleCursor 的初始化开销比 RealDictCursor 稍小在超大结果集的场景下差异会更明显。实际封装时我用 cursor_factory 参数做成可配置的默认 RealDictCursor谁想用 namedtuple 就在 Repository 子类里重写一个类属性。但核心原则是不要让上层业务代码感知到“我拿到的到底是元组还是字典”应该让结果在框架层就统一成一种方便调用的结构。psycopg2 的 Json 适配器也要提一下。如果表里有 JSONB 类型的字段查询返回时 psycopg2 会默认把 JSON 文本转成 Python 字典插入时要用 psycopg2.extras.Json 包装一下。这个适配在封装框架里是被隐藏的业务层不需要关心序列化细节但底层要保证正确。4. 实操上手从建表到真实业务查询4.1 环境准备与依赖安装先把环境跑通。这里要求 Python 3.8 以上版本PostgreSQL 9.6 以上。连接库我推荐安装 psycopg2-binary它自带编译好的二进制文件pip 直接就能装省去本地编译的麻烦。pip install psycopg2-binary如果是在生产环境部署建议用 psycopg2 源码包自己编译这样性能会好一点而且可以针对 CPU 指令集做优化。但对于绝大多数开发场景binary 版足够了。安装完成后验证一下import psycopg2 print(psycopg2.__version__)4.2 建立数据库配置和第一个表创建一个测试库和测试表。我在本地用 Docker 起了一个 PostgreSQL 实例命令行可以直接创建表CREATE DATABASE app_db; \c app_db CREATE TABLE users ( id SERIAL PRIMARY KEY, name VARCHAR(64) NOT NULL, email VARCHAR(128) UNIQUE NOT NULL, tags JSONB DEFAULT [], created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() );设计了一个 JSONB 字段 tags用来演示框架对 JSON 类型的处理。下面把框架接到这个库上。from pg_framework.config import DatabaseConfig from pg_framework.db import Database config DatabaseConfig.load() # 如果环境变量没设全也可以直接传参 # config DatabaseConfig(host127.0.0.1, dbnameapp_db) db Database(config)4.3 业务示例用户模块的增删改查定义一个用户仓储子类from pg_framework.repository import Repository class UserRepository(Repository): table_name users primary_key id fields_whitelist {name, email, tags} user_repo UserRepository(db)新增一个用户user user_repo.insert({ name: 张三, email: zhangsanexample.com, tags: [vip, beta], }) print(user) # {id: 1, name: 张三, email: zhangsanexample.com, tags: [vip, beta], created_at: ...}注意tags 字段传进去的是 Python 列表框架内部的 Json 适配器会自动转成 JSON 字符串存到 JSONB 列里。这个细节如果不处理直接传列表会报 DataError。查一下user user_repo.find_by_id(1) print(user[name])更新和删除同样简单user_repo.update_by_id(1, {name: 李四}) user_repo.delete_by_id(1)分页查询演示rows, total user_repo.paginate(page2, page_size20, wherename %s, params(张三,)) print(total) # 符合条件的总条数 print(rows)事务的用法with db.transaction() as conn: cur conn.cursor() cur.execute( INSERT INTO users (name, email, tags) VALUES (%s, %s, %s), (王五, wangwuexample.com, [normal]) ) cur.execute( UPDATE users SET email %s WHERE name %s, (wangwu_updateexample.com, 王五) )4.4 实测效果封装前后的代码量对比我用一个包含增删改查、分页、事务的 8 个方法模块做了对比封装后的代码量大约是原生写法的三分之一而且接口更统一错误路径更少。操作原生 psycopg2 写法行封装后写法行连接管理每个方法 6-8 行0框架层处理按主键查询15-20 行1 行调用新增18-25 行1 行调用更新20-28 行1 行调用分页查询30-40 行1 行调用事务操作15-20 行 手动提交回滚with db.transaction(): 包裹真实写业务时的体感差距更大。不用封装时每加一个数据表就要复制粘贴一堆样板代码复制粘贴多了就容易漏改连接串、漏写 rollback用封装之后加一个新表只需要继承 Repository、指定表名10 分钟内就能把 DAO 层跑通。5. 常见问题与排查技巧实录5.1 ERROR: too many connections 连接池耗尽这个问题我用框架之后遇到过几次表现形式是突然大量报 FATAL: sorry, too many clients already。排查步骤很有代表性。第一步看 PostgreSQL 的连接数SELECT count(*) FROM pg_stat_activity;第二步看具体是哪个应用占用的SELECT application_name, client_addr, count(*) FROM pg_stat_activity GROUP BY 1, 2 ORDER BY 3 DESC;定位到是应用服务的连接数涨到了上限。原因基本是两个一是框架里某条路径没有归还连接二是连接池 maxconn 设置得和数据源实际峰值不匹配。我这里的最终解法是把所有 get_conn/return_conn 的调用路径检查了一遍确保所有数据库查询都经过框架统一的 query/execute 方法而不是在业务里绕过框架直接 get_conn。因为一旦绕过框架连接归还的约定就失效了连接池很容易泄露。5.2 事务边界不清连接归还时机问题有个阶段性版本里我在事务里调用了一个独立函数这个函数内部又调用了 db.execute 来执行一条 UPDATE。表面看代码没错但实际运行时出现了诡异现象外层事务回滚了那条 UPDATE 的数据居然还在。原因在于 db.execute 内部用了独立的连接自动 commit。也就是说外层事务使用的连接和内部 UPDATE 使用的连接根本不是同一个内部那条 UPDATE 是独立事务提前提交了。这个问题的根本解法是调整设计所有需要参与同一事务的操作必须显式传入同一个连接。框架里我把 execute 方法增加了一个可选参数 conn如果调用方传了 conn就使用该连接执行且不自动提交把提交权交还给外层事务上下文管理器。这也是我说“事务边界必须显式化”的落地方式。5.3 LIKE 模糊查询的参数化写法写模糊查询时新手最容易犯的错是把通配符拼进参数里# 错误写法通配符会被当作值的一部分不会生效 db.query(SELECT * FROM users WHERE name LIKE %s, (f%{keyword}%,)) # 这条语句在 PostgreSQL 里实际匹配的是字面量 %keyword%查不到任何数据正确的参数化写法是用拼接符把百分号拼到 SQL 里db.query( SELECT * FROM users WHERE name LIKE % || %s || %, (keyword,) )这里要特别小心 PostgreSQL 的百分号转义。在 psycopg2 的参数化语句里SQL 文本中的 %s 是占位符所以想表达 SQL 里的百分比符号时需要用 %% 来转义。比如db.query( SELECT * FROM users WHERE name LIKE %%%s%%, (keyword,) )这两条语句效果一样但第二条容易被误读建议用第一条的写法。这个坑在写模糊搜索接口时很容易踩而且报错信息还不明显查出来的数据莫名其妙少只能在实践中靠经验积累。5.4 大批量数据导入的性能优化批量插入数据时如果一条条 execute不仅慢而且在连接池模式下会频繁取还连接放大网络开销。框架里我封装了一个 batch_insert 方法基于 psycopg2 的 execute_values 实现。def batch_insert(self, rows: list[dict]): self._validate_table_name() if not rows: return columns list(rows[0].keys()) placeholders (%s, %s, %s) # execute_values 会自动展开 sql fINSERT INTO {self.table_name} ({, .join(columns)}) VALUES %s values [tuple(r[col] for col in columns) for r in rows] conn self.db.get_conn() try: cur conn.cursor() execute_values(cur, sql, values, page_size1000) conn.commit() except Exception: conn.rollback() raise finally: self.db.return_conn(conn)实测下来10 万条数据用 execute_values 批量插入耗时比逐条 execute 减少 80% 以上。为什么差距这么大因为逐条插入每次都要等待数据库完成一次事务提交而 execute_values 会拼接成一条多行 VALUES 语句网络往返次数从 N 次降到 N/1000 次左右。这是纯数据库层性能优化效果立竿见影。6. 后续扩展方向与个人心得6.1 值得继续完善的几个能力这个封装版本目前覆盖了核心场景但真要长期演进还有几个方向值得完善。第一个是 QueryBuilder。目前 where 条件还是以字符串拼接为主虽然安全但有门槛。我计划做一个简单的条件构造器支持类似 filter(name张三, tags__containsvip) 的写法底层转换成参数化 SQL对业务方更友好。第二个是软删除和审计字段的自动填充。很多业务表都有 deleted_at、created_at、updated_at 这几个字段完全可以让框架在 insert 和 delete 时自动维护减少业务代码的重复。第三个是读写分离支持。如果数据库是一主多从架构框架可以在连接池层面区分读池和写池查询走从库写入走主库。目前框架是单连接池的扩展的话需要把 query 和 execute 的取池逻辑分开。第四个是整合 FastAPI 的依赖注入。把 db 实例挂到 FastAPI 的 request.state 里配合 Depends 自动管理连接生命周期这样接口层就不用关心连接释放了。6.2 我在这个项目里的几点体会这套框架前后改了三个版本从最早的全局单连接到后来引入线程池再到把事务和连接归还问题彻底理清每一步都踩了不同层面的坑。最深的体会是封装不要过度。一开始我试图把所有数据库操作都抽象成一套 DSL结果复杂度比直接用 psycopg2 还高维护成本爆炸。后来退回到“只封装高频重复动作”的原则框架瞬间清爽了。现在这个版本核心代码加上注释不到 500 行但已经覆盖了日常 80% 的需求。其次一个体会是事务和连接池必须放在一起设计。如果只做查询封装不处理事务边界业务代码很快就会出现“连接没归还”“事务提前提交”的问题。连接池不是简单的 get/put 两个方法它和事务上下文是紧密耦合的设计时一定要一起考虑。如果你准备拿这份代码去改造成自己的框架我的建议是先在真实业务场景里跑通一条完整链路再回头抽公共方法。反过来从抽象出发很容易设计出一堆没人用的接口又或者和业务脱节。这个版本的代码结构我已经在文章里完整拆解了照着思路改造成本很低花上一两个晚上就能搭起来。本文还有配套的精品资源点击获取