Metadata-Version: 2.4
Name: autowire-di
Version: 0.1.1
Summary: A Pythonic dependency injection framework with auto-wiring, scoped lifecycles, and async support
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.11
Description-Content-Type: text/markdown

# autowire-di

[![Python](https://img.shields.io/badge/python-≥3.11-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](LICENSE)
[![PyPI](https://img.shields.io/pypi/v/autowire-di.svg)](https://pypi.org/project/autowire-di/)

一个 Pythonic 的依赖注入框架，支持自动装配（auto-wiring）、作用域生命周期管理、AOP 方法拦截和异步解析。

专为需要将 DI 容器序列化到远程 worker（如 Ray Data、Spark UDF）的分布式计算场景设计。

## 特性一览

- **自动装配** — 通过 `__init__` 类型注解自动构建依赖图，零配置
- **三种作用域** — Transient / Singleton / Scoped，线程安全
- **四种注册方式** — 类注册、工厂函数、实例注册、别名注册
- **命名绑定** — `Annotated[T, Named("name")]` 区分同接口多实现
- **配置注入** — `Annotated[T, Inject(config="a.b.c")]` 点号路径注入
- **多重绑定 & Map 绑定** — `list[T]` / `dict[str, T]` 自动注入
- **模块系统** — `Module` / `PrivateModule` 组织和封装绑定
- **AOP 方法拦截** — 声明式横切关注点，支持 Matcher 组合
- **Assisted 注入** — 部分参数由容器注入，部分由调用方提供
- **ProviderWrapper 懒注入** — 延迟解析，解决作用域不匹配问题
- **异步支持** — `async_resolve`、异步工厂、异步作用域
- **启动时验证** — 一次性检测缺失绑定、循环依赖、作用域不匹配
- **分布式计算集成** — `make_injectable` 生成可序列化的零参数子类
- **子容器** — 继承父容器绑定，独立覆盖
- **零依赖** — 核心库无第三方依赖

## 安装

```bash
uv add autowire-di
```

或使用 pip：

```bash
pip install autowire-di
```

## 快速开始

```python
from autowire_di import Container, Scope

container = Container()

# 接口 -> 实现
container.register(OrderRepository, PostgresOrderRepository)

# 单例生命周期
container.register(DatabasePool, PostgresPool, scope=Scope.SINGLETON)

# 自动装配：根据 __init__ 类型注解自动解析依赖
service = container.resolve(OrderService)
```

## 目录

- [自动装配](#自动装配)
- [三种作用域](#三种作用域)
- [四种注册方式](#四种注册方式)
- [命名绑定](#命名绑定)
- [配置注入](#配置注入)
- [多重绑定](#多重绑定)
- [Map 绑定](#map-绑定)
- [模块系统](#模块系统)
- [私有模块](#私有模块)
- [工厂函数与资源清理](#工厂函数与资源清理)
- [Assisted 注入](#assisted-注入)
- [ProviderWrapper 懒注入](#providerwrapper-懒注入)
- [AOP 方法拦截](#aop-方法拦截)
- [异步支持](#异步支持)
- [子容器](#子容器)
- [启动时验证](#启动时验证)
- [分布式计算集成](#分布式计算集成)
- [异常类型](#异常类型)
- [API 参考](#api-参考)
- [示例](#示例)
- [项目结构](#项目结构)
- [开发](#开发)

## 自动装配

容器通过检查 `__init__` 的类型注解自动构建依赖图，无需手动连接：

```python
from typing import Protocol, runtime_checkable
from autowire_di import Container

@runtime_checkable
class UserRepository(Protocol):
    def find(self, user_id: int) -> dict: ...

class PostgresUserRepository:
    def find(self, user_id: int) -> dict:
        return {"id": user_id}

class UserService:
    def __init__(self, repo: UserRepository) -> None:
        self.repo = repo

container = Container()
container.register(UserRepository, PostgresUserRepository)

# UserService 未注册，但容器能通过类型注解自动构造它
service = container.resolve(UserService)
assert isinstance(service.repo, PostgresUserRepository)
```

## 三种作用域

```python
from autowire_di import Container, Scope

container = Container()

# TRANSIENT（默认）：每次 resolve 创建新实例
container.register(RequestHandler, scope=Scope.TRANSIENT)

# SINGLETON：全局唯一实例，线程安全
container.register(DatabasePool, PostgresPool, scope=Scope.SINGLETON)

# SCOPED：在同一作用域内共享实例，不同作用域间隔离
container.register(DbSession, scope=Scope.SCOPED)

with container.new_scope() as scope:
    s1 = scope.resolve(DbSession)
    s2 = scope.resolve(DbSession)
    assert s1 is s2  # 同一作用域内是同一实例

with container.new_scope() as scope2:
    s3 = scope2.resolve(DbSession)
    assert s3 is not s1  # 不同作用域是不同实例
```

### 作用域生命周期对比

| 作用域 | 创建时机 | 实例数量 | 适用场景 |
|---|---|---|---|
| `TRANSIENT` | 每次 `resolve()` | 每次新建 | 无状态服务、轻量对象 |
| `SINGLETON` | 首次 `resolve()` | 全局唯一 | 数据库连接池、配置对象 |
| `SCOPED` | 首次在作用域内 `resolve()` | 每个作用域一个 | 请求级别的数据库会话、事务 |

## 四种注册方式

```python
container = Container()

# 1. 类注册（ClassProvider）：自动装配 __init__
container.register(ICache, RedisCache)

# 2. 工厂函数（FactoryProvider）：自定义构造逻辑
container.register(ICache, factory=lambda: RedisCache(host="localhost"))

# 3. 实例注册（ValueProvider）：直接绑定已有对象
cache = RedisCache(host="localhost")
container.register(ICache, instance=cache)

# 4. 别名注册（AliasProvider）：将一个接口指向另一个已注册的接口
container.register(ICache, RedisCache)
# 当解析 CacheService 时，实际解析 ICache
```

## 命名绑定

同一接口注册多个实现，通过名称区分：

```python
from typing import Annotated
from autowire_di import Container, Named

container = Container()
container.register(ICache, RedisCache, name="primary")
container.register(ICache, MemoryCache, name="fallback")

# 手动按名称解析
primary = container.resolve(ICache, name="primary")

# 或通过 Annotated 标记自动注入
class OrderService:
    def __init__(
        self,
        cache: Annotated[ICache, Named("primary")],
        fallback: Annotated[ICache, Named("fallback")],
    ) -> None:
        self.cache = cache
        self.fallback = fallback
```

## 配置注入

通过 `Inject` 标记从配置字典中注入值，支持点号路径：

```python
from typing import Annotated
from autowire_di import Container, Inject

container = Container()
container.set_config({
    "db": {"host": "localhost", "port": 5432},
    "cache": {"ttl": 300},
})

class DbConnection:
    def __init__(
        self,
        host: Annotated[str, Inject(config="db.host")],
        port: Annotated[int, Inject(config="db.port")],
    ) -> None:
        self.host = host
        self.port = port

conn = container.resolve(DbConnection)
assert conn.host == "localhost"
assert conn.port == 5432
```

## 多重绑定

同一接口注册多个实现，一次性全部解析：

```python
container = Container()
container.register_multi(EventHandler, LoggingHandler)
container.register_multi(EventHandler, MetricsHandler)
container.register_multi(EventHandler, AlertHandler)

# 解析为列表
handlers = container.resolve_multi(EventHandler)
# -> [LoggingHandler(), MetricsHandler(), AlertHandler()]

# 也可以通过 list[T] 类型注解自动注入
class EventBus:
    def __init__(self, handlers: list[EventHandler]) -> None:
        self.handlers = handlers
```

## Map 绑定

将同一接口的多个实现关联到字符串键，解析为 `dict[str, T]`：

```python
container = Container()
container.register_map(ICache, "redis", RedisCache)
container.register_map(ICache, "memory", MemoryCache)
container.register_map(ICache, "disk", DiskCache)

# 解析为字典
caches = container.resolve_map(ICache)
# -> {"redis": RedisCache(), "memory": MemoryCache(), "disk": DiskCache()}

# 也可以通过 dict[str, T] 类型注解自动注入
class CacheManager:
    def __init__(self, caches: dict[str, ICache]) -> None:
        self.caches = caches
```

## 模块系统

将相关绑定组织为可复用的模块：

```python
from autowire_di import Container, Module, Scope

class InfraModule(Module):
    def configure(self, container: Container) -> None:
        container.register(DatabasePool, PostgresPool, scope=Scope.SINGLETON)
        container.register(ICache, RedisCache, scope=Scope.SINGLETON)
        container.register(MessageQueue, RabbitMQ)

class DomainModule(Module):
    def configure(self, container: Container) -> None:
        container.register(UserRepository, PostgresUserRepository)
        container.register(OrderRepository, PostgresOrderRepository)

container = Container()
container.install(InfraModule())
container.install(DomainModule())
```

## 私有模块

`PrivateModule` 的绑定默认对外不可见，只有通过 `expose()` 显式暴露的接口才能被父容器解析。适合封装内部实现细节：

```python
from autowire_di import Container, PrivateModule, Scope

class PaymentModule(PrivateModule):
    def configure(self, container: Container) -> None:
        container.register(StripeClient, scope=Scope.SINGLETON)
        container.register(PaymentValidator)
        container.register(PaymentService, PaymentServiceImpl)
        self.expose(PaymentService)  # 仅暴露 PaymentService

container = Container()
container.install(PaymentModule())

container.resolve(PaymentService)  # OK
container.resolve(StripeClient)    # ResolutionError — 内部实现不可见
```

## 工厂函数与资源清理

生成器工厂支持 teardown 逻辑——yield 之前的代码创建资源，yield 之后的代码在作用域结束时清理：

```python
from typing import Generator
from autowire_di import Container, Scope

def create_db_session(pool: DatabasePool) -> Generator[DbSession, None, None]:
    session = pool.acquire()
    yield session
    session.close()  # 作用域结束时自动执行

container = Container()
container.register(DbSession, factory=create_db_session, scope=Scope.SCOPED)

with container.new_scope() as scope:
    session = scope.resolve(DbSession)
    # 使用 session ...
# 离开 with 块时自动调用 session.close()
```

异步生成器同样支持：

```python
from typing import AsyncGenerator

async def create_async_session(pool: AsyncPool) -> AsyncGenerator[AsyncSession, None]:
    session = await pool.acquire()
    yield session
    await session.close()

container.register(AsyncSession, factory=create_async_session, scope=Scope.SCOPED)

async with container.new_async_scope() as scope:
    session = await scope.async_resolve(AsyncSession)
```

## Assisted 注入

当一个类的构造参数中，部分由容器注入、部分由调用方提供时，使用 `Assisted` 标记 + `create_factory` 生成工厂函数：

```python
from typing import Annotated
from autowire_di import Container, Assisted

class Payment:
    def __init__(
        self,
        gateway: PaymentGateway,                    # 由容器注入
        amount: Annotated[float, Assisted()],       # 由调用方提供
        currency: Annotated[str, Assisted()],       # 由调用方提供
    ) -> None:
        self.gateway = gateway
        self.amount = amount
        self.currency = currency

container = Container()
container.register(PaymentGateway, StripeGateway)

make_payment = container.create_factory(Payment)
payment = make_payment(amount=100.0, currency="USD")
# gateway 自动注入，amount 和 currency 由调用方传入
```

## ProviderWrapper 懒注入

当依赖创建开销较大，或需要在运行时按需获取多个实例时，使用 `ProviderWrapper[T]` 延迟解析：

```python
from autowire_di import Container, ProviderWrapper, Scope

class ReportGenerator:
    def __init__(self, db_provider: ProviderWrapper[DbSession]) -> None:
        self._db_provider = db_provider

    def generate(self) -> Report:
        db = self._db_provider.get()  # 调用 get() 时才真正解析
        return db.query(...)

container = Container()
container.register(DbSession, PostgresSession, scope=Scope.SCOPED)

# ProviderWrapper[T] 类型注解会被自动识别，注入一个懒包装器
# 这也解决了 Singleton 依赖 Scoped 的作用域不匹配问题
```

## AOP 方法拦截

通过 `bind_interceptor` 声明式地为解析出的实例添加横切关注点（日志、事务、缓存、鉴权等），无需修改业务代码：

```python
from typing import Any
from autowire_di import (
    Container, MethodInterceptor, MethodInvocation,
    annotated_with, any_method, aop_mark,
)

# 1. 定义标记
class Transactional: pass

# 2. 实现拦截器
class LogInterceptor(MethodInterceptor):
    def invoke(self, invocation: MethodInvocation) -> Any:
        print(f"→ {invocation.method.__name__}({invocation.args})")
        result = invocation.proceed()  # 调用下一个拦截器或真实方法
        print(f"← {invocation.method.__name__} = {result}")
        return result

# 3. 用 @aop_mark 标记目标类
@aop_mark(Transactional)
class OrderService:
    def place_order(self, item: str) -> str:
        return f"ordered {item}"

# 4. 绑定拦截器
container = Container()
container.bind_interceptor(
    class_matcher=annotated_with(Transactional),
    method_matcher=any_method(),
    interceptor=LogInterceptor(),
)

service = container.resolve(OrderService)
service.place_order("book")
# → place_order(('book',))
# ← place_order = ordered book
```

### 内置 Matcher

| Matcher | 说明 |
|---|---|
| `any_class()` / `any_method()` | 匹配所有类/方法 |
| `annotated_with(Marker)` | 匹配带有 `@aop_mark(Marker)` 的类/方法 |
| `subclass_of(Base)` | 匹配 `Base` 的子类 |
| `name_matches("get_*")` | 按名称 glob 模式匹配 |
| `m1 & m2`、`m1 \| m2`、`~m` | 逻辑组合 |

## 异步支持

autowire-di 原生支持异步解析和异步工厂函数：

### 异步解析

```python
import asyncio
from autowire_di import Container, Scope

container = Container()
container.register(AsyncService, scope=Scope.SINGLETON)

# 使用 async_resolve 异步解析
service = await container.async_resolve(AsyncService)
```

### 异步工厂

```python
from typing import AsyncGenerator

async def create_async_pool() -> AsyncGenerator[AsyncPool, None]:
    pool = await AsyncPool.create(dsn="postgresql://...")
    yield pool
    await pool.close()

container.register(AsyncPool, factory=create_async_pool, scope=Scope.SINGLETON)

# 必须使用异步作用域
async with container.new_async_scope() as scope:
    pool = await scope.async_resolve(AsyncPool)
```

### 异步 Eager 单例

```python
container = Container()
container.register(AsyncPool, factory=create_async_pool, scope=Scope.SINGLETON, eager=True)

# 异步初始化所有 eager 单例
await container.async_initialize_singletons()
```

## 子容器

创建子容器继承父容器的绑定，可独立覆盖：

```python
parent = Container()
parent.register(ICache, RedisCache)

child = parent.create_child()
child.override(ICache, MemoryCache)  # 仅在子容器中生效

parent.resolve(ICache)  # -> RedisCache
child.resolve(ICache)   # -> MemoryCache
```

## 启动时验证

在应用启动时一次性验证所有绑定，提前发现缺失依赖、循环依赖和作用域不匹配：

```python
container = Container()
container.register(UserService)  # 依赖 UserRepository，但未注册
container.register(CacheService, scope=Scope.SINGLETON)
# CacheService 依赖 DbSession(SCOPED) -> 单例不能依赖作用域服务

container.validate()  # 抛出 ValidationError，包含所有错误
```

检测的问题类型：

- **缺失绑定** — 依赖的接口未注册
- **循环依赖** — A → B → C → A
- **作用域不匹配** — 长生命周期服务依赖短生命周期服务（如 Singleton 依赖 Scoped）

## 分布式计算集成

### 核心问题

在 Ray Data、Spark 等分布式框架中，用户类需要被序列化后发送到远程 worker。但 DI 容器持有的活跃对象（数据库连接、模型权重、线程锁）无法安全序列化。

### `make_injectable`：零参数子类

`make_injectable` 生成一个零参数子类，内部仅捕获可序列化的 **recipe**（注册指令的纯数据快照），在 worker 端延迟重建容器并解析依赖：

```python
container = Container(config={"model": {"name": "bert-base", "device": "cuda"}})
container.register(ModelRegistry, HuggingFaceRegistry)
container.register(Tokenizer)
container.register(MetricsCollector)

# 生成零参数子类
InjectablePredictor = container.make_injectable(TorchPredictor)

# 可以用 overrides 固定特定参数
InjectablePredictor = container.make_injectable(TorchPredictor, device="cpu")
```

与 Ray Data 配合使用：

```python
import ray.data

ds = ray.data.read_csv("data.csv")
ds.map_batches(
    container.make_injectable(TorchPredictor),
    compute=ray.data.ActorPoolStrategy(size=4),
    num_gpus=1,
)
```

### `resolve_kwargs`：手动构造参数

对于需要 `fn_constructor_kwargs` 的场景：

```python
kwargs = container.resolve_kwargs(TorchPredictor)
# kwargs = {"registry": HuggingFaceRegistry(), "tokenizer": Tokenizer(), ...}

# 适用于 Ray Data 的 fn_constructor_kwargs 模式
ds.map_batches(
    TorchPredictor,
    fn_constructor_kwargs=container.resolve_kwargs(TorchPredictor),
)
```

### `ContainerRecipe`：可序列化快照

Recipe 是容器注册指令的纯数据快照，可安全通过 cloudpickle 序列化：

```python
import cloudpickle

recipe = container.recipe

# 序列化到远程 worker
data = cloudpickle.dumps(recipe)
restored_recipe = cloudpickle.loads(data)

# 在 worker 端重建完整容器
rebuilt = restored_recipe.build()
service = rebuilt.resolve(TorchPredictor)
```

### `make_injectable` vs `resolve_kwargs`

| | `make_injectable` | `resolve_kwargs` |
|---|---|---|
| 序列化内容 | Recipe（注册指令） | 已创建的实例 |
| 含不可序列化依赖 | 可以（延迟重建） | 不行（会失败） |
| 适用场景 | Ray `map_batches(cls=)` | Ray `fn_constructor_kwargs` |
| 实例创建时机 | worker 端首次构造时 | 调用方立即创建 |

## 异常类型

所有异常继承自 `DIError`。

| 异常 | 触发场景 |
|---|---|
| `DIError` | 所有 DI 异常的基类 |
| `ResolutionError` | 未注册的接口、缺失的配置键 |
| `CircularDependencyError` | 检测到循环依赖链（A → B → C → A） |
| `RegistrationError` | 重复注册（未使用 override） |
| `ScopeMismatchError` | 长生命周期服务依赖短生命周期服务 |
| `ScopeNotActiveError` | 在作用域外解析 SCOPED 服务 |
| `ValidationError` | `validate()` 发现一个或多个错误 |
| `ExceptionGroup` | `dispose()` 时多个 teardown 出错（Python 3.11+） |

## API 参考

### Container

| 方法 | 说明 |
|---|---|
| `register(interface, impl, *, factory, instance, scope, name, eager)` | 注册绑定 |
| `resolve(interface, *, name)` | 同步解析 |
| `async_resolve(interface, *, name)` | 异步解析 |
| `override(interface, impl, *, factory, instance, scope, name)` | 覆盖已有绑定 |
| `register_multi(interface, impl, *, factory, instance, scope)` | 添加多重绑定 |
| `resolve_multi(interface)` | 解析所有多重绑定为 `list` |
| `register_map(interface, key, impl, *, factory, instance, scope)` | 添加 Map 绑定 |
| `resolve_map(interface)` | 解析所有 Map 绑定为 `dict` |
| `install(module)` | 安装 `Module` 或 `PrivateModule` |
| `new_scope()` | 创建同步作用域（上下文管理器） |
| `new_async_scope()` | 创建异步作用域（异步上下文管理器） |
| `create_child(*, config)` | 创建子容器 |
| `create_factory(cls)` | 生成 Assisted 注入工厂函数 |
| `resolve_kwargs(cls)` | 解析构造参数为 `dict`（不实例化） |
| `make_injectable(cls, **overrides)` | 生成可序列化的零参数子类 |
| `set_config(config)` | 设置配置字典 |
| `validate()` | 启动时验证所有绑定 |
| `initialize_singletons()` | 同步初始化所有 eager 单例 |
| `async_initialize_singletons()` | 异步初始化所有 eager 单例 |
| `dispose()` | 同步清理资源 |
| `async_dispose()` | 异步清理资源 |

### Annotated 标记

| 标记 | 用法 | 说明 |
|---|---|---|
| `Named(name)` | `Annotated[T, Named("x")]` | 按名称选择绑定 |
| `Inject(config=key)` | `Annotated[T, Inject(config="a.b")]` | 注入配置值 |
| `Assisted()` | `Annotated[T, Assisted()]` | 标记为调用方提供的参数 |

### Scope 枚举

| 值 | 说明 |
|---|---|
| `Scope.TRANSIENT` | 每次解析创建新实例（默认） |
| `Scope.SINGLETON` | 全局唯一，线程安全 |
| `Scope.SCOPED` | 作用域内共享，作用域间隔离 |

### AOP

| 类/函数 | 说明 |
|---|---|
| `MethodInterceptor` | 拦截器基类，实现 `invoke(invocation)` |
| `MethodInvocation` | 方法调用上下文，调用 `proceed()` 继续链 |
| `aop_mark(*markers)` | 装饰器，为类/方法附加 AOP 标记 |
| `annotated_with(marker)` | Matcher：匹配带指定标记的目标 |
| `any_class()` / `any_method()` | Matcher：匹配所有 |
| `subclass_of(parent)` | Matcher：匹配子类 |
| `name_matches(pattern)` | Matcher：按 glob 模式匹配名称 |

### Provider 类型

| 类 | 说明 |
|---|---|
| `ClassProvider` | 通过自动装配 `__init__` 创建实例 |
| `FactoryProvider` | 调用工厂函数创建实例，支持生成器 teardown |
| `ValueProvider` | 返回预设实例 |
| `AliasProvider` | 委托到另一个已注册类型 |
| `ProviderWrapper[T]` | 懒包装器，调用 `.get()` 时解析 |

## 示例

`examples/` 目录包含完整的可运行示例：

| 文件 | 内容 |
|---|---|
| [`01_basic_autowiring.py`](examples/01_basic_autowiring.py) | 基本自动装配和 Protocol 接口 |
| [`02_scopes.py`](examples/02_scopes.py) | 三种作用域的生命周期管理 |
| [`03_config_injection.py`](examples/03_config_injection.py) | 配置注入和点号路径 |
| [`04_modules.py`](examples/04_modules.py) | 模块系统和私有模块 |
| [`05_aop_interceptors.py`](examples/05_aop_interceptors.py) | AOP 方法拦截和 Matcher 组合 |
| [`06_async_support.py`](examples/06_async_support.py) | 异步解析和异步工厂 |
| [`07_assisted_injection.py`](examples/07_assisted_injection.py) | Assisted 注入和工厂函数 |
| [`08_advanced_features.py`](examples/08_advanced_features.py) | 多重绑定、Map 绑定、ProviderWrapper |
| [`09_distributed_computing.py`](examples/09_distributed_computing.py) | 分布式计算集成（make_injectable、Recipe） |

运行示例：

```bash
uv run python examples/01_basic_autowiring.py
```

## 项目结构

```
src/autowire_di/
├── __init__.py       # 公开 API 导出
├── container.py      # Container 和 ScopedContainer
├── resolver.py       # 自动装配解析器（类型注解检查、create_factory）
├── registry.py       # 绑定存储（单绑定 + 多重绑定 + Map 绑定）
├── providers.py      # ClassProvider / FactoryProvider / ValueProvider / AliasProvider / ProviderWrapper
├── types.py          # Scope 枚举、Binding 数据类、ResolverProtocol、异常层次
├── markers.py        # Annotated 标记：Named、Inject、Assisted
├── module.py         # Module 和 PrivateModule 基类
├── interceptor.py    # AOP 方法拦截：MethodInterceptor、Matcher、代理生成
├── scope.py          # SingletonCache（线程安全）、ScopedCache
├── recipe.py         # ContainerRecipe：可序列化的容器快照
└── validator.py      # 启动时依赖图静态验证
```

## 开发

```bash
# 安装开发依赖
uv sync

# 运行测试
uv run pytest

# 类型检查
uv run mypy src/

# 代码检查
uv run ruff check src/ tests/
```

## License

MIT
