Python + PostgreSQL 异步操作：asyncpg 连接池、事务与批量写入

asyncpg 是 Python 最快的 PostgreSQL 驱动——纯 asyncio、无线程开销，典型负载比 psycopg2 快最多 3 倍，是异步 Python 后端的首选。

连接池

import asyncio, asyncpg

DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

async def create_pool() -> asyncpg.Pool:
    pool = await asyncpg.create_pool(
        DATABASE_URL,
        min_size=2, max_size=10,
        command_timeout=30,
        server_settings={"application_name": "myapp"},
    )
    return pool

create_pool() 只建一次，之后从池里 acquire 连接非常快——不要每次请求都新建连接。

建表与写入

async def setup_schema(pool):
    async with pool.acquire() as conn:
        await conn.execute(CREATE_TABLES)  # users / posts + 索引

async def create_user(pool, username, email) -> int:
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """INSERT INTO users (username, email)
               VALUES ($1, $2)
               ON CONFLICT (username) DO UPDATE SET email = EXCLUDED.email
               RETURNING id, created_at""",
            username, email,
        )
        return row["id"]

批量写入用 copy_records_to_table——比循环 INSERT 快 10–50 倍：

await conn.copy_records_to_table(
    "posts", records=rows,
    columns=["user_id", "title", "body", "published"],
)

查询

async def get_user_posts(pool, user_id, limit=20, offset=0):
    async with pool.acquire() as conn:
        return await conn.fetch(
            """SELECT p.id, p.title, p.published, p.created_at, u.username
               FROM posts p JOIN users u ON u.id = p.user_id
               WHERE p.user_id = $1
               ORDER BY p.created_at DESC LIMIT $2 OFFSET $3""",
            user_id, limit, offset,
        )

全文检索直接用 PostgreSQL 的 tsvector：WHERE to_tsvector('english', p.title || ' ' || p.body) @@ plainto_tsquery('english', $1)。

事务

async def transfer_posts(pool, from_user, to_user) -> int:
    async with pool.acquire() as conn:
        async with conn.transaction():
            # 两条语句原子执行
            result = await conn.execute(
                "UPDATE posts SET user_id = $1 WHERE user_id = $2",
                to_user, from_user,
            )
            count = int(result.split()[-1])
            await conn.execute(
                "INSERT INTO audit_log (action, detail) VALUES ($1, $2)",
                "transfer_posts", f"from={from_user} to={to_user} count={count}",
            )
            return count

async with conn.transaction() 嵌套安全——内层自动变成 savepoint。

实践建议

• 始终用 $1, $2 位置参数防 SQL 注入（asyncpg 原生支持，不要拼字符串）；
• fetchrow() 无匹配返回 None，取字段前先判空；
• 加 server_settings={"statement_timeout": "5000"} 防跑飞查询；
• 参数化查询复用 prepared statement，热点路径性能差距明显；
• 完整流程：建池 → 建表 → 写入 → 查询 → 关闭池，一个 main() 里按顺序编排即可。

来源：Python PostgreSQL with asyncpg: Async Database Operations - DEV Community