如何关闭SQLAlchemy异常栈追踪中的完整查询打印？

一、关闭控制台的完整参数化查询打印 #

SQLAlchemy的查询日志由Python标准logging模块和引擎配置控制，无需修改sessionmaker，直接在create_async_engine里调整即可：

1. 调高SQLAlchemy引擎日志级别 #

SQLAlchemy的查询日志来自sqlalchemy.engine日志器，把它的级别设为WARNING或更高，就能屏蔽冗长的参数化查询输出：

import logging
from sqlalchemy.ext.asyncio import create_async_engine

# 配置全局日志基础级别
logging.basicConfig(level=logging.INFO)
# 仅让SQLAlchemy引擎打印警告及以上级别的日志
logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING)

# 创建异步引擎
engine = create_async_engine("postgresql+asyncpg://user:pass@host/db")

2. 关闭引擎的echo参数 #

如果之前给create_async_engine传了echo=True，会强制打印所有SQL语句，直接改成echo=False或删除该参数（默认就是False）：

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@host/db",
    echo=False  # 彻底关闭SQL语句打印
)

3. 精细控制：保留SQL结构，屏蔽超长参数 #

要是想保留SQL基本结构，又不想被大量$xxx::CHAR(5)刷屏，可以自定义日志拦截逻辑，只打印参数数量而非完整列表：

from sqlalchemy.engine import Engine
from sqlalchemy import event
import logging

def filter_long_params(conn, cursor, statement, parameters, context, executemany):
    # 批量插入且参数数量超过阈值时，替换参数列表为数量统计
    if executemany and len(parameters) > 100:
        simplified_stmt = f"{statement.split('VALUES')[0]}VALUES [共{len(parameters)}条参数]"
        logging.getLogger('sqlalchemy.engine').info(simplified_stmt)
        return True  # 告诉SQLAlchemy已处理日志，无需默认打印
    return False

# 给所有引擎绑定该事件，拦截查询日志
event.listen(Engine, "before_cursor_execute", filter_long_params)

二、将异常记录到日志服务器 #

如果控制台仍无法承载，或需要持久化异常日志，可将Python日志转发到日志服务器，常用两种方法：

1. 配置日志服务器Handler #

以HTTP接口的日志服务器为例，用Python内置的HTTPHandler发送错误日志，也可使用对应日志服务的SDK（如ELK、Graylog客户端）：

import logging
from logging.handlers import HTTPHandler

# 配置日志服务器Handler，仅发送错误级别的日志
http_handler = HTTPHandler(
    host="你的日志服务器地址",
    url="/接收日志的接口路径",
    method="POST"
)
http_handler.setLevel(logging.ERROR)

# 给SQLAlchemy引擎日志器添加该Handler
logger = logging.getLogger('sqlalchemy.engine')
logger.addHandler(http_handler)

# 全局日志基础配置
logging.basicConfig(level=logging.INFO)

2. 手动捕获异常，精简日志后发送 #

在异步插入代码中主动捕获异常，提取关键信息（错误类型、栈追踪、SQL结构），去掉超长参数后再发送到日志服务器，避免日志臃肿：

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import insert

async def batch_insert(session: AsyncSession, data_list):
    try:
        stmt = insert(你的模型类).values(data_list).on_conflict_do_nothing()
        await session.execute(stmt)
        await session.commit()
    except Exception as e:
        await session.rollback()
        # 提取关键错误信息
        error_type = type(e).__name__
        error_msg = str(e)
        # 获取完整栈追踪
        import traceback
        stack_trace = traceback.format_exc()
        # 简化SQL语句，仅保留结构
        simplified_sql = f"{stmt.compile().split('VALUES')[0]}VALUES [批量参数]"
        # 组合精简日志内容
        full_log = f"批量插入失败: {error_type} - {error_msg}\nSQL结构: {simplified_sql}\n栈追踪:\n{stack_trace}"
        
        # 发送到日志服务器（示例用logging，实际替换为对应服务SDK）
        logging.error(full_log)
        raise  # 可选：是否重新抛出异常供上层处理

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