Python asyncpg库介绍（基于Python asyncio的PostgreSQL数据库驱动）连接池、SQLAlchemy

asyncpg = asynchronous + PostgreSQL

文章目录

• 深入理解 asyncpg：高性能 PostgreSQL 异步驱动实践指南
• • 一、什么是 asyncpg？
  • 二、为什么选择 asyncpg？
  • 三、安装与基础使用
  • • 1. 安装
    • 2. 基础示例
  • 四、核心 API 解析
  • • 1. 查询方法
    • 2. 执行写操作
    • 3. 批量操作
  • 五、连接池（生产必备）
  • • 示例：
    • 为什么必须用连接池？
  • 六、事务管理
  • 七、与 SQLAlchemy 2.x 的结合
  • 八、结合微服务架构的最佳实践
  • • 1️⃣ 每个服务只访问自己的 schema
    • 2️⃣ 禁止跨 schema 联表
    • 3️⃣ 使用连接池 + 异步模型
  • 九、性能优化技巧
  • • ✅ 使用 prepared statements（自动缓存）
    • ✅ 避免 N+1 查询
    • ✅ 使用 fetchval 提升性能
  • 十、常见坑
  • • ⚠️ 1. 忘记 await
    • ⚠️ 2. 连接泄漏
    • ⚠️ 3. 阻塞代码混入 async
  • 十一、适用场景总结
  • 十二、总结

深入理解 asyncpg：高性能 PostgreSQL 异步驱动实践指南

在现代微服务架构中，数据库访问逐渐从“阻塞式”转向“异步非阻塞式”。尤其是在 Python 生态中，随着 asyncio 的成熟，异步数据库驱动成为高并发场景的首选。

本文将带你深入理解 asyncpg —— 一个专为 PostgreSQL 设计的高性能异步驱动。

一、什么是 asyncpg？

asyncpg 是一个基于 Python asyncio 的 PostgreSQL 数据库驱动，具备以下特点：

• 🚀 极致性能（通常比 psycopg2 快 2~5 倍）
• ⚡ 原生异步（非线程池模拟）
• 🧠 高效协议实现（直接实现 PostgreSQL 二进制协议）
• 🔒 类型安全（自动映射 PostgreSQL 类型）

它不是简单的 ORM，而是一个更底层、更高效的数据库访问工具。

二、为什么选择 asyncpg？

在传统同步模型中：

# 同步阻塞
result = cursor.execute("SELECT * FROM users")

每一个数据库请求都会阻塞线程。

而 asyncpg：

# 异步非阻塞
rows = await conn.fetch("SELECT * FROM users")

优势非常明显：

三、安装与基础使用

1. 安装

pip install asyncpg

2. 基础示例

import asyncio
import asyncpg

async def main():
conn = await asyncpg.connect(
user='user',
password='password',
database='testdb',
host='127.0.0.1'
)

rows = await conn.fetch("SELECT id, name FROM users")

for row in rows:
print(row['id'], row['name'])

await conn.close()

asyncio.run(main())

四、核心 API 解析

1. 查询方法

await conn.fetch(...) # 返回多行
await conn.fetchrow(...) # 返回单行
await conn.fetchval(...) # 返回单个值

示例：

user = await conn.fetchrow(
"SELECT * FROM users WHERE id=$1", 1
)

👉 注意：参数使用 $1 占位符，而不是 %s

2. 执行写操作

await conn.execute(
"INSERT INTO users(name) VALUES($1)", "Alice"
)

3. 批量操作

await conn.executemany(
"INSERT INTO users(name) VALUES($1)",
[("Alice",), ("Bob",)]
)

五、连接池（生产必备）

在实际项目中，绝对不要频繁创建连接，应使用连接池。

示例：

pool = await asyncpg.create_pool(
user='user',
password='password',
database='testdb',
host='127.0.0.1',
min_size=5,
max_size=20
)

async with pool.acquire() as conn:
rows = await conn.fetch("SELECT * FROM users")

为什么必须用连接池？

• 避免连接创建开销
• 控制数据库连接数
• 提升整体吞吐量

六、事务管理

async with conn.transaction():
await conn.execute("INSERT INTO users(name) VALUES($1)", "Alice")
await conn.execute("INSERT INTO logs(action) VALUES($1)", "create_user")

如果中间发生异常，会自动 rollback。

七、与 SQLAlchemy 2.x 的结合

在现代 Python 项目中，推荐：

asyncpg + SQLAlchemy 2.x async

示例：

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
"postgresql+asyncpg://user:password@localhost/testdb"
)

优点：

• ORM + 高性能驱动结合
• 统一数据库抽象层
• 更易维护大型项目

八、结合微服务架构的最佳实践

1️⃣ 每个服务只访问自己的 schema

✔ asyncpg 支持指定 schema：

SET search_path TO my_service_schema;

2️⃣ 禁止跨 schema 联表

❌ 错误做法：

SELECT * FROM service_a.users
JOIN service_b.orders

✔ 正确做法：

• 通过 API 获取数据
• 服务解耦

3️⃣ 使用连接池 + 异步模型

推荐架构：

FastAPI / Aiohttp
↓
asyncpg Pool
↓
PostgreSQL

九、性能优化技巧

✅ 使用 prepared statements（自动缓存）

asyncpg 默认会缓存查询计划：

await conn.fetch("SELECT * FROM users WHERE id=$1", 1)

✅ 避免 N+1 查询

❌：

for user in users:
await conn.fetch(...)

✔：

SELECT * FROM users WHERE id = ANY($1)

✅ 使用 fetchval 提升性能

count = await conn.fetchval("SELECT COUNT(*) FROM users")

十、常见坑

⚠️ 1. 忘记 await

conn.fetch(...) # ❌
await conn.fetch(...) # ✅

⚠️ 2. 连接泄漏

conn = await pool.acquire()
# 忘记释放 ❌

✔：

async with pool.acquire() as conn:
...

⚠️ 3. 阻塞代码混入 async

time.sleep(1) # ❌
await asyncio.sleep(1) # ✅

十一、适用场景总结

适合使用 asyncpg 的场景：

• 高并发 API 服务（如 FastAPI）
• 微服务架构
• 实时系统（如消息处理）
• 数据密集型应用

不适合：

• 简单脚本
• 低并发后台任务

十二、总结

asyncpg 的核心价值可以归纳为三点：

🚀 高性能 ⚡ 真异步 🧩 适配现代架构

如果你正在构建：

• 微服务系统
• 高并发 API
• 云原生应用

那么 asyncpg 基本是 PostgreSQL 的“默认推荐方案”。