如何让SqlAlchemy的async_sessionmaker不再返回Decimal对象

方法一：修改asyncmy连接参数（推荐） #

asyncmy驱动默认会将MySQL的DECIMAL、NUMERIC类型转换为Python Decimal对象，你可以通过设置连接参数convert_decimal=False让驱动直接返回原生整数或浮点数类型。

修改你的create_async_engine代码：

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy import text

engine = create_async_engine(
    f"mysql+asyncmy://{_db_user}:{_db_passwd}@{_db_host}:{_db_port}/{_db_schema}",
    echo=False,
    connect_args={"convert_decimal": False}  # 添加该参数
)

async_session = async_sessionmaker(engine, expire_on_commit=False)

async with async_session() as session:
   result = await session.execute(text(query))
   result = result.mappings().fetchall()

设置后，MySQL的INT/BIGINT等整数类型会返回Python int，DECIMAL/FLOAT/DOUBLE会返回Python float（小数部分为0的DECIMAL会自动转为int）。

方法二：全局覆盖SQLAlchemy的类型映射 #

如果方法一不生效，可全局修改SQLAlchemy对MySQL数值类型的处理逻辑，强制返回原生数值类型：

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy import text
from sqlalchemy.dialects.mysql import INTEGER, DECIMAL, FLOAT

# 重写类型结果处理器
INTEGER.result_processor = lambda self, dialect, coltype: lambda value: value
DECIMAL.result_processor = lambda self, dialect, coltype: lambda value: int(value) if value % 1 == 0 else float(value)
FLOAT.result_processor = lambda self, dialect, coltype: lambda value: float(value)

engine = create_async_engine(
    f"mysql+asyncmy://{_db_user}:{_db_passwd}@{_db_host}:{_db_port}/{_db_schema}",
    echo=False,
)

async_session = async_sessionmaker(engine, expire_on_commit=False)

async with async_session() as session:
   result = await session.execute(text(query))
   result = result.mappings().fetchall()

该方式会全局修改对应类型的结果转换规则，自动将Decimal转为int或float。

方法三：批量转换查询结果 #

若上述方法无法使用，可手动遍历查询结果，将Decimal对象转换为原生数值类型：

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy import text
from decimal import Decimal

def convert_decimals(data):
    if isinstance(data, list):
        return [convert_decimals(item) for item in data]
    elif isinstance(data, dict):
        return {k: convert_decimals(v) for k, v in data.items()}
    elif isinstance(data, Decimal):
        return int(data) if data % 1 == 0 else float(data)
    else:
        return data

async_session = async_sessionmaker(engine, expire_on_commit=False)

async with async_session() as session:
   result = await session.execute(text(query))
   result = result.mappings().fetchall()
   # 转换结果中的Decimal对象
   converted_result = convert_decimals(result)

这种方式无需修改驱动或SQLAlchemy配置，但需要额外的转换步骤，适合临时场景。

内容的提问来源于stack exchange，提问作者wedrano de carvalho