在Python的数据库生态里，操作PostgreSQL最经典的方案是psycopg2，但它是同步阻塞的，一个查询没执行完，整个事件循环都会被卡住。asyncpg的出现解决了这个痛点，它是专门为PostgreSQL打造的高性能异步驱动，官方实测性能比psycopg2和aiopg都要出色。这篇文章会从安装、连接池、查询、事务到与Web框架整合，完整讲清楚asyncpg的用法和容易踩的坑。
为什么选择asyncpg而不是psycopg2或aiopg
asyncpg最大的特点是直接实现了PostgreSQL的二进制协议，而不是在C驱动或者DB-API之上做封装。这意味着数据在传输阶段就已经是PostgreSQL的原生格式，asyncpg在Python层直接解析，省去了字符串转换的中间环节。对于返回大量行的查询，这个优势非常明显，尤其是数值类型和时间类型，不需要逐行做类型推断和转换。
与aiopg相比，asyncpg不是简单地在psycopg2外面套一层线程池，而是真正的原生异步实现。aiopg的并发能力受限于线程池大小，而asyncpg完全运行在事件循环里，配合uvloop使用时性能还能进一步提升。此外，asyncpg内置了对Prepared Statement的支持，同一个SQL多次执行时会复用执行计划，减少解析开销。
安装非常简单，直接用pip即可：
pip install asyncpg
需要说明的是，asyncpg只支持PostgreSQL，不支持其他数据库。如果你的项目未来可能切换数据库，这会是一个需要权衡的点，可以选择SQLAlchemy 2.0的异步版本作为抽象层，底层仍然用asyncpg驱动。
创建连接与连接池的正确姿势
单次连接适合脚本场景，直接调用asyncpg.connect()就能拿到一个连接对象：
import asyncio
import asyncpg
async def main():
# 单连接示例
conn = await asyncpg.connect(
host='127.0.0.1',
port=5432,
user='postgres',
password='secret',
database='demo',
)
# 执行简单查询
version = await conn.fetchval('SELECT version()')
print(version)
await conn.close()
asyncio.run(main())但在实际的Web服务中，绝对不要每次请求都新建连接。建立连接的过程涉及TCP握手、认证、会话初始化，成本很高。正确的做法是使用连接池，asyncpg提供了create_pool函数：
import asyncpg
pool = None
async def init_pool():
global pool
pool = await asyncpg.create_pool(
host='127.0.0.1',
port=5432,
user='postgres',
password='secret',
database='demo',
min_size=5, # 池中常驻连接数
max_size=20, # 最大连接数
command_timeout=30, # 单条命令超时时间
)
async def query_users():
async with pool.acquire() as conn:
rows = await conn.fetch('SELECT id, name FROM users')
return rows连接池的acquire方法配合async with使用，连接用完会自动归还，即使代码抛出异常也不会泄漏连接。建议连接池在应用启动时创建一次，全局复用，不要在函数内部反复创建，否则池就失去了意义。
还有一点需要注意，max_size不要设置得过大。PostgreSQL每个连接都是一个进程，连接太多会消耗大量内存，一般按照服务的并发量乘以单请求的持有时长来估算即可，通常20到50已经能支撑大多数业务。
参数化查询与结果处理
asyncpg使用$1、$2这样的占位符做参数化查询，这一点和psycopg2的%s风格完全不同，从psycopg2迁移代码时必须改写：
async def get_user(pool, user_id):
async with pool.acquire() as conn:
# 使用 $1 占位符，防止SQL注入
row = await conn.fetchrow(
'SELECT id, name, email FROM users WHERE id = $1', user_id
)
return row
async def insert_user(pool, name, email):
async with pool.acquire() as conn:
# returning 返回插入后的数据
new_id = await conn.fetchval(
'INSERT INTO users (name, email) VALUES ($1, $2) RETURNING id',
name, email
)
return new_id查询结果有三种主要方法：fetch返回Record对象列表，fetchrow返回单条或None，fetchval返回第一行第一列的值。Record对象既支持索引访问也支持键访问，可以像字典一样用row['name']取值，也能通过dict(row)转成字典。
一个常见的坑是字段值为NULL时的处理。asyncpg对类型推断非常严格，如果参数中要传入NULL，最好显式指定类型，例如$1::int，否则可能报类型推断失败的错误：
# 显式类型转换，避免 None 导致类型推断失败
row = await conn.fetchrow(
'SELECT * FROM users WHERE age = $1::int', None
)另外，asyncpg不会自动把Python的datetime转成带时区的timestamp，如果字段是timestamptz类型，传入的datetime对象必须带时区信息，否则会抛出类型错误，这是新手最容易遇到的问题之一。
事务管理与实战整合
事务是数据库操作绕不开的话题，asyncpg提供了多种方式开启事务，最推荐的是用async with conn.transaction()这种上下文管理器形式，代码块正常结束自动提交，抛出异常自动回滚：
async def transfer_points(pool, from_user, to_user, points):
async with pool.acquire() as conn:
async with conn.transaction():
# 事务内的多条SQL，要么全部成功，要么全部回滚
await conn.execute(
'UPDATE users SET points = points - $1 WHERE id = $2',
points, from_user
)
await conn.execute(
'UPDATE users SET points = points + $1 WHERE id = $2',
points, to_user
)如果需要更细粒度的控制，可以使用conn.transaction()返回的对象手动调用start和commit，但在异步代码中手动管理事务边界容易出错，上下文管理器是更安全的选择。
最后看一下与FastAPI整合的完整例子，在应用生命周期中管理连接池：
from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncpg
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.pool = await asyncpg.create_pool(
host='127.0.0.1', user='postgres',
password='secret', database='demo',
min_size=5, max_size=20
)
yield
await app.state.pool.close()
app = FastAPI(lifespan=lifespan)
@app.get('/users/{user_id}')
async def get_user(user_id: int):
pool = app.state.pool
async with pool.acquire() as conn:
row = await conn.fetchrow(
'SELECT id, name, email FROM users WHERE id = $1', user_id
)
if row is None:
return {'error': 'user not found'}
return {'id': row['id'], 'name': row['name'], 'email': row['email']}这种写法保证了连接池随应用启动而创建、随应用关闭而释放，各个请求处理函数直接访问app.state.pool，干净且高效。
总结一下，asyncpg的核心优势在于原生异步加二进制协议带来的高性能，使用时的要点是：全局复用连接池、坚持参数化查询、用上下文管理器处理事务、注意NULL参数和时区字段的类型问题。把这些细节处理好，你的Python服务在数据库层面就不会成为瓶颈。
asyncpgPostgreSQLPython异步编程修改时间：2026-09-13 20:48:52