如何让asyncpg返回PostgreSQL原始数据，避免自动处理？

要避免asyncpg把查询结果解析成Record或字典，直接拿到PostgreSQL返回的原始JSON文本/字节，有两个靠谱的方案：

方案1：修改SQL查询，强制返回文本类型 #

直接在查询里把JSON结果转成text类型，asyncpg就会跳过JSON解析逻辑，直接返回字符串。把你的查询调整成：

query = "SELECT (ARRAY_AGG(JSON_STRIP_NULLS('my', 'data', 'from', (select array_agg(somewhere) from databasetable))))::text"

然后在生成器里直接取Record的第一个字段（因为cursor返回的是单字段Record）：

async def my_generator():
    async with get_transaction_somehow() as conn:
        async for record in conn.cursor(query, my_data):
            yield record[0]  # 直接输出原始JSON字符串

方案2：全局设置JSON类型编解码器，跳过解析 #

如果不想每次修改查询，可以在asyncpg连接上自定义类型编解码器，让所有JSON/JSONB类型直接返回原始字节或字符串：

async def setup_connection(conn):
    # 设置JSON类型直接返回原始字节（要字符串的话，把lambda里的data改成data.decode()）
    await conn.set_type_codec(
        'json',
        encoder=None,
        decoder=lambda data, _: data,
        format='binary'
    )
    # 同样处理jsonb类型
    await conn.set_type_codec(
        'jsonb',
        encoder=None,
        decoder=lambda data, _: data,
        format='binary'
    )

# 初始化连接池时传入这个setup函数
pool = await asyncpg.create_pool(
    dsn="your_postgres_dsn",
    init=setup_connection
)

之后原有查询不需要改动，直接从Record里取字段就能拿到原始数据：

async def my_generator():
    async with get_transaction_somehow() as conn:
        async for record in conn.cursor(query, my_data):
            yield record[0]  # 返回原始JSON字节，直接输出即可

性能优化逻辑 #

PostgreSQL已经输出了结构化的JSON，asyncpg默认会把它解析成Python字典，你后续又要把字典重新序列化成JSON字符串——这两步属于无意义的重复开销。数据量越大，跳过解析和二次序列化带来的CPU、内存节省越明显。

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