async-db-tools 0.1.2

Async database connection pool utilities for PostgreSQL (Redis, MySQL planned)

async-db-tools

基于 asyncpg 的 PostgreSQL 异步连接池封装，提供带自动重试的常用查询 API。后续计划支持 Redis、MySQL 等。

安装

pip install async-db-tools
# 或
uv add async-db-tools

快速开始

import asyncio

from async_db_tools import PostgresPool


async def main() -> None:
    async with PostgresPool("postgresql://user:pass@localhost:5432/mydb") as db:
        row = await db.fetchrow("SELECT 1 AS n")
        print(row["n"])


asyncio.run(main())

也可手动管理生命周期：

db = PostgresPool(
    "postgresql://user:pass@localhost:5432/mydb",
    min_size=2,
    max_size=20,
    max_retries=3,
)
await db.connect()
try:
    await db.execute("INSERT INTO t (v) VALUES ($1)", 42)
finally:
    await db.close()

API

方法	说明
connect()	创建连接池（幂等）
close()	关闭连接池
execute(query, *args)	执行写操作
executemany(query, args)	批量写操作
fetch / fetchrow / fetchval	查询
acquire()	获取底层连接（context manager），用于事务等
pool	底层 asyncpg.Pool，用于更高级用法

连接类错误（网络断开等）会自动重试，次数由 max_retries 控制。

init 回调与 codec 预设

PostgresPool 支持 init 回调，对每个新建连接做一次性初始化（如注册 type codec）：

from async_db_tools import PostgresPool, setup_json_codec, setup_numeric_float_codec


async def init(conn):
    await setup_json_codec(conn)            # jsonb/json → dict/list
    await setup_numeric_float_codec(conn)   # numeric → float


db = PostgresPool(dsn, init=init)

acquire 获取原始连接

async with db.acquire() as conn:
    async with conn.transaction():
        await conn.execute("DELETE FROM t WHERE id=$1", 1)
        await conn.executemany("INSERT INTO t VALUES ($1,$2)", rows)

项目结构

src/async_db_tools/
├── __init__.py
└── postgresql/          # PostgreSQL
    ├── __init__.py
    └── pool.py
# 后续: redis/, mysql/ ...

开发

uv sync --extra dev
uv build

测试

cp .env.example .env   # 填写 PG_DSN
uv sync --extra dev
uv run pytest -v

未设置 PG_DSN 时，依赖数据库的用例会自动跳过。

许可证

MIT