项目上线第二周凌晨三点运维在群里扔了一张截图数据库连接数打满业务全线超时。我打开代码仓库一看果不其然——几十个psycopg2.connect()散落在各个业务函数里有人写了 close有人没写还有人在except分支里直接 pass。这不是我第一次见到这种场面尤其国产数据库刚开始接入团队的时候大家都习惯性地拿它当普通 SQL 库裸连觉得能跑通就行。结果上线就被教育了。这篇文章就聊聊我是怎么用 Python FastAPI 给金仓 KingbaseES 搭了一套能扛住线上流量的 API 服务。核心思路很明确淘汰裸 SQL 直连用连接池 ORM 参数绑定 统一会话管理把数据库访问收敛成一套可观测、可管控的 API。内容会覆盖连接池参数怎么调、FastAPI 的依赖注入怎么组织、上线前要补哪些安全课以及我在金仓实测过程中踩过的一些坑。如果你正准备把 KingbaseES 接进微服务体系这篇应该能帮你少走不少弯路。1. 先说说为什么我决定淘汰裸 SQL 直连1.1 裸连最要命的三种死法第一种死法是连接泄漏。这是最常见的也是最疼的。业务代码里写了conn psycopg2.connect(...)执行完 SQL 之后呢好一点的记得conn.close()但异常路径往往没人管。一旦cursor.execute()抛了异常后面的 close 根本走不到连接就这么挂在数据库侧。Python 的垃圾回收虽然最终会回收对象但回收时机不可控在高并发下数据库的连接数会在几分钟内被消耗殆尽。金仓默认的max_connections通常就几百根本经不起这种漏法。第二种死法是 SQL 注入。裸连时代大家最喜欢用字符串拼接写查询cursor.execute(fSELECT * FROM users WHERE name {name})name 是前端传过来的查询参数一旦有人传 OR 11整个查询条件就形同虚设用户表的数据全暴露了。更进一步的注入语句甚至可以拖库、删表。金仓的协议层和 PostgreSQL 一样支持参数化查询但裸 SQL 写多了人就会变懒总觉得就一个查询而已拼一下没事。这种侥幸心理是线上事故的温床。第三种死法是事务边界混乱。不少人在裸连代码里根本没开过事务也没设置过 autocommit结果一个事务从第一条查询开始一直挂到连接被回收。事务长时间不提交持有的锁就长时间不释放一旦多个请求并发操作同一张表锁等待超时就是家常便饭。更隐蔽的问题是有些人会在一个事务里穿插调用外部 HTTP 接口事务挂十几秒甚至几十秒数据库侧的锁和连接都被拖垮。1.2 除了崩溃裸连还会埋下哪些雷崩溃是一方面代码层面的隐性成本更不容小觑。SQL 字符串散落在业务函数里问题在于不可统一治理。UUID 生成方式有的地方写gen_random_uuid()有的地方用 Python 的uuid4()生成字符串再拼进去风格完全不一致。等你想统一规范的时候只能全局搜索手动改漏一个就是事故。另外裸连是没有可观测性的。谁在什么时间执行了什么 SQL、花了多久、有没有慢查询这些问题在一堆connect execute的代码里全是盲区。上线之后想排查性能问题只能去翻数据库侧的日志效率极低而且查到的往往是聚合后的数据根本定位不到具体接口。更要命的是表结构变更的连锁反应。没有 ORM 模型没有迁移脚本DBA 在库里加了一个字段业务代码里所有相关的 SQL 都要跟着改。漏改一处接口就挂一处而且往往是运行到那行才报错线上事故就这么来的。所以我的态度很明确业务代码里不要直接写裸 SQL把数据库访问抽象到 ORM 连接池 统一会话管理这一层。这不仅是技术洁癖是保命。2. 环境准备让 Python 先和金仓正常对话2.1 金仓的默认配置与连接前提KingbaseES金仓数据库是成熟的关系型数据库产品和 PostgreSQL 的兼容度很高同时支持 PG 和 Oracle 两种兼容模式。要接它先得搞清楚几个默认参数。默认端口是 54321注意不是 PostgreSQL 的 5432这是最容易踩的第一个坑。默认超级用户是 system默认数据库是 test。安装完成之后我建议第一件事就是改掉 system 密码然后创建业务专用账号给一套最小权限这个习惯后面会讲到。连接之前还有一处经常被忽略服务端的监听地址。默认配置下金仓可能只监听了127.0.0.1你在本机用工具连没问题但应用服务器一访问就超时。这需要在kingbase.conf里把listen_addresses改成应用可达的 IP 或0.0.0.0然后重启数据库服务。注意改配置文件之前先备份改完之后用SELECT * FROM pg_settings WHERE name listen_addresses;确认生效。2.2 驱动选型psycopg2 还是官方驱动这一步很多人纠结。我的结论是如果你的金仓跑在 PG 兼容模式下直接用psycopg2-binarySQLAlchemy这是最省事、最稳的一条路。因为金仓实现了 PostgreSQL 的线协议psycopg2 完全能正常通信SQLAlchemy 的postgresql方言也能直接复用不需要写任何自定义方言。官方确实也提供自己的驱动但 Python 生态下官方驱动的文档丰富度、社区案例、踩坑资料都远不如 psycopg2。你自己写代码的时候遇到问题搜 psycopg2 能搜到一堆答案搜金仓的 Python 驱动就只能靠官方文档。除非你明确使用的是 Oracle 兼容模式而且业务里大量使用 Oracle 特有语法否则没必要给自己找麻烦。我项目里用的是 PG 兼容模式SQLAlchemy 连接串长这样postgresqlpsycopg2://your_user:your_password10.0.0.12:54321/your_db如果你的环境 SSL 是开启的可以再加上?sslmoderequire参数。这个按公司安全规范来不强求。2.3 连不上的时候按这个清单排查我在接金仓的过程中把常见的连接问题整理成了一张排查表你直接照着顺序查就行现象可能原因排查方向connection refused端口不对或者服务没起确认端口是 54321ps -ef | grep kingbase看进程password authentication failed密码加密方式或账号密码不匹配检查kingbase_hba.conf里的认证方式确认账号密码database xxx does not exist库名写错用超级用户执行\l列出所有数据库could not translate host name连接串格式错误检查 URL 里是否有特殊字符、多余空格connection timed out服务端只监听了本机改kingbase.conf的listen_addresses后重启有一回我排查了半天最后发现是应用服务器上没装psycopg2-binary直接报ModuleNotFoundError跟数据库一点关系都没有。所以先把 Python 环境里的驱动装好、在命令行里跑通一条最简单的SELECT 1再往上搭框架。3. 连接池给数据库加一道缓冲闸3.1 为什么非要连接池数据库连接的建立不是免费的。每一次 TCP 三次握手、认证握手、会话初始化算下来至少几十毫秒。如果每个请求都新建连接这几十毫秒就是纯浪费而且在高并发下会放大到灾难级别。更关键的是数据库侧对连接数是硬限制的金仓的max_connections参数决定了你最多有多少条并行会话。没有连接池你的应用就像一辆没有减震的车路稍微颠一点就散架。打个比方裸连相当于每次办业务都新开一个银行柜台办完就关连接池是固定开好几个柜台业务来了排队办理办完柜台还在下一个人继续用。柜台的数量可控排队机制透明运维也心里有数。3.2 SQLAlchemy 连接池的参数到底怎么设SQLAlchemy 自带的连接池实现已经足够成熟不需要额外引入连接池组件。你只需要在create_engine的时候把参数配好from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base DATABASE_URL postgresqlpsycopg2://user:pass10.0.0.12:54321/your_db engine create_engine( DATABASE_URL, pool_size10, max_overflow20, pool_pre_pingTrue, pool_recycle1800, echoFalse, ) SessionLocal sessionmaker(bindengine, autoflushFalse, autocommitFalse) Base declarative_base()这里每个参数背后都有讲究不是随便填的pool_size10连接池保持的最小连接数。应用启动后池子里会预创建一部分连接备用。max_overflow20当池子里的连接被全部借走时最多额外新建 20 条临时连接用完销毁。这个参数决定了你的应用能扛的瞬时峰值。pool_pre_pingTrue每次从池子里借连接之前先发一个轻量 ping 确认连接还活着。数据库重启、网络闪断之后池子里的连接可能已经失效这个参数可以避免你拿到一条假连接然后执行 SQL 报错。pool_recycle1800连接超过 1800 秒30 分钟后强制回收重建。因为数据库侧可能有自己的连接超时机制长时间不活动的连接会被服务端断开客户端不知道等到用的时候才发现连不上了。定期回收能避免这种情况。autoflushFalse关闭自动 flush避免在查询时意外触发未提交的写操作。autocommitFalse显式控制事务所有写操作必须在 commit 之后才真正落库避免隐式事务的诡异行为。3.3 池参数要与 worker 数量联动很多人在这一步栽了跟头单看每个 worker 的池子参数没问题但一上多 worker 就炸了。因为 uvicorn 每个 worker 进程都是独立的各自维护一套连接池。假设你起了 4 个 worker每个 worker 的pool_size max_overflow 30那全局最大连接数就是 4 × 30 120。如果数据库的max_connections只有 100压测一起来直接报too many clients already。所以参数设计一定要做一道简单的乘法预估最大连接数 worker 数量 × (pool_size max_overflow)再算上运维工具、其他服务的连接留 20% 到 30% 的余量。比如数据库max_connections 300你规划所有应用合计最大连接 200那每个 worker 的池子上限就要控制在合理范围内。我给团队的建议配置是pool_size5, max_overflow10这样每个 worker 最多 15 条连接4 个 worker 就是 60 条数据库完全扛得住。后续如果发现连接不够用优先优化 SQL 而不是盲目调大池子——池子越大数据库负担越重不是长久之计。4. FastAPI 层把数据库访问收敛成接口4.1 依赖清单与项目结构依赖方面用到的核心包不多就这些fastapi uvicorn[standard] sqlalchemy psycopg2-binary pydantic pydantic-settings项目结构我建议这样拆简单清晰后续加模块也不会乱app/ ├── main.py # FastAPI 实例、路由注册、异常处理器 ├── config.py # 配置读取数据库连接串、超时时间等 ├── database.py # engine、SessionLocal、get_db 依赖 ├── models/ # SQLAlchemy ORM 模型 ├── schemas/ # Pydantic 模型请求体、响应体 ├── routers/ # 业务路由 └── core/ ├── exceptions.py # 自定义异常 └── logging.py # 日志配置config.py里用pydantic-settings读取环境变量连接串不要硬编码在代码里。本地开发用.env文件生产环境用环境变量注入这是最基本的要求。4.2 get_db 依赖注入是消灭连接泄漏的关键FastAPI 的依赖注入系统是这套方案里最核心的一环。你只需要写一个生成器函数from app.database import SessionLocal def get_db(): db SessionLocal() try: yield db finally: db.close()每个请求进来的时候FastAPI 会调这个函数从连接池里借一条会话给路由用请求结束的时候finally块保证db.close()必然执行把会话还回连接池。注意这里用的是finally不管路由函数是正常返回还是抛异常close 都会执行。这就在机制上彻底解决了裸 SQL 时代的连接泄漏问题——你不需要在业务代码里写任何 close框架替你兜底。FastAPI 的这个模式还有一个隐形好处测试的时候可以覆盖get_db依赖传入一个测试数据库会话不用改任何业务代码。这一点在写单元测试的时候特别香。4.3 ORM 模型与路由实战以一张用户表为例ORM 模型长这样from sqlalchemy import Column, String, DateTime, func class User(Base): __tablename__ users id Column(String(36), primary_keyTrue, defaultgenerate_uuid) name Column(String(64), nullableFalse) email Column(String(128)) created_at Column(DateTime, server_defaultfunc.now())generate_uuid是应用层生成 UUID 字符串的函数用 Python 的uuid.uuid4().hex就行。这里我不建议依赖数据库生成 UUID因为不同兼容模式下行为有差异应用层生成更可控。路由层配合Depends(get_db)使用from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session router APIRouter(prefix/users, tags[users]) router.get(/{user_id}, response_modelUserOut) def get_user(user_id: str, db: Session Depends(get_db)): user db.query(User).filter(User.id user_id).first() if not user: raise HTTPException(status_code404, detailuser not found) return user这里有一个很多人会纠结的问题路由函数应该写async def还是普通def我的实测经验是数据库 IO 密集的接口用普通def就够了。FastAPI 会把同步函数丢到线程池里执行不会阻塞事件循环。而如果你用async def但内部还是调 SQLAlchemy 的同步查询那反而是阻塞了事件循环性能更差。除非你用asyncpg或SQLAlchemy async那一套否则老老实实用同步def就好。4.4 Pydantic 校验如何顺手解决 SQL 注入Pydantic 模型做请求体校验是这层方案里另一个关键点。它不光是做类型转换还相当于给所有入参做了一道白名单过滤from pydantic import BaseModel, Field, EmailStr class UserCreate(BaseModel): name: str Field(..., min_length1, max_length64) email: EmailStr router.post(/users, response_modelUserOut, status_code201) def create_user(payload: UserCreate, db: Session Depends(get_db)): user User(namepayload.name, emailpayload.email) db.add(user) db.commit() db.refresh(user) return user有了 Pydantic 的字段类型和长度约束请求体里想传个超长字符串、非邮箱格式的 email直接在入口就被拦下根本进不到数据库层。再加上 SQLAlchemy 的查询写法User.id user_id永远生成带占位符的参数化 SQL不会把用户输入拼进 SQL 字符串SQL 注入这个口子就彻底堵死了。这也是我说别再裸 SQL的最硬核理由不是让你不用 SQL而是让你别手拼 SQL。5. 上线前必须补的课异常、日志与安全5.1 全局异常处理器不能只返回堆栈生产环境最忌讳的是把 Python 堆栈直接打到 HTTP 响应里。一方面泄露代码结构给攻击者另一方面对调用方毫无意义。我建议在main.py里注册全局异常处理器from fastapi.responses import JSONResponse from sqlalchemy.exc import OperationalError app.exception_handler(Exception) async def unhandled_exception_handler(request, exc): logger.exception(unhandled error: %s, exc) return JSONResponse( status_code500, content{detail: internal server error}, ) app.exception_handler(OperationalError) async def db_error_handler(request, exc): logger.error(database error: %s, exc) return JSONResponse( status_code503, content{detail: database unavailable}, )这里面的关键是分层数据库相关的异常返回 503让调用方知道是后端依赖出问题了可以触发重试或者熔断其他未知异常统一返回 500错误细节只进日志不进响应。日志里要带request_id方便线上追踪一次请求的完整链路。FastAPI 的middleware里生成一个request_id塞到日志的 context 里排障的时候会感谢自己当初这一点设计。5.2 慢查询日志给每一条 SQL 计时裸 SQL 时代你根本不知道哪条 SQL 慢。用了 SQLAlchemy 之后可以用事件监听给每一条 SQL 计时from sqlalchemy import event from sqlalchemy.engine import Engine import time event.listens_for(Engine, before_cursor_execute) def before_cursor_execute(conn, cursor, statement, parameters, context, executemany): conn.info.setdefault(query_start_time, []).append(time.time()) event.listens_for(Engine, after_cursor_execute) def after_cursor_execute(conn, cursor, statement, parameters, context, executemany): total time.time() - conn.info[query_start_time].pop() if total 1.0: logger.warning( slow query: %.2fs | %s | %s, total, statement, parameters, )超过 1 秒的 SQL自动打 WARNING 日志带上完整的 SQL 内容和参数。这样慢查询不需要去翻数据库日志直接看应用日志就能定位。我在项目里还会把这类日志单独输出到一个独立的日志文件配合采集系统做告警超过阈值直接通知到值班群。另外要设置查询超时。SQLAlchemy 执行层面可以配合psycopg2的options参数或者语句级超时来兜底避免某条 SQL 卡死导致线程池被占满。比如在连接串里加上options-c statement_timeout1000010 秒没执行完的 SQL 会被数据库侧强制终止应用侧收到异常后走 500 或 503 返回不会无限挂起。5.3 最小权限与网络白名单代码层面的防护做完还有几件运维层面的功课不能省。我见过太多项目业务代码里用的还是 system 超级用户连着数据库。一旦代码出漏洞或者连接串泄露攻击者拿到的就是数据库的最高权限后果不堪设想。正确的做法是单独建业务账号只授予业务库的必要权限比如SELECT、INSERT、UPDATE、DELETE绝不授DDL权限表结构变更走独立的迁移账号和业务运行账号分离kingbase_hba.conf里限定应用服务器 IP 段访问其他来源一律拒绝生产环境把 SQLAlchemy 的echo参数设为False避免 SQL 和参数打印到日志文件里API 网关层加请求频率限制防止恶意刷接口把数据库打满。这些措施单独看都很简单但组合起来就是一套纵深防御。不要等出事了才补。6. 压测与排障我在金仓实测中踩过的坑6.1 连接数被压测打满的一次复盘有一次我给这套 API 做压测500 并发一上去数据库连接数瞬间飙到 200 多直接报too many clients already。第一次遇到这个报错的时候我第一反应是数据库的max_connections不够后来查了才发现完全是自己配置的问题。复盘后有几个根因。一是max_overflow给得太大瞬时峰值时连接被大量创建二是当时起了 8 个 uvicorn worker每个 worker 的池子上限是 30理论上限 240远远超过了数据库的max_connections设置。这个问题在压测之前没有算过账上线后必然炸。解决办法是重新计算参数数据库max_connections300业务预留最多 180 条连接uvicorn 缩到 4 个 worker每个 worker 的pool_size5, max_overflow10理论上限 60 条加上其他服务的连接也远低于 300。改完再压测连接数曲线平稳不再出现打满的情况。这里也提醒一句max_connections不是改得越大越好它受内存和操作系统文件描述符限制。盲目调大数据库侧参数只会把压力往后端传递最终把整台数据库机器拖垮。6.2 金仓与 PostgreSQL 驱动的兼容性细节虽然金仓兼容 PG 协议但兼容不等于完全一样。我在实际使用中遇到过几个细节问题列出来给大家参考func.now()这种通用函数没问题但某些 PostgreSQL 特有的函数和操作符比如jsonb_set、ARRAY的特定操作需要先在测试环境验证一遍主键自增建议使用IDENTITY或者显式序列不要用 PostgreSQL 的SERIAL写法在某些兼容模式下类型映射会出现偏差字符串类型在 PG 兼容模式下用VARCHAR没问题但如果你的库跑在 Oracle 兼容模式下VARCHAR2和VARCHAR的行为差异要特别注意事务隔离级别默认是读已提交Read Committed这和 PG 一致但如果你用了 Oracle 兼容模式默认隔离级别可能有差异涉及并发正确性的接口要单独确认。这些差异不是不能用而是要提前测试。我的建议是接金仓的时候先把业务里涉及的数据类型、函数、操作符列一张清单在测试环境逐项跑一遍别等到上线了才发现某个函数行为不一致。6.3 事务边界不清引发的死锁还有一次线上死锁查了半天才定位。场景是这样的一个下单接口里先查用户余额再更新余额中间又调用了一个内部 HTTP 接口去查同一张订单表。两个事务互相等待对方的锁数据库死锁检测把它杀掉但业务方看到的是一堆莫名其妙的报错。根因是事务边界不清晰。一个请求里开了事务又在事务中间调用了外部接口外部接口又开了新事务操作同一批数据锁的获取顺序不一致就死锁了。后来我定了三条硬性规范全团队强制执行每个请求只在路由函数里开一个事务所有数据库操作都在这个事务内完成中间不允许调用外部 HTTP 接口长事务一律拆短把耗时的查询放到事务外事务内只保留必要的读写update操作尽量放在请求的最后让锁持有时间尽可能短。这三条规范写进团队的技术清单之后死锁问题再也没有出现过。我在搭这套 API 层的过程中最深的体会其实是金仓本身并不难连难的是很多人用能用就行的心态在写数据库访问代码。连接池、参数绑定、统一会话管理、慢查询日志、最小权限这些基本功补上之后金仓跑线上业务一点都不含糊。如果你的团队正准备接国产数据库或者已经在裸 SQL 的老路上挣扎可以从一个最小的连接池方案开始改起不用一上来就上很重的框架但不再裸连这件事越早做越好。