在 Python 异步生态中，asyncpg 是操作 PostgreSQL 的高性能驱动。通过 asyncpg.connect 可以快速获得一个数据库连接对象，但很多人在拿到连接后写出 SQL 却得不到预期结果，或者遇到连接闲置被服务端断开的问题。理解 connect 的参数语义与连接对象的方法边界，是写出可靠数据库访问代码的前提。
一、asyncpg.connect 的基础参数与建立连接
asyncpg.connect 是一个协程函数，调用时必须使用 await。它并不会读取环境变量或配置文件，所有连接信息都要以关键字参数形式传入。最基础的参数包括 host、port、user、password 和 database。如果省略 host，在 Linux 环境下驱动会尝试通过 Unix 套接字连接，若 Postgres 未开启本地套接字，就会抛出 Connection refused 类错误。
下面是一段最小可用的建连代码，展示了如何显式传入参数并获得连接：
import asyncio
import asyncpg
async def main():
conn = await asyncpg.connect(
host='127.0.0.1',
port=5432,
user='postgres',
password='your_password',
database='test_db'
)
print('连接已建立:', conn)
await conn.close()
asyncio.run(main())
这段代码中，host 使用 127.0.0.1 而非 localhost，可避免某些系统上 localhost 被解析为 IPv6 地址而连不上。port 默认是 5432，但显式写出更利于后期维护。password 明文写在代码里仅适合演示，生产环境应通过密钥管理或连接字符串注入。
连接建立后，对象 conn 的生命周期由开发者控制。忘记调用 close 会导致连接泄漏，尤其在循环或请求处理中反复 connect 却不关闭，很快就会耗尽数据库的最大连接数。因此，更推荐用 async with 语法配合 connect，或者直接使用连接池。
二、执行 SQL 的方法差异与正确选择
拿到连接对象后，常见误区是把所有 SQL 都交给 execute 方法。实际上，execute 仅适用于 INSERT、UPDATE、DELETE、DDL 等不返回结果集的语句，其返回值是 None 或受影响的行数字符串（取决于版本）。如果需要查询数据，必须改用 fetch、fetchrow 或 fetchval。
以下示例对比了两种典型用法：
async def demo(conn):
# 执行写操作，不返回结果集
result = await conn.execute("INSERT INTO users(name) VALUES('tom')")
print('execute 返回:', result)
# 查询多行
rows = await conn.fetch("SELECT id, name FROM users")
for r in rows:
print(r['id'], r['name'])
# 查询单行
row = await conn.fetchrow("SELECT * FROM users WHERE name='tom'")
print('单行:', row)
# 查询单个值
count = await conn.fetchval("SELECT count(*) FROM users")
print('总数:', count)
从代码可以看出，fetch 返回列表，每个元素是 Record 对象，支持字典式访问；fetchrow 返回单条 Record 或 None；fetchval 直接取出第一个字段的值。若误用 execute 去 SELECT，虽然不会报错，但拿不到任何数据，容易让人误以为查询失败。
另外，参数化查询应使用 $1、$2 占位符而非 Python 的字符串格式化，这样既防止 SQL 注入，也避免引号转义问题。示例如下：
name = 'jerry'
age = 20
await conn.execute("INSERT INTO users(name, age) VALUES($1, $2)", name, age)
asyncpg 的占位符基于 PostgreSQL 本身的协议层，性能优于拼接 SQL。同时，它在协程内部自动处理编码，中文等非 ASCII 字符无需额外声明。
三、事务控制与异常隔离
使用 asyncpg.connect 建立的连接默认处于自动提交模式，每条 execute 立即生效。若需要多语句原子操作，应显式开启事务。连接对象提供 conn.transaction() 上下文管理器，进入块后执行的语句要么全部提交，要么回滚。
async def transfer(conn, a, b, amount):
async with conn.transaction():
await conn.execute("UPDATE accounts SET balance=balance-$1 WHERE id=$2", amount, a)
await conn.execute("UPDATE accounts SET balance=balance+$1 WHERE id=$2", amount, b)
上述代码若第二条语句失败，第一条的扣款也会回滚，保证数据一致。注意事务块内不要穿插耗时过长的网络请求，否则会长期占用连接，阻塞其他协程。
异常处理方面，建议捕获 asyncpg.PostgresError 及其子类，针对性处理唯一约束冲突、外键错误等。连接断开时通常会抛出 ConnectionDoesNotExistError，此时应重建连接而非继续使用原对象。
from asyncpg import PostgresError
try:
await conn.execute("INSERT INTO users(name) VALUES($1)", 'dup')
except PostgresError as e:
print('数据库错误:', e.message)
通过分层捕获，可以把业务异常和系统异常区分开，避免一个字段超长就导致整个服务崩溃。配合日志上报，能快速定位是 SQL 写错还是连接池配置不足。
四、从单连接到连接池的演进
虽然 asyncpg.connect 能满足脚本或低频任务，但在 Web 服务中每次请求都建连会带来明显开销。asyncpg 提供 create_pool 来复用连接，池内部调用 connect 逻辑，对外提供 acquire 方法。
async def use_pool():
pool = await asyncpg.create_pool(
host='127.0.0.1',
user='postgres',
password='your_password',
database='test_db',
min_size=2,
max_size=10
)
async with pool.acquire() as conn:
await conn.execute("INSERT INTO logs(msg) VALUES('hello')")
await pool.close()
池中连接都通过相同的 connect 参数建立，acquire 拿到的连接在释放后回到池内而非真正关闭。这样既保留了 connect 的灵活配置，又避免了频繁握手。对于标题中的问题，正确做法是：在初始化阶段用 connect 验证参数无误，在运行阶段用池获取可执行 SQL 的连接。
总结来说，asyncpg.connect 本身只是入口，能否稳定执行 SQL 取决于参数完整、方法匹配和连接回收。理清 execute 与 fetch 的分工，用事务保证边界，用池化解并发，才算真正用对了这个接口。
asyncpgasyncpg_connectPostgreSQL修改时间：2026-08-09 13:51:40