开篇先把一个很容易被带偏的点说清楚a2conn 这个包名字里那个 a2很容易让人以为它和 aio 系列一样是个异步库。我第一次看到也是这么以为的装上之后才发现它跟异步没有半点关系它的主要作用是帮你在 Python 里跟数据库建立连接而且是走那种需要经过跳板机转一跳的复杂网络环境。很多项目里数据库并不直接暴露在公网你本地代码想连上去常规做法是自己拼 SSH 隧道或者让运维开网络策略而 a2conn 把这一层封装掉了调用方只需要关心主机、用户、库名这些业务参数底层隧道怎么建立、怎么销毁它替你处理。这篇文章我会把这几天实际用它写脚本的完整记录整理出来包括它暴露了哪些核心方法、每个参数控制什么行为、用的时候有哪些反直觉的坑以及一个从零到跑通的真实案例。适合正在维护内网数据库、又不想在业务代码里写一堆隧道逻辑的 Python 开发者也适合刚接触这类工具、想快速上手的人直接抄作业。1. 内容整体设计与思路拆解1.1 为什么需要 a2conn先搞清楚它解决什么问题先说场景。很多公司的数据库分两层线上业务库在核心内网开发机在办公网两边网络不互通。开发要查数据常规办法是先用 ssh 登录到一台和内网互通的跳板机再在跳板机上执行数据库命令或者手动搭一条 SSH 隧道把内网端口映射到本地。手动隧道的问题是隧道生命周期得自己管理代码里要处理端口冲突、断了重连、多环境切换写出来的东西跟数据库操作逻辑缠在一起特别丑。a2conn 的思路是把建立连接这件事收敛成一个配置对象和一个连接对象。你预先声明好运维给你的跳板机地址、账号、目标数据库地址和库名然后调用它的 connect 方法去建立隧道和连接之后你的代码就聚焦在 SQL 操作上隧道细节被封装在内部。这对需要定期跑数据任务、临时写脚本抽数的人来说省下的是大量跟业务无关的样板代码。1.2 它和 SQLAlchemy、pymysql 这些库的边界新手最容易混淆的是a2conn 不是用来替代 pymysql 或 psycopg2 的。它更像是一个连接管理器负责建立和维护底层连接真正执行 SQL 的仍然是数据库驱动。官方文档里它其实支持 postgres 和 mysql 两大类具体用哪个驱动由你传进来的后端配置决定。打个比方pymysql 是水管本身负责把水流送过来a2conn 是水龙头和阀门的组合负责在哪接水、怎么接、开关顺序。你写代码时照样用 pymysql 的游标和查询方法只是游标的来源从自己连一下数据库变成了从 a2conn 的连接对象里拿。这个区分很重要不然你会觉得它提供的功能太少实际上它的定位就不是大而全的 ORM。1.3 被名字误导的异步印象这一点值得单独说。PyPI 上有很多 a2 开头或者 a 开头的库是异步实现的比如 aiohttp、asyncssh所以看到 a2conn 的第一反应是又一个异步数据库连接库。打开源码你会发现它内部用的是 paramiko 来处理 SSH 通道而 paramiko 本身是同步的。整个连接和查询流程完全同步执行跟 asyncio 没有关系。那这个 a2 到底什么意思我翻了它的源码和注释它其实是从一个内部项目拆出来的沿用了项目代号。这个教训是不要光靠名字推断库的特性用之前花两分钟 pip download 看一眼源码比什么文档都管用。2. 核心语法与参数拆解2.1 最核心的两个对象Config 和 A2Conna2conn 的使用基本围绕两个 import 展开from a2conn.config import Config from a2conn.connection import A2ConnConfig 是一个数据类职责非常纯粹把数据库连接需要的信息集中管理然后对外提供 get_engine 和 get_connection 两个方法。A2Conn 则负责实际的动作connect 和 run 这两个方法一个建立连接一个执行操作。我建议的使用方式是一个项目里所有环境开发、测试、生产都定义成不同的 Config 实例用名字区分这样代码里不会出现一堆散落的连接字符串环境切换只是一行赋值的事。2.2 Config 各参数逐项说明Config 构造的时候不是所有参数都要填。按我的经验可以把参数分成三组必填的、可选但有默认值的、以及某些特殊场景才需要的。整理成表如下参数名是否必填作用host必填数据库服务器地址不是跳板机地址user必填目标数据库用户名password选填数据库密码如果配置里没有会从 keyring 里取database选填目标数据库名port选填数据库端口默认根据后端类型自动判断Pg 是 5432MySQL 是 3306backend选填后端类型传字符 pg 或 mysql默认 pgconfig选填给底层驱动用的配置比如 pg 的话传 connect_timeoutssh_tunnel选填要不要启用 SSH 隧道默认 Falsessh_config选填隧道相关参数的集合见下文keyring选填是否使用系统密钥环默认 Falsemodel选填模型的层定义用于 SQLAlchemy 的 Base 绑定schema选填模式名有个反直觉的地方port 可以不填。我最早用的时候习惯把所有参数写全后来发现后端类型一旦定了默认端口就是对的多传反而容易传错。如果你数据库跑在非默认端口再显式指定 port 就行。2.3 ssh_config 参数跳板机配置的关键ssh_config 是 Config 里的嵌套结构它控制隧道这头的所有网络参数。如果你不需要走隧道这一整块都可以不理会。需要走的时候注意这几个参数名作用host跳板机 IP 或域名user跳板机登录用户password跳板机密码如果有密钥优先用密钥portSSH 端口默认 22很多公司改成别的端口这里要留意private_key私钥路径用密钥登录时填allow_agent是否允许使用 ssh-agent 里的密钥默认 False实测经验能用私钥别用密码。密码方式在自动化任务里会面临过期、被安全策略锁定的问题私钥配合 ssh-agent 是长时间跑批任务最稳的方式。allow_agent 这个参数一开始容易被忽略如果你本机 ssh-agent 里已经加了密钥把它设为 True 能省去写 private_key 路径的麻烦尤其是同一个密钥管多台机器的时候。2.4 connect 与 run 的调用约定先看一个最小的连接示例config Config( host10.20.30.40, userappuser, passwordYourPassword, databaseappdb, backendmysql, config{charset: utf8mb4} ) conn A2Conn() conn.connect(config) run_result conn.run(SELECT 1)注意 connect 方法在铺隧道时是懒加载思路只有实际执行 run 或获取连接对象时隧道才真正建立起来。我第一次用的时候connect 之后立刻检查隧道端口发现什么动静都没有一度怀疑代码写错了。这个行为本身是设计使然不用惊慌只要在 run 之后去确认即可。run 方法是对外最主要的入口。它接受一个字符串形式的 SQL或者是带参查询的元组列表。如果是多条带参数查询每个元组的结构是 (sql, params_list)params_list 里每一项是这条 SQL 的参数集合这样一次调用就能批量执行多条相似 SQL减少网络往返。下面这个例子展示了批量插入的典型写法rows [] for i in range(5000): rows.append((fINSERT INTO test_table (id, name) VALUES (%s, %s), (i, fname_{i}))) result conn.run(rows, query_size500)query_size 参数控制一次往数据库发送多少条默认 1000。如果你的 SQL 比较复杂或者单条数据量大调低这个值可以避免数据库端报 too many SQL variables 之类的错误。这个参数是我在使用中认为最值得调的一个对性能的改善是实打实的。2.5 get_engine 和 get_connection 的适用场景除了 runConfig 还提供另外两条取连接的路径engine config.get_engine() connection config.get_connection()get_engine 返回的是 SQLAlchemy 的 Engine 对象适合要配合 ORM 或者要使用事务、连接池的场景。get_connection 返回的是一个裸连接适合直接用原生 SQL、又不想引入 ORM 上层概念的场景。最典型的用法是在 FastAPI 或 Flask 项目里启动时创建一个全局 Config每个请求需要访问数据库时调用 get_connection 获取独立连接用完关闭。因为底层连接实际上是经由隧道转发的连接的生命周期和你处理隧道资源的方式是绑定的务必确保使用完调用 close 方法释放。3. 实操过程与核心环节实现3.1 环境准备与安装安装很简单直接用 pippip install a2conn它会自动带上 paramiko、SQLAlchemy、keyring 这些依赖。如果你只需要 Pg 或 MySQL 其中一种装完基础包就能用如果你两种都要确保本机有对应的驱动比如 psycopg2-binary 或 PyMySQL驱动不在基础依赖里需要自己补pip install psycopg2-binary PyMySQL这里有个小坑paramiko 依赖 cryptography而 cryptography 在一些老的 Python 3.6 环境上需要编译经常导致安装失败。我的建议是别在 Python 3.6 上折腾直接上 3.8 及以上少很多事。3.2 三个不同后端的完整配置示例我把三种最常见的情况都跑了一遍直接贴配置。走隧道连接 PostgreSQLconfig Config( host10.0.0.10, userpguser, passwordsecret, databaseanalytics, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jumpuser, password: jumpsecret, port: 22, allow_agent: True } )走隧道连接 MySQLconfig Config( host10.0.0.11, usermysqluser, passwordsecret, databasebilling, backendmysql, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jumpuser, private_key: /Users/me/.ssh/id_rsa, port: 2222 } )不走隧道、直连config Config( host127.0.0.1, userlocaluser, passwordlocalpass, databaselocaldb, backendmysql, config{charset: utf8mb4} )注意一个细节不传 ssh_tunnel 时默认就是 False不传 backend 默认就是 pg。如果你连的是 MySQL 却忘了写 backend后续查询会报错所以 MySQL 用户一定要显式写。这也是我踩过的一个坑花了不少时间排查。3.3 典型场景一批量导入数据到内网 PostgreSQL我实际遇到的场景是把一批 CSV 数据清洗后批量写入内网 PostgreSQL。流程分四步第一步读 CSV第二步清洗数据并转成元组列表第三步通过 a2conn 建立连接第四步分批执行插入。代码如下import csv from a2conn.config import Config from a2conn.connection import A2Conn config Config( host10.0.0.10, useretl_user, passwordetl_pass, databasedw, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jump, private_key: /Users/me/.ssh/id_rsa }, config{connect_timeout: 10} ) data [] with open(data.csv, r) as f: reader csv.DictReader(f) for row in reader: data.append(( row[order_id], row[product_id], row[amount], )) sql INSERT INTO dw.orders (order_id, product_id, amount) VALUES (%s, %s, %s) executions [(sql, row) for row in data] conn A2Conn() conn.connect(config) try: result conn.run(executions, query_size200) print(f写入完成受影响行数{result.rowcount}) finally: conn.close()对比没用 a2conn 之前的写法最明显的差距是原来需要自己管理隧道进程写完了还得确保进程被清掉很容易留下孤儿进程。用 a2conn 只需要在 finally 里调一下 close隧道会跟着一起关闭干净利落。有个经验分享query_size 设到 200是因为单条 SQL 里有三个字段5000 条数据会分成 25 批发送这样既不会造成单次网络包过大也不会因为分批太多增加太多往返。如果你插入的是几十个字段的大宽表建议把 query_size 进一步调低到 100 甚至 50数据库对单次 SQL 的长度有硬限制超过会直接报错。3.4 典型场景二在 Flask 接口中查询 MySQL 并返回 JSON另一个常见需求是给内部系统写个小接口查库返回数据。用 a2conn 连接 MySQL 时注意连接的关闭时机每请求先获取连接、用完关闭是比较稳妥的模式。from flask import Flask, jsonify from a2conn.config import Config from a2conn.connection import A2Conn app Flask(__name__) config Config( host10.0.0.11, userreadonly_user, passwordreadonly_pass, databasebilling, backendmysql, config{charset: utf8mb4} ) app.route(/api/orders/int:user_id) def get_orders(user_id): conn A2Conn() conn.connect(config) try: result conn.run( SELECT order_id, amount FROM orders WHERE user_id %s, params_list[(user_id,)] ) rows result.fetchall() return jsonify([dict(row) for row in rows]) finally: conn.close() if __name__ __main__: app.run(debugTrue)这段代码里我用 params_list 传入了参数而没有把 user_id 直接拼进 SQL 字符串。这样做不仅是为了防注入还因为带参查询能利用数据库的预编译缓存在高频查询场景下性能更好。3.5 典型场景三走跳板机把线上数据拉到本地做分析很多数据分析师其实也需要这个工具。本地要拉线上数据又不想每次手动建隧道可以用一段脚本把整张表读成 DataFrameimport pandas as pd from a2conn.config import Config from a2conn.connection import A2Conn config Config( host10.0.0.10, useranalyst, passwordanalyst_pass, databaseappdb, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jump, private_key: /Users/me/.ssh/id_rsa } ) conn A2Conn() conn.connect(config) try: result conn.run(SELECT * FROM users LIMIT 10000) df pd.DataFrame(result.fetchall(), columnsresult.keys()) df.to_csv(users_snapshot.csv, indexFalse) finally: conn.close()这个脚本我放在一个固定目录里每次要拉数据就改两行表名和字段其余不动。对团队里不熟悉 SSH 原理的同学来说这段代码比告诉他们先执行 ssh -L 命令再手动开个 MySQL 客户端要友好太多。3.6 资源释放close 的正确姿势前面几个例子里我都在 finally 里调用 conn.close()。这是必须养成的习惯原因在于如果启用了 ssh_tunnelclose 不仅要关闭数据库连接还要清理 SSH 隧道进程。如果你只关连接而不关隧道隧道进程会残留在系统里占着端口下一次连接如果端口复用规则不一致就可能冲突。手动管理的话可以显式调用 conn.close_tunnel() 来专门关隧道。但最省心的方案还是把整个操作放在 try/finally 里不管中间执行是否报错资源都会被回收。这也是我为什么在所有示例里都坚持这个写法实际项目中不加这个保护等隧道进程堆多了再排查相当头疼。4. 常见问题与排查技巧实录4.1 连接报错隧道没建立起来最常见的一类问题是连接直接抛异常提示访问目标主机超时或者拒绝连接。先别急着查数据库优先确认 ssh_tunnel 是否设置成了 True。很多人写完 Config复制了一个直连环境的配置忘了加这个开关导致代码在尝试直连内网数据库那肯定是连不上的。第二个检查点是 ssh_config 里的 host 和 port。很多运维会给跳板机设置非标准 SSH 端口比如 2222、22022你填 22 自然不对。第三个检查点是私钥路径是否正确尤其是路径里的 ~ 符号Python 不会自动展开它。建议用 os.path.expanduser 展开或者直接填绝对路径。4.2 链路通了但数据库用户名密码一直报错如果你确认隧道没问题、连接到数据库了但登录失败那大概率是用户名或密码的问题。用密码方式时密码不要写进 Config 对象里打印出来的日志里尽量从环境变量读import os config Config( host10.0.0.10, useros.getenv(DB_USER), passwordos.getenv(DB_PASSWORD), databaseappdb )另外a2conn 有 keyring 参数开启后它会优先从系统密钥环读密码。这个功能在公司电脑上很方便但要注意如果密钥环里存的是旧密码代码里又没显式传 password你会莫名报错而且是那种时好时坏的诡异错误。排查思路是先关掉 keyring 试一次能连再把问题定位到旧密码上。4.3 MySQL 查询结果字典化a2conn 默认的 run 结果集直接用 fetchall 拿到的是一行行的传统元组游标结果。如果你想拿到字典形式比如 key 是列名、value 是值可以在配置的 config 参数里指定游标类config Config( host10.0.0.11, userappuser, passwordapppass, databaseappdb, backendmysql, config{ charset: utf8mb4, cursorclass: DictCursor } )连接 MySQL 时传 cursorclass: DictCursor后续 fetchall 拿到的每一行就是字典。很多接口代码用了这个选项后会简化不少不需要再用 dict(row) 或者 row._mapping 去转格式。注意 Pg 后端没有这个参数Pg 的 SQLAlchemy 结果集默认就能通过 row._mapping 访问字典形式。4.4 批量执行时参数占位符不一致这是最容易踩的跨数据库坑。PostgreSQL 的参数占位符用 %sMySQL 也是 %s绝大多数情况下没问题。但如果你用 MySQL 且 SQL 里有 LIMIT 这种字面量或者用了其他驱动特有的格式占位符会出问题。我的经验是批量执行前先用一条最小 SQL 试跑确认参数能正确替换再全量执行别拿几万条数据去试错。有一条排查心得插入时如果总报数据类型错误检查一下 params 里的 Python 类型和数据库列类型是否匹配。a2conn 内部做的主要是参数传递它不会帮你做类型转换Decimal 传进 float 字段、datetime 传成字符串都会在数据库端报错。4.5 无法 import a2conn偶尔会遇到 import 失败的问题。最常见原因是环境混用pip 装到了一个 Python 环境但代码用另一个环境解释器跑。建议建虚拟环境来隔离项目依赖python -m venv venv source venv/bin/activate pip install a2conn如果你在公司内网pip 下载不了 public PyPI 包那就需要配置内部镜像源把 pip 源指到公司内部的 PyPI 代理上。这个属于环境问题和业务代码无关但非常影响入门体验。4.6 常见问题速查表现象优先排查点说明连接超时ssh_tunnel 是否开启直连配置直接连内网必失败SSH 登录失败ssh_config 端口与账号非标准端口很容易被忽视数据库登录失败keyring 旧密码关闭 keyring 排除干扰批量执行报错query_size 过大大字段表需要调低数值import 失败虚拟环境确认装的包与解释器同一环境5. 使用建议与后续扩展思路5.1 把 Config 收敛成项目级配置类别把 Config 的创建散落在各个脚本里。实际项目中我倾向于做一个 db.py 模块把每种环境的 Config 定义成模块级变量统一暴露给业务代码使用。这样运维要改跳板机地址或者数据库密码只需要改一处。# db.py from a2conn.config import Config PROD Config( host10.0.0.10, useros.getenv(PROD_DB_USER), passwordos.getenv(PROD_DB_PASSWORD), databaseappdb, ssh_tunnelTrue, ssh_config{ host: os.getenv(BASTION_HOST), user: os.getenv(BASTION_USER), private_key: os.path.expanduser(~/.ssh/id_rsa) } ) TEST Config( host10.0.0.20, usertest_user, passwordtest_pass, databaseappdb_test, )调用方只需要 from db import PROD然后 A2Conn().connect(PROD)特别清爽。5.2 与 SQLAlchemy ORM 结合如果你的项目已经在用 SQLAlchemy 管理模型a2conn 的 Config.get_engine 可以无缝对接。你只需要在模型定义时把 engine 和 declarative_base 绑定起来后续的 session 操作完全走正常 ORM 流程。有一点要提醒ORM 的 session 生命周期管理还是要按老规矩办不能让 session 跨请求存活。底层连接虽然是 a2conn 管的但 session 和事务仍然是你自己控制该 close 就 close该 rollback 就 rollback。5.3 定时任务的场景延展我的一个线上需求是每小时从内网数据库同步数据到另一个平台的 API这个场景把 a2conn 和 APScheduler 结合了起来。固定一个全局连接池定时任务每次从池里取连接跑查询组装数据调外部 API。运行了一个多月没有出现连接泄漏的问题。关键就在于定时任务里无论有没有异常都会释放连接。这一点对长跑任务来说比追求查询效率更重要。结合我个人实测a2conn 最值得肯定的一点是你只需要在写代码时想清楚配置和执行这两层它帮你把最脏最乱的网络细节扛掉了。虽然它名字有迷惑性、文档也不算多但对于内网开发环境的 Python 开发者来说属于那种一旦用上就回不去的工具。建议你拿到手之后先用后面那个最简示例跑通直连再加上隧道跑通跳板模式最后再上批量查询——按这个节奏来基本半小时内能上手。
开篇先把一个很容易被带偏的点说清楚a2conn 这个包名字里那个 a2很容易让人以为它和 aio 系列一样是个异步库。我第一次看到也是这么以为的装上之后才发现它跟异步没有半点关系它的主要作用是帮你在 Python 里跟数据库建立连接而且是走那种需要经过跳板机转一跳的复杂网络环境。很多项目里数据库并不直接暴露在公网你本地代码想连上去常规做法是自己拼 SSH 隧道或者让运维开网络策略而 a2conn 把这一层封装掉了调用方只需要关心主机、用户、库名这些业务参数底层隧道怎么建立、怎么销毁它替你处理。这篇文章我会把这几天实际用它写脚本的完整记录整理出来包括它暴露了哪些核心方法、每个参数控制什么行为、用的时候有哪些反直觉的坑以及一个从零到跑通的真实案例。适合正在维护内网数据库、又不想在业务代码里写一堆隧道逻辑的 Python 开发者也适合刚接触这类工具、想快速上手的人直接抄作业。1. 内容整体设计与思路拆解1.1 为什么需要 a2conn先搞清楚它解决什么问题先说场景。很多公司的数据库分两层线上业务库在核心内网开发机在办公网两边网络不互通。开发要查数据常规办法是先用 ssh 登录到一台和内网互通的跳板机再在跳板机上执行数据库命令或者手动搭一条 SSH 隧道把内网端口映射到本地。手动隧道的问题是隧道生命周期得自己管理代码里要处理端口冲突、断了重连、多环境切换写出来的东西跟数据库操作逻辑缠在一起特别丑。a2conn 的思路是把建立连接这件事收敛成一个配置对象和一个连接对象。你预先声明好运维给你的跳板机地址、账号、目标数据库地址和库名然后调用它的 connect 方法去建立隧道和连接之后你的代码就聚焦在 SQL 操作上隧道细节被封装在内部。这对需要定期跑数据任务、临时写脚本抽数的人来说省下的是大量跟业务无关的样板代码。1.2 它和 SQLAlchemy、pymysql 这些库的边界新手最容易混淆的是a2conn 不是用来替代 pymysql 或 psycopg2 的。它更像是一个连接管理器负责建立和维护底层连接真正执行 SQL 的仍然是数据库驱动。官方文档里它其实支持 postgres 和 mysql 两大类具体用哪个驱动由你传进来的后端配置决定。打个比方pymysql 是水管本身负责把水流送过来a2conn 是水龙头和阀门的组合负责在哪接水、怎么接、开关顺序。你写代码时照样用 pymysql 的游标和查询方法只是游标的来源从自己连一下数据库变成了从 a2conn 的连接对象里拿。这个区分很重要不然你会觉得它提供的功能太少实际上它的定位就不是大而全的 ORM。1.3 被名字误导的异步印象这一点值得单独说。PyPI 上有很多 a2 开头或者 a 开头的库是异步实现的比如 aiohttp、asyncssh所以看到 a2conn 的第一反应是又一个异步数据库连接库。打开源码你会发现它内部用的是 paramiko 来处理 SSH 通道而 paramiko 本身是同步的。整个连接和查询流程完全同步执行跟 asyncio 没有关系。那这个 a2 到底什么意思我翻了它的源码和注释它其实是从一个内部项目拆出来的沿用了项目代号。这个教训是不要光靠名字推断库的特性用之前花两分钟 pip download 看一眼源码比什么文档都管用。2. 核心语法与参数拆解2.1 最核心的两个对象Config 和 A2Conna2conn 的使用基本围绕两个 import 展开from a2conn.config import Config from a2conn.connection import A2ConnConfig 是一个数据类职责非常纯粹把数据库连接需要的信息集中管理然后对外提供 get_engine 和 get_connection 两个方法。A2Conn 则负责实际的动作connect 和 run 这两个方法一个建立连接一个执行操作。我建议的使用方式是一个项目里所有环境开发、测试、生产都定义成不同的 Config 实例用名字区分这样代码里不会出现一堆散落的连接字符串环境切换只是一行赋值的事。2.2 Config 各参数逐项说明Config 构造的时候不是所有参数都要填。按我的经验可以把参数分成三组必填的、可选但有默认值的、以及某些特殊场景才需要的。整理成表如下参数名是否必填作用host必填数据库服务器地址不是跳板机地址user必填目标数据库用户名password选填数据库密码如果配置里没有会从 keyring 里取database选填目标数据库名port选填数据库端口默认根据后端类型自动判断Pg 是 5432MySQL 是 3306backend选填后端类型传字符 pg 或 mysql默认 pgconfig选填给底层驱动用的配置比如 pg 的话传 connect_timeoutssh_tunnel选填要不要启用 SSH 隧道默认 Falsessh_config选填隧道相关参数的集合见下文keyring选填是否使用系统密钥环默认 Falsemodel选填模型的层定义用于 SQLAlchemy 的 Base 绑定schema选填模式名有个反直觉的地方port 可以不填。我最早用的时候习惯把所有参数写全后来发现后端类型一旦定了默认端口就是对的多传反而容易传错。如果你数据库跑在非默认端口再显式指定 port 就行。2.3 ssh_config 参数跳板机配置的关键ssh_config 是 Config 里的嵌套结构它控制隧道这头的所有网络参数。如果你不需要走隧道这一整块都可以不理会。需要走的时候注意这几个参数名作用host跳板机 IP 或域名user跳板机登录用户password跳板机密码如果有密钥优先用密钥portSSH 端口默认 22很多公司改成别的端口这里要留意private_key私钥路径用密钥登录时填allow_agent是否允许使用 ssh-agent 里的密钥默认 False实测经验能用私钥别用密码。密码方式在自动化任务里会面临过期、被安全策略锁定的问题私钥配合 ssh-agent 是长时间跑批任务最稳的方式。allow_agent 这个参数一开始容易被忽略如果你本机 ssh-agent 里已经加了密钥把它设为 True 能省去写 private_key 路径的麻烦尤其是同一个密钥管多台机器的时候。2.4 connect 与 run 的调用约定先看一个最小的连接示例config Config( host10.20.30.40, userappuser, passwordYourPassword, databaseappdb, backendmysql, config{charset: utf8mb4} ) conn A2Conn() conn.connect(config) run_result conn.run(SELECT 1)注意 connect 方法在铺隧道时是懒加载思路只有实际执行 run 或获取连接对象时隧道才真正建立起来。我第一次用的时候connect 之后立刻检查隧道端口发现什么动静都没有一度怀疑代码写错了。这个行为本身是设计使然不用惊慌只要在 run 之后去确认即可。run 方法是对外最主要的入口。它接受一个字符串形式的 SQL或者是带参查询的元组列表。如果是多条带参数查询每个元组的结构是 (sql, params_list)params_list 里每一项是这条 SQL 的参数集合这样一次调用就能批量执行多条相似 SQL减少网络往返。下面这个例子展示了批量插入的典型写法rows [] for i in range(5000): rows.append((fINSERT INTO test_table (id, name) VALUES (%s, %s), (i, fname_{i}))) result conn.run(rows, query_size500)query_size 参数控制一次往数据库发送多少条默认 1000。如果你的 SQL 比较复杂或者单条数据量大调低这个值可以避免数据库端报 too many SQL variables 之类的错误。这个参数是我在使用中认为最值得调的一个对性能的改善是实打实的。2.5 get_engine 和 get_connection 的适用场景除了 runConfig 还提供另外两条取连接的路径engine config.get_engine() connection config.get_connection()get_engine 返回的是 SQLAlchemy 的 Engine 对象适合要配合 ORM 或者要使用事务、连接池的场景。get_connection 返回的是一个裸连接适合直接用原生 SQL、又不想引入 ORM 上层概念的场景。最典型的用法是在 FastAPI 或 Flask 项目里启动时创建一个全局 Config每个请求需要访问数据库时调用 get_connection 获取独立连接用完关闭。因为底层连接实际上是经由隧道转发的连接的生命周期和你处理隧道资源的方式是绑定的务必确保使用完调用 close 方法释放。3. 实操过程与核心环节实现3.1 环境准备与安装安装很简单直接用 pippip install a2conn它会自动带上 paramiko、SQLAlchemy、keyring 这些依赖。如果你只需要 Pg 或 MySQL 其中一种装完基础包就能用如果你两种都要确保本机有对应的驱动比如 psycopg2-binary 或 PyMySQL驱动不在基础依赖里需要自己补pip install psycopg2-binary PyMySQL这里有个小坑paramiko 依赖 cryptography而 cryptography 在一些老的 Python 3.6 环境上需要编译经常导致安装失败。我的建议是别在 Python 3.6 上折腾直接上 3.8 及以上少很多事。3.2 三个不同后端的完整配置示例我把三种最常见的情况都跑了一遍直接贴配置。走隧道连接 PostgreSQLconfig Config( host10.0.0.10, userpguser, passwordsecret, databaseanalytics, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jumpuser, password: jumpsecret, port: 22, allow_agent: True } )走隧道连接 MySQLconfig Config( host10.0.0.11, usermysqluser, passwordsecret, databasebilling, backendmysql, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jumpuser, private_key: /Users/me/.ssh/id_rsa, port: 2222 } )不走隧道、直连config Config( host127.0.0.1, userlocaluser, passwordlocalpass, databaselocaldb, backendmysql, config{charset: utf8mb4} )注意一个细节不传 ssh_tunnel 时默认就是 False不传 backend 默认就是 pg。如果你连的是 MySQL 却忘了写 backend后续查询会报错所以 MySQL 用户一定要显式写。这也是我踩过的一个坑花了不少时间排查。3.3 典型场景一批量导入数据到内网 PostgreSQL我实际遇到的场景是把一批 CSV 数据清洗后批量写入内网 PostgreSQL。流程分四步第一步读 CSV第二步清洗数据并转成元组列表第三步通过 a2conn 建立连接第四步分批执行插入。代码如下import csv from a2conn.config import Config from a2conn.connection import A2Conn config Config( host10.0.0.10, useretl_user, passwordetl_pass, databasedw, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jump, private_key: /Users/me/.ssh/id_rsa }, config{connect_timeout: 10} ) data [] with open(data.csv, r) as f: reader csv.DictReader(f) for row in reader: data.append(( row[order_id], row[product_id], row[amount], )) sql INSERT INTO dw.orders (order_id, product_id, amount) VALUES (%s, %s, %s) executions [(sql, row) for row in data] conn A2Conn() conn.connect(config) try: result conn.run(executions, query_size200) print(f写入完成受影响行数{result.rowcount}) finally: conn.close()对比没用 a2conn 之前的写法最明显的差距是原来需要自己管理隧道进程写完了还得确保进程被清掉很容易留下孤儿进程。用 a2conn 只需要在 finally 里调一下 close隧道会跟着一起关闭干净利落。有个经验分享query_size 设到 200是因为单条 SQL 里有三个字段5000 条数据会分成 25 批发送这样既不会造成单次网络包过大也不会因为分批太多增加太多往返。如果你插入的是几十个字段的大宽表建议把 query_size 进一步调低到 100 甚至 50数据库对单次 SQL 的长度有硬限制超过会直接报错。3.4 典型场景二在 Flask 接口中查询 MySQL 并返回 JSON另一个常见需求是给内部系统写个小接口查库返回数据。用 a2conn 连接 MySQL 时注意连接的关闭时机每请求先获取连接、用完关闭是比较稳妥的模式。from flask import Flask, jsonify from a2conn.config import Config from a2conn.connection import A2Conn app Flask(__name__) config Config( host10.0.0.11, userreadonly_user, passwordreadonly_pass, databasebilling, backendmysql, config{charset: utf8mb4} ) app.route(/api/orders/int:user_id) def get_orders(user_id): conn A2Conn() conn.connect(config) try: result conn.run( SELECT order_id, amount FROM orders WHERE user_id %s, params_list[(user_id,)] ) rows result.fetchall() return jsonify([dict(row) for row in rows]) finally: conn.close() if __name__ __main__: app.run(debugTrue)这段代码里我用 params_list 传入了参数而没有把 user_id 直接拼进 SQL 字符串。这样做不仅是为了防注入还因为带参查询能利用数据库的预编译缓存在高频查询场景下性能更好。3.5 典型场景三走跳板机把线上数据拉到本地做分析很多数据分析师其实也需要这个工具。本地要拉线上数据又不想每次手动建隧道可以用一段脚本把整张表读成 DataFrameimport pandas as pd from a2conn.config import Config from a2conn.connection import A2Conn config Config( host10.0.0.10, useranalyst, passwordanalyst_pass, databaseappdb, ssh_tunnelTrue, ssh_config{ host: bastion.example.com, user: jump, private_key: /Users/me/.ssh/id_rsa } ) conn A2Conn() conn.connect(config) try: result conn.run(SELECT * FROM users LIMIT 10000) df pd.DataFrame(result.fetchall(), columnsresult.keys()) df.to_csv(users_snapshot.csv, indexFalse) finally: conn.close()这个脚本我放在一个固定目录里每次要拉数据就改两行表名和字段其余不动。对团队里不熟悉 SSH 原理的同学来说这段代码比告诉他们先执行 ssh -L 命令再手动开个 MySQL 客户端要友好太多。3.6 资源释放close 的正确姿势前面几个例子里我都在 finally 里调用 conn.close()。这是必须养成的习惯原因在于如果启用了 ssh_tunnelclose 不仅要关闭数据库连接还要清理 SSH 隧道进程。如果你只关连接而不关隧道隧道进程会残留在系统里占着端口下一次连接如果端口复用规则不一致就可能冲突。手动管理的话可以显式调用 conn.close_tunnel() 来专门关隧道。但最省心的方案还是把整个操作放在 try/finally 里不管中间执行是否报错资源都会被回收。这也是我为什么在所有示例里都坚持这个写法实际项目中不加这个保护等隧道进程堆多了再排查相当头疼。4. 常见问题与排查技巧实录4.1 连接报错隧道没建立起来最常见的一类问题是连接直接抛异常提示访问目标主机超时或者拒绝连接。先别急着查数据库优先确认 ssh_tunnel 是否设置成了 True。很多人写完 Config复制了一个直连环境的配置忘了加这个开关导致代码在尝试直连内网数据库那肯定是连不上的。第二个检查点是 ssh_config 里的 host 和 port。很多运维会给跳板机设置非标准 SSH 端口比如 2222、22022你填 22 自然不对。第三个检查点是私钥路径是否正确尤其是路径里的 ~ 符号Python 不会自动展开它。建议用 os.path.expanduser 展开或者直接填绝对路径。4.2 链路通了但数据库用户名密码一直报错如果你确认隧道没问题、连接到数据库了但登录失败那大概率是用户名或密码的问题。用密码方式时密码不要写进 Config 对象里打印出来的日志里尽量从环境变量读import os config Config( host10.0.0.10, useros.getenv(DB_USER), passwordos.getenv(DB_PASSWORD), databaseappdb )另外a2conn 有 keyring 参数开启后它会优先从系统密钥环读密码。这个功能在公司电脑上很方便但要注意如果密钥环里存的是旧密码代码里又没显式传 password你会莫名报错而且是那种时好时坏的诡异错误。排查思路是先关掉 keyring 试一次能连再把问题定位到旧密码上。4.3 MySQL 查询结果字典化a2conn 默认的 run 结果集直接用 fetchall 拿到的是一行行的传统元组游标结果。如果你想拿到字典形式比如 key 是列名、value 是值可以在配置的 config 参数里指定游标类config Config( host10.0.0.11, userappuser, passwordapppass, databaseappdb, backendmysql, config{ charset: utf8mb4, cursorclass: DictCursor } )连接 MySQL 时传 cursorclass: DictCursor后续 fetchall 拿到的每一行就是字典。很多接口代码用了这个选项后会简化不少不需要再用 dict(row) 或者 row._mapping 去转格式。注意 Pg 后端没有这个参数Pg 的 SQLAlchemy 结果集默认就能通过 row._mapping 访问字典形式。4.4 批量执行时参数占位符不一致这是最容易踩的跨数据库坑。PostgreSQL 的参数占位符用 %sMySQL 也是 %s绝大多数情况下没问题。但如果你用 MySQL 且 SQL 里有 LIMIT 这种字面量或者用了其他驱动特有的格式占位符会出问题。我的经验是批量执行前先用一条最小 SQL 试跑确认参数能正确替换再全量执行别拿几万条数据去试错。有一条排查心得插入时如果总报数据类型错误检查一下 params 里的 Python 类型和数据库列类型是否匹配。a2conn 内部做的主要是参数传递它不会帮你做类型转换Decimal 传进 float 字段、datetime 传成字符串都会在数据库端报错。4.5 无法 import a2conn偶尔会遇到 import 失败的问题。最常见原因是环境混用pip 装到了一个 Python 环境但代码用另一个环境解释器跑。建议建虚拟环境来隔离项目依赖python -m venv venv source venv/bin/activate pip install a2conn如果你在公司内网pip 下载不了 public PyPI 包那就需要配置内部镜像源把 pip 源指到公司内部的 PyPI 代理上。这个属于环境问题和业务代码无关但非常影响入门体验。4.6 常见问题速查表现象优先排查点说明连接超时ssh_tunnel 是否开启直连配置直接连内网必失败SSH 登录失败ssh_config 端口与账号非标准端口很容易被忽视数据库登录失败keyring 旧密码关闭 keyring 排除干扰批量执行报错query_size 过大大字段表需要调低数值import 失败虚拟环境确认装的包与解释器同一环境5. 使用建议与后续扩展思路5.1 把 Config 收敛成项目级配置类别把 Config 的创建散落在各个脚本里。实际项目中我倾向于做一个 db.py 模块把每种环境的 Config 定义成模块级变量统一暴露给业务代码使用。这样运维要改跳板机地址或者数据库密码只需要改一处。# db.py from a2conn.config import Config PROD Config( host10.0.0.10, useros.getenv(PROD_DB_USER), passwordos.getenv(PROD_DB_PASSWORD), databaseappdb, ssh_tunnelTrue, ssh_config{ host: os.getenv(BASTION_HOST), user: os.getenv(BASTION_USER), private_key: os.path.expanduser(~/.ssh/id_rsa) } ) TEST Config( host10.0.0.20, usertest_user, passwordtest_pass, databaseappdb_test, )调用方只需要 from db import PROD然后 A2Conn().connect(PROD)特别清爽。5.2 与 SQLAlchemy ORM 结合如果你的项目已经在用 SQLAlchemy 管理模型a2conn 的 Config.get_engine 可以无缝对接。你只需要在模型定义时把 engine 和 declarative_base 绑定起来后续的 session 操作完全走正常 ORM 流程。有一点要提醒ORM 的 session 生命周期管理还是要按老规矩办不能让 session 跨请求存活。底层连接虽然是 a2conn 管的但 session 和事务仍然是你自己控制该 close 就 close该 rollback 就 rollback。5.3 定时任务的场景延展我的一个线上需求是每小时从内网数据库同步数据到另一个平台的 API这个场景把 a2conn 和 APScheduler 结合了起来。固定一个全局连接池定时任务每次从池里取连接跑查询组装数据调外部 API。运行了一个多月没有出现连接泄漏的问题。关键就在于定时任务里无论有没有异常都会释放连接。这一点对长跑任务来说比追求查询效率更重要。结合我个人实测a2conn 最值得肯定的一点是你只需要在写代码时想清楚配置和执行这两层它帮你把最脏最乱的网络细节扛掉了。虽然它名字有迷惑性、文档也不算多但对于内网开发环境的 Python 开发者来说属于那种一旦用上就回不去的工具。建议你拿到手之后先用后面那个最简示例跑通直连再加上隧道跑通跳板模式最后再上批量查询——按这个节奏来基本半小时内能上手。
