如何创建可自动设置编解码器的asyncpg连接池？

直接用asyncpg连接池的init参数就能搞定——它接受一个异步函数，会在每个新连接建立后自动执行，刚好用来注册hstore、json、jsonb的内置编解码器，完全不用自定义连接类或者搞繁琐的上下文包装。

具体代码实现 #

import asyncpg

async def init_connection(conn):
    # 注册hstore编解码器
    await conn.set_builtin_type_codec(
        'hstore',
        codec_name='pg_catalog.hstore'
    )
    # 注册json编解码器
    await conn.set_builtin_type_codec(
        'json',
        codec_name='json'
    )
    # 注册jsonb编解码器
    await conn.set_builtin_type_codec(
        'jsonb',
        codec_name='jsonb'
    )

# 创建带初始化逻辑的连接池
async def create_db_pool():
    pool = await asyncpg.create_pool(
        user='你的用户名',
        password='你的密码',
        database='你的数据库',
        host='localhost',
        init=init_connection  # 核心：每个新连接都会自动跑这个异步初始化函数
    )
    return pool

为什么这方案最省心 #

• 绕开连接类的限制：不用纠结构造函数不能调异步方法的问题
• 全局生效：不管连接池是新建连接还是复用闲置连接（仅新创建的连接会触发init），所有可用连接都会提前注册好编解码器
• 代码极简：只需要一个初始化函数，传入连接池即可，完全不用额外跟踪连接状态

验证示例 #

拿到连接后直接用对应类型就行，比如：

async def test_codecs():
    pool = await create_db_pool()
    async with pool.acquire() as conn:
        # 测试hstore
        hstore_result = await conn.fetchval("SELECT 'name=>张三, age=>25'::hstore")
        print(hstore_result)  # 直接输出Python字典：{'name': '张三', 'age': '25'}
        
        # 测试jsonb写入与读取
        await conn.execute("INSERT INTO test_json (data) VALUES ($1)", {"role": "admin", "perms": ["read", "write"]})
        jsonb_result = await conn.fetchval("SELECT data FROM test_json LIMIT 1")
        print(jsonb_result)  # 输出Python字典：{'role': 'admin', 'perms': ['read', 'write']}

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