A Pythonic dependency injection framework with auto-wiring, scoped lifecycles, and async support
Project description
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生成可序列化的零参数子类 - 子容器 — 继承父容器绑定,独立覆盖
- 零依赖 — 核心库无第三方依赖
安装
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)
目录
- 自动装配
- 三种作用域
- 四种注册方式
- 命名绑定
- 配置注入
- 多重绑定
- Map 绑定
- 模块系统
- 私有模块
- 工厂函数与资源清理
- Assisted 注入
- ProviderWrapper 懒注入
- AOP 方法拦截
- 异步支持
- 子容器
- 启动时验证
- 分布式计算集成
- 异常类型
- API 参考
- 示例
- 项目结构
- 开发
自动装配
容器通过检查 __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 & m2、m1 | 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) |
安装 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 |
基本自动装配和 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
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
55336f0b5a08b871cb3c36e0ee59ff38f591a8ba47db34118cf0efa2ee1c99a8
|
|
| MD5 |
52e2cd9897f55ee5b3a6eea8d6bb4d67
|
|
| BLAKE2b-256 |
66f5ec2df1006da8ffceb7df80e61cf04cfe35e0f33fcfb4c1aa5b6cea0f7058
|
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
edda1448816dc98117572a64452fa1617c44fad01c77a45d0b3a0126f356f1f1
|
|
| MD5 |
6b5f35eb74aa5f7e375034f18a468c77
|
|
| BLAKE2b-256 |
7e5babe9067cef4b59ea1c9b65f502e6d4e3d71aa6abb3c244083f9ecb143bcc
|