在Sanic中使用asyncpg连接池：现有实现是否有更优方案？

嘿，你的代码能跑已经很靠谱了！不过确实有不少可以优化的地方，让代码更简洁、更符合Python和Sanic的最佳实践，我来给你拆解下：

1. 消除重复代码，统一连接获取逻辑 #

你现在的fetch和execute方法里都重复写了async with self.pg_pool.acquire() as connection，可以把这部分抽成一个内部辅助方法，减少冗余，后续加新的数据库操作方法（比如fetchrow、fetchval）也能直接复用。

2. 遵循Python命名规范 #

Python类名应该用大驼峰命名法，把pg改成PostgresClient这样的名字，可读性会强很多；另外你的Blueprint命名是dp，看起来像是笔误，改成db更贴合数据库模块的定位。

3. 结合Sanic的上下文管理 #

Sanic本身提供了app.ctx和bp.ctx来存储应用级/蓝图级的上下文对象，我们可以在服务启动时初始化连接池，把数据库客户端实例挂载到上下文里，不用手动传参，更符合Sanic的生态设计。

4. 增加错误处理与日志 #

生产环境中，数据库操作很可能出现异常（比如连接超时、SQL语法错误），加上错误捕获和日志输出，能快速定位问题，同时把异常重新抛出，让上层逻辑可以处理。

5. 增加类型提示 #

给方法参数和返回值加上类型提示，不仅能提升代码可读性，IDE也能提供更精准的代码提示，减少低级错误。

优化后的完整代码示例 #

第一步：在Sanic主应用中初始化连接池

from sanic import Sanic
from asyncpg import create_pool
from user import db_bp

app = Sanic("MyDatabaseApp")

# 服务启动前初始化数据库连接池
@app.before_server_start
async def setup_database_pool(app, loop):
    app.ctx.pg_pool = await create_pool(
        user="your_db_user",
        password="your_db_password",
        database="your_db_name",
        host="localhost",
        port=5432
    )

# 服务停止后关闭连接池
@app.after_server_stop
async def close_database_pool(app, loop):
    await app.ctx.pg_pool.close()

app.blueprint(db_bp)

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

第二步：修改user.py中的数据库客户端与蓝图

from asyncpg import Pool, Connection
from sanic import Blueprint
import logging

# 配置日志
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 修正Blueprint命名
db_bp = Blueprint('db')

class PostgresClient:
    def __init__(self, pool: Pool):
        self.pool = pool

    async def _get_connection(self) -> Connection:
        """内部辅助方法：统一获取数据库连接"""
        return await self.pool.acquire()

    async def fetch(self, sql: str, *args, **kwargs):
        try:
            async with self._get_connection() as conn:
                return await conn.fetch(sql, *args, **kwargs)
        except Exception as e:
            logger.error(f"Fetch操作失败: {str(e)} | SQL语句: {sql} | 参数: {args}")
            raise  # 重新抛出异常，让上层逻辑处理

    async def execute(self, sql: str, *args, **kwargs):
        try:
            async with self._get_connection() as conn:
                return await conn.execute(sql, *args, **kwargs)
        except Exception as e:
            logger.error(f"Execute操作失败: {str(e)} | SQL语句: {sql} | 参数: {args}")
            raise

    # 可以按需扩展其他常用方法，比如fetchrow、fetchval
    async def fetchrow(self, sql: str, *args, **kwargs):
        try:
            async with self._get_connection() as conn:
                return await conn.fetchrow(sql, *args, **kwargs)
        except Exception as e:
            logger.error(f"Fetchrow操作失败: {str(e)} | SQL语句: {sql} | 参数: {args}")
            raise

# 把数据库客户端实例挂载到蓝图上下文
@db_bp.before_server_start
async def setup_db_client(app, loop):
    db_bp.ctx.client = PostgresClient(app.ctx.pg_pool)

# 示例路由：使用数据库客户端
@db_bp.route("/users", methods=["GET"])
async def get_all_users(request):
    users = await request.app.ctx.client.fetch("SELECT * FROM users")
    # 把asyncpg的Record对象转成字典返回
    return request.app.json([dict(user) for user in users])

额外进阶优化建议 #

1. 复用连接减少开销：如果一个请求里需要多次执行数据库操作，可以一次性获取连接并复用，避免多次调用acquire()：

@db_bp.route("/update-user/<user_id:int>", methods=["POST"])
async def update_user(request, user_id):
    new_name = request.json.get("name")
    async with request.app.ctx.client._get_connection() as conn:
        # 先执行更新
        await conn.execute("UPDATE users SET name = $1 WHERE id = $2", new_name, user_id)
        # 再查询更新后的用户信息
        updated_user = await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
    return request.app.json(dict(updated_user))

2. 预编译SQL提升性能：如果有频繁执行的SQL语句，可以预编译（asyncpg支持自动缓存预编译语句），或者手动管理预编译语句，减少数据库解析SQL的开销。

3. 分离SQL语句：把SQL语句放到单独的常量文件或配置文件中，避免硬编码，代码更整洁易维护：

# sql_constants.py
GET_ALL_USERS = "SELECT * FROM users"
UPDATE_USER_NAME = "UPDATE users SET name = $1 WHERE id = $2"

# 在user.py中导入使用
from sql_constants import GET_ALL_USERS, UPDATE_USER_NAME

内容的提问来源于stack exchange，提问作者ZachkGordon