在FastAPI中使用AsyncSession时asyncpg set_type_codec不生效问题

我正在使用FastAPI搭配asyncpg和SQLAlchemy开发应用，为提升请求速度，应用采用「长生命周期」的SQLAlchemy会话以支持PostgreSQL预编译语句（预编译语句仅在当前数据库会话周期内有效）。

在此配置下，数据库查询部分列返回的类型为decimal.Decimal，但我需要将其转为float类型。我尝试按照SQLAlchemy文档中的方法修复，但数据库查询仍返回decimal.Decimal类型。请问我遗漏了什么？

代码示例 #

from fastapi import FastAPI
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

app = FastAPI()

db_engine = create_async_engine(f"postgresql+asyncpg://user:pw@localhost:5432/my_db")

db_session = None 

@app.on_event("startup")
async def startup_event():
    # Generate db session
    global db_session 
    db_session = AsyncSession(db_engine, autocommit=False, future=True)

    # Prepare SQL statements for speeding up request times
    db_session = await prepare_statements(...)

    # Register type codec
    @event.listens_for(db_session.bind.sync_engine, "connect")
    def register_custom_types(dbapi_connection):
        dbapi_connection.run_async(
            lambda connection: connection.set_type_codec(
                'numeric', encoder=str, decoder=float,
                 format='text', schema = "pg_catalog"
            )
        )

环境配置 #

• asyncpg 0.25.0
• sqlalchemy 1.4.17
• fastapi 0.65.1
• python 3.9.4
• Windows & Linux（两者均不生效）

你的问题核心在于类型编解码器注册的时机和对象错误，以下是具体问题和修复步骤：

1. 监听对象错误：绑定了同步引擎而非异步引擎 #

你注册的是db_session.bind.sync_engine的connect事件，但sync_engine是SQLAlchemy用于同步操作的实例，而你使用的是异步引擎create_async_engine，异步连接的事件监听应该绑定到异步引擎本身，而非同步引擎。

2. 时机错误：会话已建立后才注册事件 #

你在startup事件中先创建了db_session，之后才注册connect事件——但connect事件仅在新连接建立时触发，已存在的连接不会回溯触发该事件，导致类型编解码器从未应用到当前使用的长生命周期会话连接上。

修复后的代码 #

调整事件监听的绑定对象，并确保在创建会话前完成注册：

from fastapi import FastAPI
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

app = FastAPI()

db_engine = create_async_engine(f"postgresql+asyncpg://user:pw@localhost:5432/my_db")
db_session = None 

# 先注册异步引擎的连接事件，确保会话创建前生效
@event.listens_for(db_engine, "connect")
async def register_custom_types(dbapi_connection):
    # 直接调用asyncpg的异步set_type_codec方法，无需run_async包裹
    await dbapi_connection.set_type_codec(
        'numeric',
        encoder=str,
        decoder=float,
        format='text',
        schema="pg_catalog"
    )

@app.on_event("startup")
async def startup_event():
    global db_session 
    # 事件注册完成后再创建会话
    db_session = AsyncSession(db_engine, autocommit=False, future=True)
    db_session = await prepare_statements(...)

额外注意事项 #

• 长生命周期会话的连接一旦建立，后续注册的类型编解码器无法生效，必须在会话创建前完成所有编解码器配置。
• asyncpg的set_type_codec是异步方法，在SQLAlchemy 1.4+的异步引擎connect事件中可以直接使用协程函数，不需要run_async包裹。
• 若后续需要修改编解码规则，需重新创建会话（因为预编译语句和类型编解码器均绑定在数据库会话生命周期内）。

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