跳转到主要内容

基于注解的依赖注入库

项目描述

PyPI Version Supported Python Versions Build Status Coverage report

andi简化了使用类型注解表达依赖的定制依赖注入机制的实现。

andi作为一个构建块或库,有助于实现依赖注入(因此得名 - 基于注解的依赖注入)。

许可证为BSD 3条款。

安装

pip install andi

andi需要Python >= 3.8.1。

目标

查看以下代表汽车(及汽车本身)各个部分的类

class Valves:
    pass

class Engine:
    def __init__(self, valves):
        self.valves = valves

class Wheels:
    pass

class Car:
    def __init__(self, engine, wheels):
        self.engine = engine
        self.wheels = wheels

以下为构建Car实例的常规方式

valves = Valves()
engine = Engine(valves)
wheels = Wheels()
car = Car(engine, wheels)

这些类之间存在一些依赖关系:汽车需要发动机和车轮来构建,而发动机需要气门。这些是汽车的依赖和子依赖。

问题在于,我们能否有一种自动构建实例的方法?例如,我们能否有一个 build 函数,给定 Car 类或任何其他类,即使该类本身有一些其他依赖项,也能返回一个实例?

car = build(Car)  # Andi helps creating this generic build function

andi 会检查依赖树并创建一个计划,使得创建这样的 build 函数变得简单。

这个计划看起来是这样的

  1. 用空参数调用 Valves

  2. 使用第1步中创建的实例作为参数 valves 调用 Engine

  3. 用空参数调用 Wheels

  4. 用第2步中创建的实例作为参数 engine 和第3步中创建的实例作为参数 wheels 调用 Cars

类型注解

但在 Car 示例中还有一个缺失的部分。 andi 如何知道类 Valves 是构建参数 valves 所必需的?一个初步的想法是使用参数名称作为类名称的提示(就像 pinject 所做的那样),但 andi 选择依靠参数的类型注解。

因此,Car 的类应该被重写为

class Valves:
    pass

class Engine:
    def __init__(self, valves: Valves):
        self.valves = valves

class Wheels:
    pass

class Car:
    def __init__(self, engine: Engine, wheels: Wheels):
        self.engine = engine
        self.wheels = wheels

注意现在有一个显式的注解表明 valves 参数的类型为 Valves(同样适用于 enginewheels)。

andi.plan 函数现在可以创建一个构建 Car 类的计划(现在忽略 is_injectable 参数)

plan = andi.plan(Car, is_injectable={Engine, Wheels, Valves})

这是 plan 变量包含的内容

[(Valves, {}),
 (Engine, {'valves': Valves}),
 (Wheels, {}),
 (Car,    {'engine': Engine,
           'wheels': Wheels})]

注意这个计划与前面章节中描述的 4 步计划完全对应。

根据计划构建

创建一个通用函数,从 andi 生成的计划中构建实例,非常简单

def build(plan):
    instances = {}
    for fn_or_cls, kwargs_spec in plan:
        instances[fn_or_cls] = fn_or_cls(**kwargs_spec.kwargs(instances))
    return instances

让我们看看将这些部件组合在一起。以下代码使用 andi 创建 Car 的一个实例

plan = andi.plan(Car, is_injectable={Engine, Wheels, Valves})
instances = build(plan)
car = instances[Car]

is_injectable

并不是总是希望 andi 管理找到的每个单个注解。相反,通常最好是明确声明哪些类型可以被 andi 处理。参数 is_injectable 允许自定义此功能。

andi 将在出现无法解决的依赖项时引发错误,因为这些依赖项不是可注入的。

通常,通过创建一个继承的基本类来声明注入性是首选的。例如,我们可以创建一个名为 Injectable 的基类作为汽车组件的基类

class Injectable(ABC):
    pass

class Valves(Injectable):
    pass

class Engine(Injectable):
    def __init__(self, valves: Valves):
        self.valves = valves

class Wheels(Injectable):
    pass

然后对 andi.plan 的调用将是这样

is_injectable = lambda cls: issubclass(cls, Injectable)
plan = andi.plan(Car, is_injectable=is_injectable)

函数和方法

依赖注入在应用于函数时也非常有用。想象一下,你有一个名为 drive 的函数,该函数通过 Road 驾驶 Car

class Road(Injectable):
    ...

def drive(car: Car, road: Road, speed):
    ... # Drive the car through the road

必须在调用 drive 函数之前解决依赖项

plan = andi.plan(drive, is_injectable=is_injectable)
instances = build(plan.dependencies)

现在可以调用 drive 函数了

drive(instances[Car], instances[Road], 100)

请注意,speed 参数没有进行注解。因此生成的计划将不会包含它,因为 andi.planfull_final_kwargs 参数默认为 False。否则,将会抛出异常(更多信息请参阅 full_final_kwargs 参数的文档)。

调用驱动函数的另一种替代方法(更通用)

drive(speed=100, **plan.final_kwargs(instances))

dataclasses 和 attrs

andi 支持使用 attrsdataclasses 定义的类。例如,Car 类可以定义如下

# attrs class example
@attr.s(auto_attribs=True)
class Car:
    engine: Engine
    wheels: Wheels

# dataclass example
@dataclass
class Car(Injectable):
    engine: Engine
    wheels: Wheels

使用 attrsdataclass 非常方便,因为它们避免了某些样板代码。

外部提供的依赖项

在某些情况下,可能需要保留对对象实例化的控制权。例如,创建数据库连接可能需要访问一些凭证注册表或从连接池中获取连接,因此您可能希望在不使用常规依赖注入机制的情况下控制构建此类实例。

andi.plan 允许指定哪些类型会被外部提供。让我们看一个例子

class DBConnection(ABC):

    @abstractmethod
    def getConn():
        pass

@dataclass
class UsersDAO:
    conn: DBConnection

    def getUsers():
       return self.conn.query("SELECT * FROM USERS")

UsersDAO 需要数据库连接来运行查询。但连接将从外部连接池提供,因此我们使用 andi.plan 时还使用了 externally_provided 参数

plan = andi.plan(UsersDAO, is_injectable=is_injectable,
                 externally_provided={DBConnection})

然后应稍微修改构建方法,以便能够注入外部提供的实例

def build(plan, instances_stock=None):
    instances_stock = instances_stock or {}
    instances = {}
    for fn_or_cls, kwargs_spec in plan:
        if fn_or_cls in instances_stock:
            instances[fn_or_cls] = instances_stock[fn_or_cls]
        else:
            instances[fn_or_cls] = fn_or_cls(**kwargs_spec.kwargs(instances))
    return instances

现在我们已准备好使用 andi 创建 UserDAO 实例

plan = andi.plan(UsersDAO, is_injectable=is_injectable,
                 externally_provided={DBConnection})
dbconnection = DBPool.get_connection()
instances = build(plan.dependencies, {DBConnection: dbconnection})
users_dao = instances[UsersDAO]
users = user_dao.getUsers()

请注意,对于外部提供的依赖项,不需要实现注入。

可选

在依赖项可能为可选的情况下,可以使用 Optional 类型注解。例如

@dataclass
class Dashboard:
    conn: Optional[DBConnection]

    def showPage():
        if self.conn:
            self.conn.query("INSERT INTO VISITS ...")
        ...  # renders a HTML page

在这个例子中,Dashboard 类生成一个要服务的 HTML 页面,并将访问次数存储到数据库中。在某些环境中,数据库可能不存在,但您可能希望即使在无法记录访问次数的情况下,仪表板也能工作。

当存在数据库连接时,计划调用将是

plan = andi.plan(UsersDAO, is_injectable=is_injectable,
                 externally_provided={DBConnection})

而当连接不存在时,调用将是以下内容

plan = andi.plan(UsersDAO, is_injectable=is_injectable,
                 externally_provided={})

还需要注册 None 类型的注入。否则,andi.plan 将抛出异常,表示“NoneType 不可注入”。

Injectable.register(type(None))

Union

Union 也可以用来表示替代选项。例如

@dataclass
class UsersDAO:
    conn: Union[ProductionDBConnection, DevelopmentDBConnection]

在没有 ProductionDBConnection 的情况下,将注入 DevelopmentDBConnection

注解

在 Python 3.9+ 中,可以使用 Annotated 类型注解来附加任意元数据,这些元数据将在计划中保留。类型注解相同但元数据不同的出现将不会被考虑为重复。例如

@dataclass
class Dashboard:
    conn_main: Annotated[DBConnection, "main DB"]
    conn_stats: Annotated[DBConnection, "stats DB"]

计划将包含这两个依赖项。

自定义构建器

有时依赖项不能直接创建,但需要一些额外的代码来构建。而这部分代码也可能有自己的依赖项

class Wheels:
    pass

def wheel_factory(wheel_builder: WheelBuilder) -> Wheels:
    return wheel_builder.get_wheels()

由于默认情况下 andi 无法知道如何创建 Wheels 实例或计划需要首先创建 WheelBuilder 实例,因此需要使用 custom_builder_fn 参数来告知这一点

custom_builders = {
    Wheels: wheel_factory,
}

plan = andi.plan(Car, is_injectable={Engine, Wheels, Valves},
                 custom_builder_fn=custom_builders.get,
                 )

custom_builder_fn 应该是一个函数,它接受一个类型并返回该类型的工厂。

构建代码还需要知道如何构建Wheels实例。使用自定义构建器构建的对象的计划步骤使用包含在result_class_or_fn属性中要构建的类型和在factory属性中用于构建它的可调用实例的andi.CustomBuilder包装器。

from andi import CustomBuilder

def build(plan):
    instances = {}
    for fn_or_cls, kwargs_spec in plan:
        if isinstance(fn_or_cls, CustomBuilder):
            instances[fn_or_cls.result_class_or_fn] = fn_or_cls.factory(**kwargs_spec.kwargs(instances))
        else:
            instances[fn_or_cls] = fn_or_cls(**kwargs_spec.kwargs(instances))
    return instances

完整最终kwargs模式

默认情况下,如果andi.plan无法提供给定输入的一些直接依赖项,它不会失败(参见上面示例中的一个的speed参数)。

当检查已经知道某些参数不会注入但将通过其他方式(如上面的drive函数)提供的函数时,这种行为是期望的。

但在其他情况下,最好确保所有依赖项都已满足,否则失败。类就是这样。因此,建议在调用andi.plan进行类时设置full_final_kwargs=True

覆盖

让我们回到Car的例子。假设你再次想构建一辆车。但这次你想替换Engine,因为这将是一辆电动车!当然,电动引擎包含电池并且完全没有阀门。这可能是一个新的Engine

class Battery:
    pass

class ElectricEngine(Engine):

    def __init__(self, battery: Battery):
        self.battery = valves

Andi提供在计划时替换依赖项的可能性,这正是构建电动车所需的:我们需要用ElectricEngine替换对Engine的所有依赖项。这正是覆盖提供的内容。让我们看看在这种情况下应该如何调用plan

plan = andi.plan(Car, is_injectable=is_injectable,
                 overrides={Engine: ElectricEngine}.get)

请注意,Andi将适当地展开新的依赖项。也就是说,ValvesEngine不会出现在结果计划中,但ElectricEngineBattery将会。

总的来说,覆盖提供了一种方法,可以在树中的任何位置覆盖默认依赖项,用替代项更改它们。

默认情况下,覆盖不是递归的:覆盖不应用于已覆盖的依赖项的子项。有一个标志可以打开递归,如果需要的话。请参阅andi.plan文档以获取更多信息。

为什么需要类型注解?

andi使用类型注解来声明依赖项(输入)。它具有几个优点,也有一些局限性。

优点

  1. 内置语言功能。

  2. 指定类型时,你并没有撒谎——这些注解仍然像通常的类型注解一样工作。

  3. 在许多项目中,你会注释参数,所以andi的支持是“免费”的。

局限性

  1. 可调用函数不能有两个相同类型的参数。

  2. 此功能可能与常规类型注解的使用发生冲突。

如果你的可调用函数有两个相同类型的参数,请考虑使它们具有不同的类型。例如,一个可调用函数可能接收网页的url和html。

def parse(html: str, url: str):
    # ...

为了使它更好地与andi配合,你可能会为url和html定义单独的类型

class HTML(str):
    pass

class URL(str):
    pass

def parse(html: HTML, url: URL):
    # ...

但这需要更多的样板代码。

为什么andi不处理对象的创建?

目前,andi只是检查可调用函数并选择框架需要创建和传递给可调用函数的最佳具体类型,而不规定如何创建它们。这使得andi在各种环境中都很有用——例如。

  • 创建某些对象可能需要异步函数,并且它可能依赖于使用的库(asyncio、twisted等)

  • 在流式架构中(例如基于Kafka),检查可能在一台机器上进行,而对象的创建可能在分布式系统中的不同节点上进行,然后实际运行可调用功能可能在另一台机器上。

很难设计出足够灵活的API来满足所有这些用例。尽管如此,andi可能会在模式出现后提供更多辅助工具,即使它们仅在特定环境中有用。

示例:基于回调的框架

Spider示例

没有什么比示例更能理解andi如何有用的了。让我们想象一下,你想要实现一个基于回调的框架来编写爬虫以爬取网页。

基本思路是存在一个框架,用户可以在其中编写爬虫。每个爬虫都是一系列回调,可以处理页面数据,发出提取的数据或请求新页面。然后,有一个引擎负责下载网页并调用用户定义的回调,将请求与其对应的回调链式连接。

让我们看看一个爬虫的示例,用于从烹饪页面下载食谱

class MySpider(Spider):
    start_url = "htttp://a_page_with_a_list_of_recipes"

    def parse(self, response):
        for url in recipes_urls_from_page(response)
            yield Request(url, callback=parse_recipe)

    def parse_recipe(self, response):
        yield extract_recipe(response)

如果用户只需通过注释回调参数来定义一些要求,那将是非常方便的。而andi使其成为可能。

例如,特定的回调可能需要访问cookie

def parse(self, response: Response, cookies: CookieJar):
    # ... Do something with the response and the cookies

在这种情况下,引擎可以使用andi来检查parse方法,并检测到需要ResponseCookieJar。然后框架将构建它们并调用回调。

此功能旨在仅在需要时将某些组件注入到用户的回调中。

它还可以更好地封装用户代码。例如,我们可以将食谱提取解耦成它自己的类

@dataclass
class RecipeExtractor:
    response: Response

    def to_item():
        return extract_recipe(self.response)

回调可以定义为

def parse_recipe(extractor: RecipeExtractor):
    yield extractor.to_item()

注意,使用andi,引擎可以创建一个RecipesExtractor实例,并向其提供声明的Response依赖项。

最终,在这样一个框架中使用andi可以为用户提供极大的灵活性并减少样板代码。

Web服务器示例

andi对于实现新的Web框架也是有用的。

让我们想象一个框架,你可以像下面这样声明你的服务器

class MyWeb(Server):

    @route("/products")
    def productspage(self, request: Request):
        ... # return the composed page

    @route("/sales")
    def salespage(self, request: Request):
        ... # return the composed page

前者由两个端点组成,一个用于提供销售摘要页面,另一个用于提供产品列表。

要提供这些页面,可能需要连接到数据库。这种逻辑可以封装在一些类中

@dataclass
class Products:
    conn: DBConnection

    def get_products()
        return self.conn.query("SELECT ...")

@dataclass
class Sales:
    conn: DBConnection

    def get_sales()
        return self.conn.query("SELECT ...")

现在productspagesalespage方法只需声明它们需要这些对象

class MyWeb(Server):

    @route("/products")
    def productspage(self, request: Request, products: Products):
        ... # return the composed page

    @route("/sales")
    def salespage(self, request: Request, sales: Sales):
        ... # return the composed page

然后框架将负责满足这些依赖项。提供的灵活性将是一个巨大的优势。例如,如果需要同时提供销售和产品信息的页面,这将非常容易实现

@route("/overview")
def productspage(self, request: Request,
                 products: Products, sales: Sales):
    ... # return the composed overview page

贡献

使用tox在多个Python版本上运行测试

tox

上面的命令还运行类型检查;我们使用mypy。

变更

0.6.0 (2023-12-26)

  • 放弃对Python 3.5-3.7的支持。

  • 添加对需要使用自定义可调用函数构建的依赖项的支持。

0.5.0 (2023-12-12)

  • 添加对通过typing.Annotated的依赖项元数据支持(需要Python 3.9+)。

  • 添加对覆盖的文档。

  • 添加对Python 3.10-3.12的支持。

  • CI改进。

0.4.1 (2021-02-11)

  • andi.plan中的覆盖支持

0.4.0 (2020-04-23)

  • andi.inspect 现在可以处理类了(它们的 __init__ 方法被检查)

  • andi.planandi.inspect 现在可以处理可以通过 __call__ 方法调用的对象。

0.3.0 (2020-04-03)

  • andi.plan 函数替换了 andi.to_provide

  • 重新编写 README,解释基于 plan 方法的新的方法。

  • andi.inspect 还返回未注释的参数。

0.2.0 (2020-02-14)

  • 更好的 attrs 支持(解决与字符串类型注释相关的问题)。

  • 声明支持 Python 3.8。

  • 更多测试;确保支持 dataclasses。

0.1 (2019-08-28)

初始版本。

项目详情


下载文件

下载适用于您的平台的文件。如果您不确定要选择哪一个,请了解有关 安装包 的更多信息。

源代码分发

andi-0.6.0.tar.gz (30.6 kB 查看哈希值)

上传时间 源代码

构建分发

andi-0.6.0-py3-none-any.whl (18.6 kB 查看哈希值)

上传时间 Python 3

由以下机构支持