PyDBPool - Python Database Connection Pool

PyDBPool 是一个高性能、通用的数据库连接池实现，支持主流关系型数据库，提供完善的连接生命周期管理和监控功能。

特性

• 🚀 多数据库支持：PostgreSQL、MySQL、SQLite 等
• 🔒 线程安全：严格的锁机制保证高并发安全
• 📊 实时监控：内置连接池指标统计
• 🩺 健康检查：自动心跳检测与失效连接剔除
• ⚡ 异步就绪：支持协程环境（需异步驱动）
• 🔌 智能调度：动态扩缩容与最小空闲维持
• 🛠️ 扩展接口：钩子函数与自定义策略支持

安装

# 基础安装
pip install pydbpool

# 按需选择数据库驱动
pip install pydbpool[postgres]   # PostgreSQL支持
pip install pydbpool[mysql]      # MySQL支持

快速开始

基本用法

from pydbpool.dbpool import DBPool
from pydbpool.base import PoolOption
import pymysql

# 初始化连接池
opt = PoolOption(
    minsize=1,
    maxsize=3,
    max_usage=3,
    # max_overflow=2,
    max_retries=2,
    pool_recycle=5,
    # pool_timeout=3,

)
pool = DBPool(
    connect_func=pymysql.connect,
    option=opt,
    host="127.0.0.1",
    port=3306,
    user="root",
    password="*********",
    database="db_stg_demo",
)

# 使用连接
sql = """SHOW FULL PROCESSLIST;"""
with pool.connection() as conn:
    cursor = conn.cursor()
    cursor.execute(sql)
    print(cursor.fetchall())

监控指标

print(pool.meta.as_dict())
# 输出示例：
{
    "created_at": 272169.148343291,
    "last_active": 272169.148343708,
    "is_closed": False,
    "usage_count": 0,
    "lifetime": 0.0,
    "total_conn_count": 13,
    "using_conn_count": 13,
    "free_conn_count": 0
}

重写自己的连接池和连接类功能

连接基类 BaseConnection

from pydbpool.base import BaseConnection
from pydbpool.dbpool import DBPool


# 实现自己的连接类
class YourselfConnection(BaseConnection):
    pass


class YourselfPool(DBPool):
    pass

pool option 配置选项

性能建议

1. 连接数配置：

   # 推荐公式
   max_size = (avg_concurrent_requests × avg_query_time) + buffer
   

开发指南

# 安装开发依赖
pip install -e .[dev]

# 运行测试
pytest tests/

# 生成文档
cd docs && make html

贡献

欢迎通过 GitHub Issues 报告问题或提交 Pull Request

许可证

MIT License

关键要素说明

1. 徽章系统：显示版本、Python兼容性和许可证信息
2. 多代码块：使用不同语言标签实现语法高亮
3. 配置表格：清晰展示主要参数
4. Web集成示例：展示与Flask的整合
5. 监控集成：提供Prometheus对接示例
6. 开发工作流：明确贡献者指南

建议配合以下内容增强文档：

1. 添加架构图（使用Mermaid语法）
2. 性能基准测试数据
3. 与常用框架（Django、FastAPI）的集成示例
4. 故障排除指南
5. 版本更新日志