Skip to main content

A Pythonic dependency injection framework with auto-wiring, scoped lifecycles, and async support

Project description

autowire-di

Python License: MIT PyPI

一个 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 生成可序列化的零参数子类
  • 子容器 — 继承父容器绑定,独立覆盖
  • 零依赖 — 核心库无第三方依赖

安装

uv add autowire-di

或使用 pip:

pip install autowire-di

快速开始

from autowire_di import Container, Scope

container = Container()

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

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

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

目录

自动装配

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

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)

三种作用域

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() 每个作用域一个 请求级别的数据库会话、事务

四种注册方式

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

命名绑定

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

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 标记从配置字典中注入值,支持点号路径:

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

多重绑定

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

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]

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

模块系统

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

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() 显式暴露的接口才能被父容器解析。适合封装内部实现细节:

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 之后的代码在作用域结束时清理:

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()

异步生成器同样支持:

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 生成工厂函数:

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] 延迟解析:

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 声明式地为解析出的实例添加横切关注点(日志、事务、缓存、鉴权等),无需修改业务代码:

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 & m2m1 | m2~m 逻辑组合

异步支持

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

异步解析

import asyncio
from autowire_di import Container, Scope

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

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

异步工厂

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 单例

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

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

子容器

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

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

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

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

启动时验证

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

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 端延迟重建容器并解析依赖:

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 配合使用:

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 的场景:

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 序列化:

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

运行示例:

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      # 启动时依赖图静态验证

开发

# 安装开发依赖
uv sync

# 运行测试
uv run pytest

# 类型检查
uv run mypy src/

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

License

MIT

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

autowire_di-0.1.1.tar.gz (72.5 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

autowire_di-0.1.1-py3-none-any.whl (32.6 kB view details)

Uploaded Python 3

File details

Details for the file autowire_di-0.1.1.tar.gz.

File metadata

  • Download URL: autowire_di-0.1.1.tar.gz
  • Upload date:
  • Size: 72.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.10.3 {"installer":{"name":"uv","version":"0.10.3","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"macOS","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

File hashes

Hashes for autowire_di-0.1.1.tar.gz
Algorithm Hash digest
SHA256 55336f0b5a08b871cb3c36e0ee59ff38f591a8ba47db34118cf0efa2ee1c99a8
MD5 52e2cd9897f55ee5b3a6eea8d6bb4d67
BLAKE2b-256 66f5ec2df1006da8ffceb7df80e61cf04cfe35e0f33fcfb4c1aa5b6cea0f7058

See more details on using hashes here.

File details

Details for the file autowire_di-0.1.1-py3-none-any.whl.

File metadata

  • Download URL: autowire_di-0.1.1-py3-none-any.whl
  • Upload date:
  • Size: 32.6 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.10.3 {"installer":{"name":"uv","version":"0.10.3","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"macOS","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}

File hashes

Hashes for autowire_di-0.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 edda1448816dc98117572a64452fa1617c44fad01c77a45d0b3a0126f356f1f1
MD5 6b5f35eb74aa5f7e375034f18a468c77
BLAKE2b-256 7e5babe9067cef4b59ea1c9b65f502e6d4e3d71aa6abb3c244083f9ecb143bcc

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page