关于路径操作函数使用依赖方法时重复实例化对象的问题

[linwanlong]

# 专有的依赖项，不适用于通过工厂函数来构造，就单独写
def get_hero_service(
    session: AsyncSession = Depends(get_db), redis: Redis = Depends(get_cache_redis)
) -> HeroService:
    """Dependency for getting HeroService instance."""
    repository = HeroRepository(session)
    return HeroService(repository, redis)

如上代码逻辑在路径操作函数中，陆续收到多个请求时，会多次重复的实例化 HeroRepository HeroService，既然是无状态对象，为什么不能保证全局只有一个 HeroRepository 和 HeroService呢，这样的话还能减少性能开销。

[acelee0621]

这个其实也是可以做成全局的，直接在服务层底部写一个 hero_service= Herodervice() 这种，用起来也没问题，但是通过依赖来请求还是更加方便隔离，这个性能开销和通过依赖来获取数据库连接类似，几乎没有性能开销的。换来的是代码结构清晰，避免了全局变量，这个的问题在于如果不写工厂函数，那就得在每个路由层里面或者其他地方写好获取某个服务的依赖函数来使用，稍微有点麻烦。我代码里写了个工厂函数，使用工厂函数还是更推荐的做法。对于这个问题我这么说不够细致，我来复制一段 gpt 的详细解释，你参考一下：

TL;DR：
在 FastAPI 中像 HeroRepository 和 HeroService 这样通过依赖注入生成的对象，本质上是请求作用域（request-scoped）依赖。它们虽然看似“无状态”，但依赖于 AsyncSession（数据库会话）和 Redis 连接，这些底层资源与请求生命周期绑定，必须避免跨请求共享。重复实例化的性能开销极小，换取的是线程安全、连接隔离和错误边界。

---

Step-by-step 解答

1. FastAPI 的依赖注入机制

   • 每次请求进来时，FastAPI 会解析依赖树，调用 get_hero_service。
   • get_db 提供一个独立的数据库会话，get_cache_redis 提供一个 Redis 连接（通常是从连接池里获取，而不是新建）。
   • 因此，每次请求 HeroService → HeroRepository 的构造，都是和请求绑定的。

2. 为什么不能全局单例？

   • 数据库会话 (AsyncSession)：它不是线程安全的，必须请求内独占。全局化会导致不同请求共享一个会话 → 数据污染、事务错乱、并发异常。
   • Redis：虽然连接池是全局的，但连接对象是短暂借用的，仍然需要在依赖中获取。
   • Service/Repository 层：表面上“无状态”，但实际上持有了 session 等有状态对象，所以必须随着请求生命周期构建和销毁。

3. 性能开销是不是问题？

   • 你担心的多次实例化 HeroRepository / HeroService，其实只是在 Python 层分配一个轻量对象。
   • 真正重的资源是 DB 连接和网络 IO，这些 FastAPI 已经通过连接池优化过。
   • 换句话说，多次实例化 Service/Repository 的开销几乎可以忽略，甚至小于一次字典查找。

4. 最佳实践

   • Repository/Service 这种“胶水层”：保持请求作用域（request-scoped），安全且清晰。

   • 全局单例化的资源：只对无状态且线程安全的对象（例如配置、纯函数工具类、全局缓存池）才合适。

   • 这也是 DDD（领域驱动设计）或分层架构常见的模式：

     • Infra 层（DB session, Redis client） → Scoped
     • Service 层 → Scoped，因为依赖 Scoped
     • 工具类 → 可以 Singleton

---

✅ 一句话总结：
HeroService/HeroRepository 不是单纯“无状态对象”，它们持有请求级别的数据库会话和缓存连接，所以必须随着请求重复实例化，才能保证数据隔离和线程安全；至于性能开销，几乎可以忽略不计。

[linwanlong]

我清楚了 依赖链中 get_db 并不是无状态的，并且 HeroRepository 与 HeroService 实例化的开销很小可以忽略。并且原写法结构层次清楚便于维护。

[acelee0621]

我在后来的关于 Lifespan State 那篇文章里又写了一个新的获取数据库的写法，就是通过 request.state 来拿，我也是最近才学到的，我感觉这个方式更好点，完全避免了数据库连接的全局变量，这个代码库里的数据库定义文件里还是有全局变量，感觉新的方法更好一点，也是几乎没有性能开销。
首先定义数据库连接和关闭函数，以及依赖：

# --- 1. 数据库初始化和关闭函数 ---
async def setup_database_connection() -> tuple[
    AsyncEngine, async_sessionmaker[AsyncSession]
]:
    """
    初始化数据库引擎和会话工厂。
    lifespan 会在 FastAPI 启动时调用本函数。
    """
    # 从配置中获取定制参数
    engine_options = settings.SQLALCHEMY_ENGINE_OPTIONS.copy()

    # 创建异步引擎
    engine: AsyncEngine = create_async_engine(
        settings.DATABASE_URL,
        **engine_options,
    )

    # 创建会话工厂
    session_factory = async_sessionmaker(
        class_=AsyncSession, expire_on_commit=False, bind=engine
    )

    logger.info(f"数据库引擎和会话工厂已创建: {settings.DB_TYPE}")
    return engine, session_factory


async def close_database_connection(engine: AsyncEngine) -> None:
    """
    关闭数据库引擎连接池。
    lifespan 会在 FastAPI 关闭时调用本函数。
    """
    await engine.dispose()
    logger.info("数据库引擎连接池已关闭。")


# --- 2. 依赖注入函数 ---
async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
    """
    为每个请求或任务提供数据库会话。
    它依赖于 lifespan 中初始化的 session_factory。
    """

    session_factory = cast(
        async_sessionmaker[AsyncSession], request.state.session_factory
    )  # 显式告诉类型检查器 request.state.session_factory 是 async_sessionmaker[AsyncSession]
    async with session_factory() as session:
        yield session

然后在 lifespan 里挂到 request.state 上面去：

class AppState(TypedDict):
    """
    定义应用生命周期中共享状态的结构。
    这为类型检查器和编辑器提供了明确的类型信息。
    """

    engine: AsyncEngine
    session_factory: async_sessionmaker[AsyncSession]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[AppState]:
    # -------- 启动 --------
    get_settings()
    engine, session_factory = await setup_database_connection()
    logger.info("🚀 应用启动，数据库已就绪。")
    await create_db_and_tables(engine)   

    # 更加类型安全的写法
    yield AppState(
        engine=engine,
        session_factory=session_factory,
    )

    # -------- 关闭 --------
    await close_database_connection(engine)
    logger.info("应用关闭，资源已释放。")

这个感觉更方便，以后要挂什么 Redis 或者 RabbitMQ 之类的，只要写好连接关闭函数，在 AppState 里面写好类型，在 lifespan 里面挂上去，写个依赖从 request.state 里面拿一下就完事。

[linwanlong]

这个我在你的文章中看到啦，结合 fastapi 的官方文档来看，这确实是最优雅的实现方式。