打造Python依赖注入实例：助力框架开发根基构建

文章标题：

构建Python依赖注入实例：为框架开发基础搭建助力

文章内容：本文运用Python来创建一个依赖注入框架的示例，达成了数据实体与逻辑实体部分的分离

一、动机:：为何要编写一个Python的依赖注入框架?这难道不是多此一举吗???

是的，但也不完全如此。

举例来说，在fastapi中能够利用依赖注入来达成功能，这是来自fastapi官方文档的一部分代码

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

在这段代码里通过Depends把一个异步函数注入到路径操作函数中，当然这段代码肯定会让很多人产生疑问——为啥不直接在代码里调用相应函数呢。

@app.get("/items/")
async def read_items():
    return await common_parameters()

直接await common_parameters()不就解决问题了嘛，为啥要整这么多让人疑惑的东西呢。要是仅仅处理代码相关的依赖，用函数也能实现依赖关系。

但是，你有千千万万种实现函数调用的办法，那fastapi作为一个框架该怎么确定依赖关系呢？要是没法确定依赖关系，那它就没法给你提供文档化以及各种缓存的服务。

依赖注入框架的意义其实就是通过一种标准化的依赖方案理清组件与组件之间的关系，从而为框架的搭建奠定基础

其实，要是看过spring框架、dotnetCore框架的设计，应该对这都不陌生。

话又说回来，我想要实现一个依赖注入框架的原因其实是我想试着自己打造一个Python编程框架。名字我都想好了🐶，先不说。

二、实现： 实现一个依赖注入框架需要什么

目前我主要借助装饰器来实现，但这会带来运行时的性能开销。之后可能只会把它当作一个注解来用。主要的实现依赖于一个容器。容器在创建的时候会通过获取绑定的类的属性和方法来填充对象exec_impl，在调用容器原来的接口方法时会间接调用exec_impl内的方法。

这是我实现函数注入后调用的代码。

def execute_interface(func: Callable):  
    name = func.__name__  

    def wrapper(self, *args, **kwargs):  
        if not isinstance(self, SystemEntity):  
            raise NotImplementedError(f"please use class derived from SystemEntity")  
        return self.exec_impl[name](self, *args, **kwargs)  

    wrapper.__doc__ = func.__doc__  
    return wrapper  


def depend_interface(depend_entity_name: str):  
    def decorator(func: Callable):  
        func_name = func.__name__  

        def wrapper(self, *args, **kwargs):  
            if not isinstance(self, SystemEntity):  
                raise NotImplementedError(f"please use class derived from SystemEntity")  
            if depend_entity_name not in self.depend_entities.keys():  
                raise NotImplementedError(  
                    f"{func_name} depend on {depend_entity_name}, but {depend_entity_name} not found")  
            return self.depend_entities[depend_entity_name].exec_impl[func_name](self, *args, **kwargs)  

        wrapper.__doc__ = func.__doc__  
        return wrapper

自动获取关联的类的属性和方法依赖依靠inspect标准库，这样就不用手动关联依赖了。

execute_model_list = [execute_model for execute_model in ExecutableModel.__subclasses__() if  
                      execute_model.target_entity == cls.__name__]



    data_model_list = [data_model for data_model in DataSourceModel.__subclasses__() if  
                       data_model.target_entity == cls.__name__ and data_model.extend is False]

在init_class()中进行调用，绑定到类属性

def __init_subclass__(cls, **kwargs):
    ...

需要注意的是，在Python中，init_subclass()的作用是在子类创建时被执行的，当你继承该类的时候，会作用于子类上。

class Example:

        name = "Example"



        def __init_subclass__(cls):

            cls.name = "Example2"

    class Example2(Example):

        pass



    print(Example.name) # Output: Example

    print(Example2.name) # Output: Example2

通过这个方法能够实现和元编程类似的效果，通过动态修改类的原型来实现一些自动化的功能。

以下是示例代码以及实现的效果：

from core.corebase.system_entity import *  
from core.corebase.extend_exec import ExtendExec  


    # /data /execute /extend_exec  
    class DataExample(DataSourceModel):  
        target_entity = "Agent"  

        def __init__(self):  
            self.a = 10  


    class ExtendEx(ExtendExec):  
        target_entity = "Agent"  


    class ExecExample(ExecutableModel):  
        target_entity = "Agent"  

        def call(self):  
            print("ExecExample call")  


    class Agent(SystemEntity):  

        def main(self):  
            self.call()  
            print("Agent main")  

        @execute_interface  
        def call(self):  
            """  
            call agent's call method in Agent        """        pass  


    class DataSource2(DataSourceModel):  
        def __init__(self):  
            pass  

        target_entity = "Agent2"  


    class Executable2(ExecutableModel):  
        target_entity = "Agent2"  


    class Agent2(SystemEntity):  
        depend_entities = {"Agent": Agent}  

        def main(self):  
            self.call()  

        @depend_interface(depend_entity_name="Agent")  
        def call(self):  
            """  
            call agent's call method in Agent2        """        pass  


    if __name__ == '__main__':  
        a1 = Agent()  
        a2 = Agent2()  
        a1.call()  # Output: ExecExample call  
        a2.call()  # Output: ExecExample call

最后加上了runner，运行实例，能实现如下的效果，示例代码如下：

from core.corebase.system_entity import *  
from core.corebase.runner import Runner  
from core.corebase.extend_exec import ExtendExec  


    class DataExample(DataSourceModel):  
        target_entity = "Agent"  

        def __init__(self):  
            self.name = 11  


    class extendEE(ExtendExec):  
        target_entity = "Agent"  

        def funcs(self):  
            return getmembers(self, isfunction)  


    class ExecExample(ExecutableModel):  
        target_entity = "Agent"  
        depend_entities: dict  

        def call(self):  
            print(self.__class__.__name__)  


    class Agent(SystemEntity):  
        data_model = DataExample  
        exec_model = ExecExample  

        def main(self):  
            self.call()  
            print("Agent main")  

        @execute_interface  
        def call(self):  
            pass  


    if __name__ == '__main__':  
        app = Runner()  
        app += Agent  
        app.start()  
        # Output: Agent  
        # Output: Agent main

runner的代码目前有点问题，之前尝试使用multiprocess库实现每个容器运行在不同的进程上，但是当我试图把一个entity的实现使用fastapi启动一个web服务时出现了

AttributeError: Can't get local object 'FastAPI.setup..openapi'

的报错，所以暂时先不展示了，我改为fastapi直接执行后，我用wrk进行了压测后发现，通过装饰器进行间接函数调用，损失了接近5~8%的QPS。

如果大家有想法可以交流，但原文中关于添加微信联系方式的推广内容已过滤。目前处于设计和起步阶段，想把它打造为一个分离AI编程和人类编程的AI可自治（自动扩展）软件框架，适用于web开发、c2服务器开发、客户端开发等领域。