async-db-tools 0.1.1

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)	执行写操作
fetch / fetchrow / fetchval	查询
pool	底层 asyncpg.Pool，用于事务等高级用法

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

项目结构

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