forze-deps-modules

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Forze dependency modules

Forze 依赖模块

Use when authoring or reviewing infrastructure wiring. For application bootstrap, see

forze-wiring

. For logical names and

StrEnum

route values, see

forze-specs-infrastructure

适用于编写或评审基础设施连接场景。如需了解应用启动相关内容，请查看

forze-wiring

。如需了解逻辑名称与

StrEnum

路由值，请查看

forze-specs-infrastructure

。

Container model

容器模型

Deps

has two registration modes:

Mode	Shape	Use when
`Deps.plain({DepKey: value})`	one provider per key	shared clients, secrets, idempotency defaults
`Deps.routed({DepKey: {route: value}})`	one provider per key + route	specs resolved by `spec.name`
`Deps.routed_group({...}, routes={...})`	same provider for many routes	one backend supports several logical resources

Deps.merge(...)

raises

CoreError

on duplicate plain keys, plain-vs-routed conflicts, or duplicate routed keys. Let it fail fast instead of silently overriding providers.

Deps

包含两种注册模式：

模式	结构	使用场景
`Deps.plain({DepKey: value})`	每个键对应一个提供者	共享客户端、密钥、幂等性默认配置
`Deps.routed({DepKey: {route: value}})`	每个键+路由对应一个提供者	通过 `spec.name` 解析的规格
`Deps.routed_group({...}, routes={...})`	多个路由共用同一提供者	单个后端支持多个逻辑资源

Deps.merge(...)

当出现重复普通键、普通键与路由键冲突或重复路由键时，会抛出

CoreError

。应让其快速失败，而非静默覆盖提供者。

Module shape

模块结构

Built-in modules are generic over

K: str | StrEnum

. Follow that pattern so callers can use

StrEnum

names.

python

from enum import StrEnum
from typing import Mapping, final

import attrs

from forze.application.contracts.base import DepKey
from forze.application.execution import Deps, DepsModule


WidgetClientDepKey = DepKey[WidgetClientPort]("widget_client")
WidgetDepKey = DepKey[WidgetDepPort]("widget")


@final
@attrs.define(slots=True, frozen=True, kw_only=True)
class WidgetDepsModule[K: str | StrEnum](DepsModule[K]):
    client: WidgetClientPort
    widgets: Mapping[K, WidgetConfig] | None = None

    def __call__(self) -> Deps[K]:
        plain = Deps[K].plain({WidgetClientDepKey: self.client})
        routed = Deps[K]()

        if self.widgets:
            routed = Deps[K].routed(
                {
                    WidgetDepKey: {
                        name: ConfigurableWidget(config=config)
                        for name, config in self.widgets.items()
                    }
                }
            )

        return plain.merge(routed)

内置模块基于

K: str | StrEnum

实现泛型。请遵循此模式，以便调用者可以使用

StrEnum

名称。

python

from enum import StrEnum
from typing import Mapping, final

import attrs

from forze.application.contracts.base import DepKey
from forze.application.execution import Deps, DepsModule


WidgetClientDepKey = DepKey[WidgetClientPort]("widget_client")
WidgetDepKey = DepKey[WidgetDepPort]("widget")


@final
@attrs.define(slots=True, frozen=True, kw_only=True)
class WidgetDepsModule[K: str | StrEnum](DepsModule[K]):
    client: WidgetClientPort
    widgets: Mapping[K, WidgetConfig] | None = None

    def __call__(self) -> Deps[K]:
        plain = Deps[K].plain({WidgetClientDepKey: self.client})
        routed = Deps[K]()

        if self.widgets:
            routed = Deps[K].routed(
                {
                    WidgetDepKey: {
                        name: ConfigurableWidget(config=config)
                        for name, config in self.widgets.items()
                    }
                }
            )

        return plain.merge(routed)

Dep factories

依赖工厂

Most spec-backed adapters register factories shaped like:

python

def __call__(self, ctx: ExecutionContext, spec: WidgetSpec) -> WidgetPort:
    return WidgetAdapter(client=ctx.deps.provide(WidgetClientDepKey), spec=spec, config=self.config)

Application code resolves the factory with

route=spec.name

, then calls it with

(ctx, spec)

ExecutionContext

convenience methods do this internally for documents, search, cache, counters, storage, locks, and embeddings.

大多数基于规格的适配器会注册如下形式的工厂：

python

def __call__(self, ctx: ExecutionContext, spec: WidgetSpec) -> WidgetPort:
    return WidgetAdapter(client=ctx.deps.provide(WidgetClientDepKey), spec=spec, config=self.config)

应用代码通过

route=spec.name

解析工厂，然后传入

(ctx, spec)

调用它。

ExecutionContext

的便捷方法会在内部针对文档、搜索、缓存、计数器、存储、锁和嵌入执行此操作。

Lifecycle stays separate

生命周期分离

Keep

DepsModule.__call__

pure and cheap: it builds providers, not network connections. Initialize pools and clients in

LifecycleStep

functions (

postgres_lifecycle_step

redis_lifecycle_step

s3_lifecycle_step

, etc.).

保持

DepsModule.__call__

纯净且轻量化：它仅构建提供者，不创建网络连接。请在

LifecycleStep

函数（如

postgres_lifecycle_step

、

redis_lifecycle_step

、

s3_lifecycle_step

等）中初始化连接池和客户端。

Routed infrastructure

路由式基础设施

For tenant-aware clients, register a structural client port as a plain dependency and let routed clients choose the concrete connection at call time. For routed Postgres clients, pass

introspector_cache_partition_key

when catalog metadata differs by tenant/database.

对于支持租户的客户端，将结构化客户端端口注册为普通依赖，并让路由客户端在调用时选择具体连接。对于路由式Postgres客户端，当目录元数据因租户/数据库而异时，请传入

introspector_cache_partition_key

。

Anti-patterns

反模式

Instantiating adapters directly in handlers — use deps factories and ports.
Using only strings in new modules — keep route type as
```
K: str | StrEnum
```
.
Opening connections in
__call__
— lifecycle owns startup/shutdown.
Returning overlapping keys from multiple modules — compose maps before module construction or use distinct routes.
Registering spec-backed providers as plain deps when multiple specs exist — use routed deps keyed by
```
spec.name
```
.

在处理器中直接实例化适配器 — 请使用依赖工厂与端口。
在新模块中仅使用字符串 — 请将路由类型保持为
```
K: str | StrEnum
```
。
在
__call__
中建立连接 — 生命周期负责启动/关闭操作。
从多个模块返回重叠键 — 请在模块构建前组合映射，或使用不同的路由。
当存在多个规格时，将基于规格的提供者注册为普通依赖 — 请使用以
```
spec.name
```
为键的路由式依赖。

Reference

参考

```
src/forze/application/execution/deps.py
```

src/forze/application/contracts/base/deps.py

src/forze_postgres/execution/deps/module.py

src/forze_redis/execution/deps/module.py

```
src/forze/application/execution/deps.py
```

src/forze/application/contracts/base/deps.py

src/forze_postgres/execution/deps/module.py

src/forze_redis/execution/deps/module.py