🔹 数据库连接不能每请求新建/销毁

🔹 沙箱生命周期要跨请求保持

🔹 用户上下文要在多个 Service 间共享

🔹 不同部署环境（本地 / GCP / SaaS）需要不同实现

📄 openhands/app_server/services/injector.py (第 1-26 行)

class InjectorState:
    """Mutable state shared across injectors within a single request."""
    pass


class Injector(Generic[T], ABC):
    """Base class for all dependency injectors."""

    @abstractmethod
    async def inject(
        self, state: InjectorState, request: Request | None = None
    ) -> AsyncGenerator[T, None]:
        """Inject a dependency into the request context.

        Args:
            state: Mutable state shared across all injectors in this request.
            request: The FastAPI request object.

        Yields:
            An instance of type T to be injected."""

    def context(
        self, state: InjectorState, request: Request | None = None
    ) -> AsyncContextManager[T]:
        """Wrap inject() as an async context manager."""
        return asynccontextmanager(self.inject)(state, request)

    def depends(self, *, scope: str | None = None) -> Callable[..., T]:
        """Create a FastAPI dependency for this injector."""
        return Depends(self.context, scope=scope)

🔹 inject() 是抽象方法——每个 Injector 子类必须实现，签名固定：接收 state + request，yield 出服务实例

🔹 context() 将 inject() 包装为 AsyncContextManager，可在任意 async with 中使用

🔹 depends() 生成 FastAPI 的 Depends 对象，直接挂到路由函数参数上

🔹 InjectorState 是一个空壳——纯粹作为请求级共享状态的"钩子"，通过 setattr/getattr 挂载任意属性

📄 openhands/app_server/services/db_session.py (第 313-341 行)

DB_SESSION_ATTR = 'db_session'
DB_SESSION_KEEP_OPEN_ATTR = 'db_session_keep_open'

async def inject(self, state, request):
    # 1. 检查 state 上是否已有 session（跨 injector 共享）
    db_session = getattr(state, DB_SESSION_ATTR, None)
    if db_session:
        yield db_session  # 复用已有 session
        return
    # 2. 创建新 session 并写入 state
    db_session = session_maker()
    try:
        setattr(state, DB_SESSION_ATTR, db_session)
        yield db_session
        if not getattr(state, DB_SESSION_KEEP_OPEN_ATTR, False):
            await db_session.commit()
    except Exception:
        await db_session.rollback()
        raise
    finally:
        # 清理 state 上的引用
        if hasattr(state, DB_SESSION_ATTR):
            delattr(state, DB_SESSION_ATTR)
        await db_session.close()

🔹 同一个请求中，多个 Service 需要 DB session 时只创建一次——通过 state 上的属性标记去重

🔹 DB_SESSION_KEEP_OPEN_ATTR 标志位允许跨请求保持连接（如长轮询场景）

🔹 异常时 自动 rollback，finally 中 保证 close——标准的 try-yield-finally 模式

📄 openhands/app_server/services/config.py (第 38-91 行)

class GlobalConfig(BaseModel):
    # === 核心服务 ===
    db: DbSessionInjector
    event: EventServiceInjector
    event_callback: EventCallbackServiceInjector
    sandbox: SandboxServiceInjector
    sandbox_spec: SandboxSpecServiceInjector
    llm_model: LLMModelServiceInjector

    # === 会话管理 ===
    app_conversation: AppConversationServiceInjector
    app_conversation_info: AppConversationInfoServiceInjector
    app_conversation_start_task: AppConversationStartTaskServiceInjector

    # === 用户与安全 ===
    user: UserContextInjector
    jwt: JwtServiceInjector | None = None

    # === 基础设施 ===
    httpx: HttpxClientInjector
    pending_message: PendingMessageServiceInjector

    # === 生命周期 ===
    lifespan: AppLifespanService | None = None

    def configure(self) -> None:
        """Validate all injectors are properly initialized."""
        for name in dir(self):
            injector = getattr(self, name)
            if isinstance(injector, Injector):
                injector.configure()

📄 openhands/app_server/services/config.py (第 264-590 行)

# === API 1: get_* → 返回 AsyncContextManager（手动管理）===
def get_db_session_service(state, request) \
    -> AsyncContextManager[AsyncSession]:
    injector = get_global_config().db
    return injector.context(state, request)

def get_sandbox_service(state, request) \
    -> AsyncContextManager[SandboxService]:
    injector = get_global_config().sandbox
    return injector.context(state, request)

# === API 2: depends_* → 返回 FastAPI Depends（自动注入）===
def depends_db_session_service():
    injector = get_global_config().db
    return Depends(injector.depends)

def depends_sandbox_service():
    injector = get_global_config().sandbox
    return Depends(injector.depends)

# 实际使用示例：
@app.post('/api/v1/sessions')
async def create_session(
    db: AsyncSession = Depends(depends_db_session_service()),
    sandbox: SandboxService = Depends(depends_sandbox_service()),
):
    ...

🔹 get_* → 返回 AsyncContextManager，适合在 Service 内部用 async with get_db_session_service(state) 手动管理

🔹 depends_* → 返回 FastAPI 的 Depends 对象，框架自动注入到路由函数参数，零样板代码

6.1 DbSessionInjector — 数据库注入器

DbSessionInjector 是最典型的 Injector——管理数据库连接池和 session 生命周期：

class DbSessionInjector(BaseModel, Injector[AsyncSession]):
    persistence_dir: Path
    host: str | None = None        # None → SQLite 本地模式
    port: int | None = None
    name: str | None = None
    user: str | None = None
    password: SecretStr | None = None
    pool_size: int = 25
    max_overflow: int = 10
    pool_recycle: int = 1800
    gcp_db_instance: str | None = None  # GCP Cloud SQL 专用

    @model_validator(mode='after')
    def fill_empty_fields(self):
        # 环境变量覆盖：DB_HOST, DB_PORT, DB_NAME 等
        if self.host is None:
            self.host = os.getenv('DB_HOST')
        if self.port is None:
            self.port = int(os.getenv('DB_PORT', '5432'))
        ...
        return self

三种数据库后端自动切换：

6.2 HttpxClientInjector — HTTP 客户端注入器

HttpxClientInjector 展示了 Injector 的"连接池复用"模式——同一个请求内只创建一个 httpx.AsyncClient：

class HttpxClientInjector(BaseModel, Injector[httpx.AsyncClient]):
    timeout: int = Field(default=15)

    async def inject(self, state, request):
        # 1. 检查 state 上是否已有 client
        httpx_client = getattr(state, HTTPX_CLIENT_ATTR, None)
        if httpx_client:
            yield httpx_client
            return
        # 2. 创建新 client
        httpx_client = httpx.AsyncClient(timeout=self.timeout)
        try:
            setattr(state, HTTPX_CLIENT_ATTR, httpx_client)
            yield httpx_client
        finally:
            if not getattr(state, HTTPX_CLIENT_KEEP_OPEN_ATTR, False):
                if hasattr(state, HTTPX_CLIENT_ATTR):
                    delattr(state, HTTPX_CLIENT_ATTR)
                await httpx_client.aclose()

这个模式与 DbSessionInjector 完全一致——state 属性去重 → yield → finally 清理。OpenHands 所有 Injector 都遵循这个模板。

🔹 每个 Service 接口有一个 Injector 抽象类（如 EventServiceInjector）

🔹 不同后端/部署模式有各自的 Injector 子类（如 FilesystemEventServiceInjector vs AwsEventServiceInjector）

🔹 GlobalConfig 在启动时实例化具体子类，运行时所有路由共享同一套 Injector

Injector 模式核心要点

🔹 一个基类（Injector[T]）+ 一个状态总线（InjectorState）+ 一个注册中心（GlobalConfig）= 完整的 DI 系统

🔹 所有 Service 遵循统一模板：state 属性去重 → yield → finally 清理

🔹 双 API 设计：get_* 用于手动管理，depends_* 用于 FastAPI 自动注入

🔹 继承树实现多态后端切换（SQLite/PostgreSQL/GCP、Docker/Remote/Process 等）

🔹 比第三方 DI 框架更轻、更贴合 FastAPI 异步模型，且配置通过 Pydantic 可序列化