Azure SQL Serverless数据库断开，SQLAlchemy配置为何未生效？

一、sessionmaker 使用正确性确认 #

如果你的代码是如下常规写法，sessionmaker（SQLModel的Session基于它实现）的使用是正确的：

from sqlmodel import Session, create_engine
from fastapi import Depends

# 初始化引擎，配置连接池参数
engine = create_engine(
    "mssql+pyodbc://<username>:<password>@<server>.database.windows.net/<dbname>?driver=ODBC+Driver+17+for+SQL+Server",
    pool_recycle=3600,
    pool_pre_ping=True
)

# 定义会话依赖
def get_session():
    with Session(engine) as session:
        yield session

# 接口中注入会话
@app.get("/data")
def fetch_data(session: Session = Depends(get_session)):
    # 业务逻辑实现
    pass

这种模式下，每个请求会从连接池获取可用连接，请求结束后自动归还，符合SQLAlchemy的最佳实践。

二、pool_recycle 和 pool_pre_ping 未生效的原因 #

1. pool_recycle 的局限性
   pool_recycle 是基于连接的创建时间来回收连接，但 Azure SQL Serverless 自动暂停是实例级别的完全停止，和连接本身的存活周期无关。比如你设置了1小时回收，但数据库在2小时后暂停，此时池内的连接还没到回收时间，就已经因为实例下线完全失效，pool_recycle 无法提前感知这种外部触发的失效。

2. pool_pre_ping 的触发逻辑缺陷
   pool_pre_ping 会在从池内取出连接前发送SELECT 1测试，但 Azure SQL Serverless 暂停后，第一次测试请求会直接失败（实例未启动），此时 SQLAlchemy 会抛出InvalidatePoolError并标记整个连接池为不可用，不会自动重建连接——它默认认为这种失效是永久性的，而非临时的实例启动过程。

三、可行的解决方案 #

1. 捕获异常并重置连接池 #

在会话依赖中捕获InvalidatePoolError，手动重置连接池后重试：

from sqlalchemy.exc import InvalidatePoolError

def get_session():
    try:
        with Session(engine) as session:
            yield session
    except InvalidatePoolError:
        # 销毁当前失效的连接池
        engine.dispose()
        # 重新创建连接并返回会话
        with Session(engine) as session:
            yield session

2. 缩短 pool_recycle 周期并配置连接重试 #

把pool_recycle缩短到30分钟（1800秒），同时在连接参数中添加Azure SQL专属的重试配置，应对实例启动的延迟：

engine = create_engine(
    "mssql+pyodbc://<connection-string>",
    pool_recycle=1800,
    pool_pre_ping=True,
    connect_args={
        "driver": "ODBC Driver 17 for SQL Server",
        "timeout": 30,  # 延长连接超时，给实例启动留时间
        "RetryCount": 5,  # 重试次数
        "RetryInterval": 2  # 每次重试间隔（秒）
    }
)

3. 全局异常处理优化 #

添加全局异常处理器，捕获数据库不可用的异常，返回友好提示的同时重置连接池：

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import http_exception_handler
from sqlalchemy.exc import InvalidatePoolError

app = FastAPI()

@app.exception_handler(InvalidatePoolError)
async def handle_db_pool_error(request, exc):
    # 重置连接池
    engine.dispose()
    # 返回503服务不可用提示
    return await http_exception_handler(
        request,
        HTTPException(status_code=503, detail="数据库正在启动，请稍后重试")
    )

内容的提问来源于stack exchange，提问作者J-H