FastAPI中如何在终端打印每次请求响应的实际执行SQL查询
在FastAPI中打印原生SQL执行语句的调试方案
直接用HTTP中间件不太合适,因为SQL执行属于数据库操作层面,和HTTP请求的生命周期不完全绑定。下面给你两种实用的实现方式:
方案一:封装执行方法(手动替换参数生成最终SQL)
如果需要直接看到填充完参数的完整SQL,可以封装execute方法,在执行前把占位符替换成实际值再打印。
示例代码
import databases import logging # 初始化数据库连接(根据你的实际数据库URL修改) DATABASE_URL = "sqlite:///./test.db" database = databases.Database(DATABASE_URL) # 配置调试日志 logging.basicConfig(level=logging.INFO) sql_logger = logging.getLogger("sql_debug") async def debug_execute(query: str, values: dict = None): # 仅调试用:替换命名参数为实际值 if values: debug_sql = query for key, val in values.items(): placeholder = f":{key}" # 处理字符串类型的参数,添加单引号并转义内部单引号 if isinstance(val, str): escaped_val = val.replace("'", "''") debug_sql = debug_sql.replace(placeholder, f"'{escaped_val}'") else: debug_sql = debug_sql.replace(placeholder, str(val)) sql_logger.info(f"执行的SQL: {debug_sql}") # 执行实际的数据库操作 return await database.execute(query, values)
在路由中使用
from fastapi import FastAPI app = FastAPI() # 启动/关闭数据库连接 @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() @app.post("/users/") async def create_user(firstname: str, lastname: str): await debug_execute( '''INSERT INTO users(firstname, lastname) VALUES (:firstname, :lastname)''', {"firstname": firstname, "lastname": lastname} ) return {"msg": "用户创建成功"}
方案二:利用数据库库的原生日志(更安全)
如果你用的是databases库(FastAPI常用的原生SQL工具),可以直接开启它的DEBUG日志,不需要自己封装,就能打印原始SQL和对应的参数。
示例代码
import logging import databases # 开启databases的DEBUG日志 logging.getLogger("databases").setLevel(logging.DEBUG) # 初始化数据库连接 DATABASE_URL = "sqlite:///./test.db" database = databases.Database(DATABASE_URL)
运行后终端会输出类似内容:
DEBUG:databases:Executing SQL: INSERT INTO users(firstname, lastname) VALUES (:firstname, :lastname) DEBUG:databases:Parameters: {'firstname': 'Alice', 'lastname': 'Smith'}
注意点
- 方案一的参数替换仅用于调试,生产环境绝对不要用这种方式拼接SQL,会有SQL注入风险。
- 方案二更安全,适合生产环境临时开启调试,不会修改原始SQL执行逻辑。
内容的提问来源于stack exchange,提问作者Ajay
相关产品推荐
相关产品推荐

