实现一个python依赖注入demo，为了开发框架做准备

本文利用python实现一个依赖注入的框架的demo，实现了拆分数据实体和逻辑实体部分
一、动机:：为什么要写一个python的依赖注入框架?这不是多此一举吗???

是,但也不完满是。
例如,在fastapi中就可以利用依赖注入来完乐成能
这是一部分来自fastapi官方文档的代码

1. from typing import Union
3. from fastapi import Depends, FastAPI
5. app = FastAPI()
7. async def common_parameters(
8.     q: Union[str, None] = None, skip: int = 0, limit: int = 100
9. ):
10.     return {"q": q, "skip": skip, "limit": limit}
12. @app.get("/items/")
13. async def read_items(commons: dict = Depends(common_parameters)):
14.     return commons
16. @app.get("/users/")
17. async def read_users(commons: dict = Depends(common_parameters)):
18.     return commons

复制代码

在这部分代码中通过Depends将一个异步函数注入到路径操纵函数,当然这段代码肯定许多人有疑问———为什么不直接在代码内调用对应函数.

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

复制代码

直接 await common_parameters() 不就完事了， 为什么要搞这么多令人疑惑的东西。
如果仅仅是处理代码相关的依赖，用函数也可以实现依赖关系。
但是，
你实现函数调用的方法千千万，fastapi作为一个框架该如何定位依赖关系呢？，如果无法定位依赖关系，那么它就无法给你提供文档化以及各种缓存的服务。
依赖注入框架的意义其实就是通过一种尺度化的依赖方案理清组件与组件之间的关系，从而为框架的建设铺路

1. 其实，有看过spring框架，dotnetCore框架的设计的，应该对此都不陌生。

复制代码

话由说回来了，我要实现一个依赖注入框架的缘故原由其实就是我想试着自己实现一个python编程框架。名字我都想好了🐶，暂时不说。
二、实现： 实现一个依赖注入框架需要什么

目前我主要利用装饰器，不过这带来了运行时的性能开销。之后可能会仅仅把它作为一个注解利用。主要的实现依赖于一个容器。容器在创建时会通过获取绑定的类的属性和方法来填充对象exec_impl，在调用到容器的原来的接口方法时会间接调用exec_impl内的方法。
这是我实现函数注入后调用的代码。

1. def execute_interface(func: Callable):  
2.     name = func.__name__  
3.   
4.     def wrapper(self, *args, **kwargs):  
5.         if not isinstance(self, SystemEntity):  
6.             raise NotImplementedError(f"please use class derived from SystemEntity")  
7.         return self.exec_impl[name](self, *args, **kwargs)  
8.   
9.     wrapper.__doc__ = func.__doc__  
10.     return wrapper  
11.   
12.   
13. def depend_interface(depend_entity_name: str):  
14.     def decorator(func: Callable):  
15.         func_name = func.__name__  
16.   
17.         def wrapper(self, *args, **kwargs):  
18.             if not isinstance(self, SystemEntity):  
19.                 raise NotImplementedError(f"please use class derived from SystemEntity")  
20.             if depend_entity_name not in self.depend_entities.keys():  
21.                 raise NotImplementedError(  
22.                     f"{func_name} depend on {depend_entity_name}, but {depend_entity_name} not found")  
23.             return self.depend_entities[depend_entity_name].exec_impl[func_name](self, *args, **kwargs)  
24.   
25.         wrapper.__doc__ = func.__doc__  
26.         return wrapper

复制代码

自动获取关联的类的属性和方法依赖 inspect 尺度库，这样子就无需手动关联依赖。

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

复制代码

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

复制代码

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

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

复制代码

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

1. class Example:
3.     name = "Example"
5.   
7.     def __init_subclass__(cls):
9.         cls.name = "Example2"
11. class Example2(Example):
13.     pass
15.   
17. print(Example.name) # Output: Example
19. print(Example2.name) # Output: Example2

复制代码

通过这个方法可以实现和元编程类似的效果，通过动态修改类的原型来实现一些自动化的功能。
以下是示例代码以及实现的效果：

1. from core.corebase.system_entity import *  
2. from core.corebase.extend_exec import ExtendExec  
3.   
4.   
5. # /data /execute /extend_exec  
6. class DataExample(DataSourceModel):  
7.     target_entity = "Agent"  
8.   
9.     def __init__(self):  
10.         self.a = 10  
11.   
12.   
13. class ExtendEx(ExtendExec):  
14.     target_entity = "Agent"  
15.   
16.   
17. class ExecExample(ExecutableModel):  
18.     target_entity = "Agent"  
19.   
20.     def call(self):  
21.         print("ExecExample call")  
22.   
23.   
24. class Agent(SystemEntity):  
25.   
26.     def main(self):  
27.         self.call()  
28.         print("Agent main")  
29.   
30.     @execute_interface  
31.     def call(self):  
32.         """  
33.         call agent's call method in Agent        """        pass  
34.   
35.   
36. class DataSource2(DataSourceModel):  
37.     def __init__(self):  
38.         pass  
39.   
40.     target_entity = "Agent2"  
41.   
42.   
43. class Executable2(ExecutableModel):  
44.     target_entity = "Agent2"  
45.   
46.   
47. class Agent2(SystemEntity):  
48.     depend_entities = {"Agent": Agent}  
49.   
50.     def main(self):  
51.         self.call()  
52.   
53.     @depend_interface(depend_entity_name="Agent")  
54.     def call(self):  
55.         """  
56.         call agent's call method in Agent2        """        pass  
57.   
58.   
59. if __name__ == '__main__':  
60.     a1 = Agent()  
61.     a2 = Agent2()  
62.     a1.call()  # Output: ExecExample call  
63.     a2.call()  # Output: ExecExample call

复制代码

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

1. from core.corebase.system_entity import *  
2. from core.corebase.runner import Runner  
3. from core.corebase.extend_exec import ExtendExec  
4.   
5.   
6. class DataExample(DataSourceModel):  
7.     target_entity = "Agent"  
8.   
9.     def __init__(self):  
10.         self.name = 11  
11.   
12.   
13. class extendEE(ExtendExec):  
14.     target_entity = "Agent"  
15.   
16.     def funcs(self):  
17.         return getmembers(self, isfunction)  
18.   
19.   
20. class ExecExample(ExecutableModel):  
21.     target_entity = "Agent"  
22.     depend_entities: dict  
23.   
24.     def call(self):  
25.         print(self.__class__.__name__)  
26.   
27.   
28. class Agent(SystemEntity):  
29.     data_model = DataExample  
30.     exec_model = ExecExample  
31.   
32.     def main(self):  
33.         self.call()  
34.         print("Agent main")  
35.   
36.     @execute_interface  
37.     def call(self):  
38.         pass  
39.   
40.   
41. if __name__ == '__main__':  
42.     app = Runner()  
43.     app += Agent  
44.     app.start()  
45.     # Output: Agent  
46.     # Output: Agent main

复制代码

runner的代码目前有点问题，之前尝试利用multiprocess库实现每个容器运行在不同的进程上，但是当我试图把一个entity的实现利用fastapi启动一个web服务时出现了
AttributeError: Can't get local object 'FastAPI.setup..openapi'
的报错,所以暂时先不展示了，我改为fastapi直接实行后，我用wrk举行了压测后发现，通过装饰器举行间接函数调用，损失了接近5~8%的QPS。
如果大家有什么想法可以添加我的微信：lry13890140665 ， 我想把这个写成一个分离AI编程和人类编程的AI可以自治（自动扩展）软件框架，能够适用于web开发，c2服务器开发，客户端开发等领域。
目前处于设计和起步阶段，，，，