FastAPI中IdleInTransactionSessionTimeout导致无法重连问题求助

问题根源 #

PostgreSQL的idle_in_transaction_session_timeout参数会自动终止长时间处于空闲状态的事务连接。此时SQLAlchemy的全局会话仍认为连接有效，但实际连接已被PG断开，后续操作会触发Can't reconnect until invalid transaction is rolled back错误——普通回滚操作因连接失效无法执行，导致会话持续处于无效状态，直到重启服务。

针对性解决方案 #

1. 配置SQLAlchemy连接池主动检测失效连接 #

在database.py中开启连接池预检测和主动回收，从源头避免使用已断开的连接：

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

# 核心配置：pool_pre_ping检测连接有效性，pool_recycle主动回收连接
engine = create_engine(
    "postgresql://user:password@host:port/dbname",
    pool_pre_ping=True,
    pool_recycle=280,  # 小于PG的idle_in_transaction_session_timeout（建议设为300秒）
    pool_size=10,
    max_overflow=20
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 全局会话（若后续能迁移到依赖注入则更优）
db: Session = SessionLocal()

def reset_db_session():
    """强制重置失效的数据库会话"""
    global db
    try:
        # 尝试回滚（即使失败也不影响后续操作）
        db.rollback()
    except Exception:
        pass
    # 关闭旧会话并创建新会话
    db.close()
    db = SessionLocal()

2. 捕获IdleInTransactionSessionTimeout异常并重置会话 #

在执行数据库操作的模块中，专门捕获该异常并触发会话重置，而不是仅做普通回滚：

# query_utils.py
from sqlalchemy.exc import OperationalError
import psycopg2
from database import db, reset_db_session
from models import YourModel

def batch_upsert(data_list):
    try:
        with db.begin():
            # 批量更新/插入逻辑
            db.bulk_insert_mappings(YourModel, data_list)
    except OperationalError as e:
        # 判断是否为PG事务空闲超时错误
        if isinstance(e.orig, psycopg2.errors.IdleInTransactionSessionTimeout):
            reset_db_session()
            raise RuntimeError("事务超时，会话已重置，请重试") from e
        # 其他操作型异常的常规处理
        db.rollback()
        raise
    except Exception as e:
        db.rollback()
        raise

在模型校验器中同样处理：

# object_save_and_get_model.py
from pydantic import validator, BaseModel
from sqlalchemy.exc import OperationalError
import psycopg2
from database import db, reset_db_session
from models import DBReferenceModel

class YourPydanticModel(BaseModel):
    reference_id: int

    @validator('reference_id')
    def validate_reference_exists(cls, value):
        try:
            exists = db.query(DBReferenceModel).filter_by(id=value).first() is not None
            if not exists:
                raise ValueError("引用ID不存在")
            return value
        except OperationalError as e:
            if isinstance(e.orig, psycopg2.errors.IdleInTransactionSessionTimeout):
                reset_db_session()
                raise RuntimeError("会话已重置，请重新提交请求") from e
            raise

3. 迁移到FastAPI依赖注入管理会话（推荐） #

全局会话是事务空闲超时的高发诱因，改用FastAPI的依赖注入，为每个请求分配独立会话，请求结束自动关闭：

# database.py 新增依赖注入方法
from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 路由中使用依赖注入
from fastapi import APIRouter
router = APIRouter()

@router.post("/batch-upsert")
def handle_batch_upsert(data_list: list, db: Session = Depends(get_db)):
    with db.begin():
        db.bulk_insert_mappings(YourModel, data_list)
    return {"status": "success"}

关键注意事项 #

• 确保pool_recycle的值小于PostgreSQL的idle_in_transaction_session_timeout配置（默认是0，需手动设置为300秒左右）
• 批量操作时尽量缩短事务时长，避免在事务中执行非DB操作（如IO、外部API调用）
• 若必须使用全局会话，需定期检查会话状态，避免长时间持有未提交/回滚的事务

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