如何在Discord Python机器人的Cog中实现SQL单次连接复用

现有代码核心问题 #

1. asyncpg.create_pool是异步方法，在同步的__init__方法中直接调用不会返回可用的连接池，只会返回未等待的协程对象，后续调用execute方法必然报错
2. 单独在Cog中创建连接池会重复占用数据库连接资源，不符合你全局仅维护一个连接的需求

最优实现方案（全局共用同一个连接池） #

第一步：主文件初始化全局连接池 #

在机器人启动入口处一次性创建连接池，挂载到bot实例的属性上，所有Cog都可以直接复用这个池，无需重复创建：

# 主文件（示例为main.py）
import asyncio
import discord
from discord.ext import commands
import asyncpg

async def main():
    intents = discord.Intents.default()
    intents.message_content = True
    bot = commands.Bot(command_prefix="!", intents=intents)
    
    # 全局仅创建一次连接池，挂载到bot实例
    bot.db_pool = await asyncpg.create_pool(dsn='postgres://postgres:Password1@localhost:5434/discord')
    
    # 必须在连接池创建完成后再加载Cog
    await bot.load_extension("你的Cog文件路径，比如cogs.level")
    
    # 机器人退出时自动关闭连接池
    @bot.event
    async def on_close():
        await bot.db_pool.close()
    
    await bot.start("你的机器人Token")

if __name__ == "__main__":
    asyncio.run(main())

第二步：Cog中直接复用全局连接池 #

不需要在Cog中额外创建连接，直接调用self.bot上挂载的连接池即可，注意所有异步数据库操作都要加await，用上下文管理器获取连接避免连接泄漏：

from discord.ext import commands

class LevelCog(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        # 无需在此处创建连接池，直接复用全局实例

    @commands.Cog.listener()
    async def on_ready(self):
        print("Level Online!")

    @commands.command()
    async def test(self, ctx):
        guild = ctx.author.guild
        member = ctx.author
        # 从连接池获取连接，操作完成后自动释放回池
        async with self.bot.db_pool.acquire() as conn:
            await conn.execute('INSERT INTO levels(member, guild_id) VALUES ($1, $2)', member.id, guild.id)
        await ctx.send("数据写入成功")

async def setup(bot):
    await bot.add_cog(LevelCog(bot))

特殊场景方案（Cog需单独使用独立数据库） #

如果当前Cog需要连接和主库不同的独立数据库，不要在__init__中创建连接池，使用Cog自带的异步生命周期方法实现：

from discord.ext import commands
import asyncpg

class LevelCog(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.db_pool = None

    # Cog加载时自动执行，异步创建连接池
    async def cog_load(self):
        self.db_pool = await asyncpg.create_pool(dsn='独立数据库的DSN')

    # Cog卸载时自动关闭连接池，避免资源泄漏
    async def cog_unload(self):
        await self.db_pool.close()

    @commands.command()
    async def test(self, ctx):
        # 用法和全局池一致
        async with self.db_pool.acquire() as conn:
            await conn.execute('INSERT INTO levels(member, guild_id) VALUES ($1, $2)', ctx.author.id, ctx.guild.id)

async def setup(bot):
    await bot.add_cog(LevelCog(bot))

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