Ninject 🥷

Ninject使用现代Python特性来提供简单且高效的依赖注入框架。

• 安装
• 基本用法
• 提供者的类型
• 组合提供者
• 提供不同类型的值
• 提供静态值
• 具有依赖关系的提供者
• 提供多个依赖
• 并发提供依赖
• 混合异步和同步提供者

安装

pip install ninject

基本用法

import ninject as n
from dataclasses import dataclass


# Define a type to be used as a dependency

@dataclass
class Config:
    greeting: str
    recipient: str


# Define a provider for the dependency

@n.provider
def provide_config() -> Config:
    return Config("Hello", "World")


# Injec the dependency into a function

@n.inject
def make_message(*, config: Config = n.inject.ed) -> str:
    return f"{config.greeting}, {config.recipient}!"


# Run the function with in the context of the provider

with provide_config():
    assert make_message() == "Hello, World!"

    # Or access the dependency directly

    with n.Current(Config) as config:
        assert config == Config("Hello", "World")

提供者的类型

提供者可以是以下之一

• 返回值的函数
• 产生单个值的生成器
• 产生值的上下文管理器类
• 返回值的异步函数
• 产生单个值的异步生成器
• 产生值的异步上下文管理器类

@n.provider
def sync_function() -> ...:
    return ...


@n.provider
def sync_generator() -> ...:
    try:
        yield ...
    finally:
        pass


@n.provider
class SyncContextManager:
    def __enter__(self) -> ...:
        return ...

    def __exit__(self, *args) -> None:
        pass


@n.provider
async def async_function() -> ...:
    return ...


@n.provider
async def async_generator() -> ...:
    try:
        yield ...
    finally:
        pass


@n.provider
class AsyncContextManager:
    async def __aenter__(self) -> ...:
        return ...

    async def __aexit__(self, *args) -> None:
        pass

组合提供者

您可以使用 | 将提供者组合起来，以便它们可以一起激活

from dataclasses import dataclass
import ninject as n


@dataclass
class GreetingConfig:
    greeting: str
    recipient: str


@dataclass
class FarewellConfig:
    farewell: str
    recipient: str


@n.provider
def provide_greeting_config() -> GreetingConfig:
    return GreetingConfig("Hello", "Bob")


@n.provider
def provide_farewell_config() -> FarewellConfig:
    return FarewellConfig("Goodbye", "Bob")


provide_all_configs = provide_greeting_config | provide_farewell_config


@n.inject
def make_message(
    *,
    greeting_config: GreetingConfig = n.inject.ed,
    farewell_config: FarewellConfig = n.inject.ed,
) -> str:
    greeting_str = f"{greeting_config.greeting}, {greeting_config.recipient}!"
    farewell_str = f"{farewell_config.farewell}, {farewell_config.recipient}!"
    return f"{greeting_str} ... {farewell_str}"


with provide_all_configs():
    assert make_message() == "Hello, Bob! ... Goodbye, Bob!"

链中的最后一个提供者将覆盖任何具有相同类型的先前提供者。

@n.provider
def provide_bob_greeting_config() -> GreetingConfig:
    return GreetingConfig("Hello", "Bob")


@n.provider
def provide_alice_greeting_config() -> GreetingConfig:
    return GreetingConfig("Hi", "Alice")


provide_greeting_config = provide_bob_greeting_config | provide_alice_greeting_config


with provide_greeting_config:
    with n.current(GreetingConfig) as config:
        assert config == GreetingConfig("Hi", "Alice")

您还可以在同一个 with 语句中分别激活它们，但如果您的 提供者有依赖关系，则顺序很重要

提供内置类型

提供易于区分的类型很重要。对于内置类型，您可以使用 NewType 定义一个新的子类型。在下面的示例中，Greeting 和 Recipient 都是Ninject识别的不同的 str 子类型

from typing import NewType
import ninject as n

Greeting = NewType("Greeting", str)
Recipient = NewType("Recipient", str)


@n.provider
def provide_greeting() -> Greeting:
    return Greeting("Hello")


@n.provider
def provide_recipient() -> Recipient:
    return Recipient("World")

这样，您可以使用内置类型作为依赖

@n.provider
def provide_message(*, greeting: Greeting = inject.ed, recipient: Recipient = inject.ed) -> str:
    return f"{greeting}, {recipient}!"

提供静态值

为此，您可以使用 let 上下文

from dataclasses import dataclass
import ninject as n


@dataclass
class Config:
    greeting: str
    recipient: str


@n.inject
def make_message(*, config: Config = n.inject.ed) -> str:
    return f"{config.greeting}, {config.recipient}!"


with n.let(Config(greeting="Hello", recipient="World")):
    assert make_message() == "Hello, World!"

当使用类型别名或NewType定义依赖时，请分别传递类型和值

from typing import NewType
import ninject as n

Greeting = NewType("Greeting", str)
Recipient = NewType("Recipient", str)


@n.inject
def make_message(*, config: Config = n.inject.ed) -> str:
    return f"{config.greeting}, {config.recipient}!"


with (
    n.let(Greeting, "Hello"),
    n.let(Recipient, "World"),
):
    assert make_message() == "Hello, World!"

具有依赖关系的提供者

提供者可以有自己的依赖

from dataclasses import dataclass
from typing import NewType
import ninject as n


@dataclass
class Config:
    greeting: str
    recipient: str


Message = Dependency("Message", str)


@n.provider
def provide_config() -> Greeting:
    return Config("Hello", "World")


@n.provider
def provide_message(*, config: Config = n.inject.ed) -> Message:
    return Message(f"{greeting}, {recipient}!")


@n.inject
def print_message(*, message: Message = n.inject.ed):
    print(message)


if __name__ == "__main__":
    with provide_config(), provide_message():
        print_message()

输出将是

Hello, World!

提供多个依赖

单个提供者可以通过返回元组来提供多个依赖

from typing import NewType
import ninject as n

Greeting = NewType("Greeting", str)
Recipient = NewType("Recipient", str)
MessageContent = tuple[Greeting, Recipient]


@n.provider
def provide_message_content() -> MessageContent:
    return "Hello", "World"


@n.inject
def print_message(*, greeting: Greeting = inject.ed, recipient: Recipient = inject.ed):
    print(f"{greeting}, {recipient}!")


if __name__ == "__main__":
    with provide_message_content():
        print_message()

您也可以直接依赖于元组，在这种情况下是MessageContent

from typing import NewType
import ninject as n

Greeting = NewType("Greeting", str)
Recipient = NewType("Recipient", str)
MessageContent = tuple[Greeting, Recipient]


@n.provider(MessageContent)
def provide_message_content() -> dict:
    return {"greeting": "Hello", "recipient": "World"}


@n.inject
def print_message(*, message_content: MessageContent = inject.ed):  # TypeError!
    greeting, recipient = message_content
    print(f"{greeting}, {recipient}!")



if __name__ == "__main__":
    with provide_message_content():
        print_message()

并发提供依赖

Ninject不会并发执行异步提供者，因为这样做如果不需要可能会给异步函数调用增加大量的开销。如果您想并发满足依赖，可以利用一次提供多个依赖的能力。考虑到这一点，您可以使用asyncio.gather在返回依赖之前并发运行多个异步函数

import asyncio
from typing import NewType
import ninject as n

Greeting = NewType("Greeting", str)
Recipient = NewType("Recipient", str)
MessageContent = tuple[Greeting, Recipient]


async def get_message() -> str:
    ...  # Some async operation
    return "Hello"


async def get_recipient() -> str:
    ...  # Some async operation
    return "World"


@n.provider
async def provide_message_content() -> MessageContent:
    return tuple(await asyncio.gather(get_message(), get_recipient()))


@n.inject
async def print_message(*, greeting: Greeting = inject.ed, recipient: Recipient = inject.ed):
    print(f"{greeting}, {recipient}!")


if __name__ == "__main__":
    with provide_message_content():
        asyncio.run(print_message())

混合异步和同步提供者

只要它们在异步上下文中使用，允许混合同步和异步提供者

import asyncio
from typing import NewType
import ninject as n

Recipient = NewType("Recipient", str)
Message = NewType("Message", str)


@n.provider
async def provide_recipient() -> Recipient:
    return Recipient("World")


@n.provider
def provide_message(*, recipient: Recipient = inject.ed) -> Message:
    return Message(f"Hello, {recipient}!")


@n.inject
async def print_message(*, message: Message = inject.ed):
    print(message)


if __name__ == "__main__":
    with provide_recipient(), provide_message():
        asyncio.run(print_message())

如果print_message是同步的，则会引发以下错误

RuntimeError: Cannot use an async context manager in a sync context