FastAPI非端点注入时如何获取应用依赖？适配测试环境

我的FastAPI应用定义在main.py中，启动时需要从数据库加载数据到缓存完成初始化。正常运行时，通过get_db()函数获取SqlServer会话，该函数在接口端点中以session=Depends(get_db)的方式注入。

我需要让应用根据当前环境加载正确的get_db函数：运行pytest时使用testconf.py中定义的重写版本，但目前测试时应用无法识别这个重写的依赖，依然用了SqlServer的会话加载缓存，该怎么解决？

相关代码 #

database.py #

engine = create_engine("sqlserver connection string here")
SessionLocal = sessionmaker(autocommit=False,autoflush=False,bind=engine)
def get_db():
   # FastAPI标准的数据库会话获取方式
   db = SessionLocal()
   yield db

main.py #

from database import get_db

# 运行pytest时，这里找不到app依赖中重写的get_db版本
if fn := app.dependency_overrides.get(get_db):
    sess = next(fn())  
else:
    sess = next(get_db())
CacheService.load_cache(sess) 

testconf.py #

# pytest使用的fixture定义
@pytest.fixture 
def engine():
  url = "sqlite://"
  return create_engine(url, ...)

@pytest.fixture
def app_session(engine):
  TestingSessionLocal = sessionmaker(... bind=engine)
  def override_get_db():
      db = TestingSessionLocal()
      yield db
  app.dependency_overrides[get_db] = override_get_db
  return app

@pytest.fixture
def client(app_session):
    return TestClient(app=app_session)

测试用例示例 #

# 某个测试用例
def test_something(app_session, client):
# 这里缓存还是用SqlServer会话加载的，不符合预期
   client.get("url")

问题核心是缓存初始化时机不对：你现在在main.py的全局作用域直接调用get_db加载缓存，这部分代码在pytest的app_session fixture设置依赖重写之前就已经执行了，自然会用原始的SqlServer版本。

具体修复步骤：

1. 把缓存初始化移到FastAPI启动事件中
   FastAPI的@app.on_event("startup")装饰器会在应用启动时执行逻辑，且这部分代码会在依赖重写完成后运行，能自动获取当前生效的get_db版本：
   修改main.py：

   from database import get_db
   from fastapi import Depends
   
   @app.on_event("startup")
   def startup_event(db=Depends(get_db)):
       CacheService.load_cache(db)
   

   这样无论是生产还是测试环境，FastAPI都会通过依赖注入机制使用正确的数据库会话加载缓存。

2. 添加测试后依赖清理逻辑
   为避免测试之间互相干扰，在app_session fixture中添加依赖恢复步骤：

   @pytest.fixture
   def app_session(engine):
       TestingSessionLocal = sessionmaker(... bind=engine)
       def override_get_db():
           db = TestingSessionLocal()
           yield db
       # 保存原始依赖，测试后恢复
       original_get_db = app.dependency_overrides.get(get_db)
       app.dependency_overrides[get_db] = override_get_db
       
       yield app
       
       # 测试结束后还原依赖
       app.dependency_overrides[get_db] = original_get_db
   

3. 确保测试用例先触发fixture执行
   你的测试用例已经注入了app_session参数，这没问题；如果后续有测试用例没显式用app_session，也要把它作为参数传入，保证fixture先执行完成。

额外注意：如果CacheService是全局单例，测试之间要记得清空缓存，避免测试数据污染，可以在app_session fixture的清理步骤中添加缓存重置逻辑。

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