实现一个python依赖注入demo，为了开发框架做准备
<p><strong>本文使用python实现一个依赖注入的框架的demo，实现了拆分数据实体和逻辑实体部分</strong></p><h3 id="一动机为什么要写一个python的依赖注入框架这不是多此一举吗">一、动机:：为什么要写一个python的依赖注入框架?这不是多此一举吗???</h3>
<p><strong>是,但也不完全是。</strong></p>
<p>例如,在fastapi中就可以使用依赖注入来完成功能<br>
这是一部分来自fastapi官方文档的代码</p>
<pre><code class="language-python">from typing import Union
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(
q: Union = 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
</code></pre>
<p>在这部分代码中通过Depends将一个异步函数注入到路径操作函数,当然这段代码肯定很多人有疑问———为什么不直接在代码内调用对应函数.</p>
<pre><code class="language-python">@app.get("/items/")
async def read_items():
return await common_parameters()
</code></pre>
<p>直接 await common_parameters() 不就完事了， 为什么要搞这么多令人疑惑的东西。<br>
如果仅仅是处理代码相关的依赖，用函数也可以实现依赖关系。<br>
但是，<br>
<strong>你实现函数调用的方法千千万，fastapi作为一个框架该如何定位依赖关系呢？</strong>，如果无法定位依赖关系，那么它就无法给你提供文档化以及各种缓存的服务。</p>
<h4 id="依赖注入框架的意义其实就是通过一种标准化的依赖方案理清组件与组件之间的关系从而为框架的建设铺路"><strong>依赖注入框架的意义其实就是通过一种标准化的依赖方案理清组件与组件之间的关系，从而为框架的建设铺路</strong></h4>
<pre><code>其实，有看过spring框架，dotnetCore框架的设计的，应该对此都不陌生。
</code></pre>
<p>话由说回来了，我要实现一个依赖注入框架的原因其实就是我想试着自己实现一个python编程框架。名字我都想好了🐶，暂时不说。</p>
<h3 id="二实现-实现一个依赖注入框架需要什么">二、实现： 实现一个依赖注入框架需要什么</h3>
<p>目前我主要使用装饰器，不过这带来了运行时的性能开销。之后可能会仅仅把它作为一个注解使用。主要的实现依赖于一个容器。容器在创建时会通过获取绑定的类的属性和方法来填充对象exec_impl，在调用到容器的原来的接口方法时会间接调用exec_impl内的方法。</p>
<p>这是我实现函数注入后调用的代码。</p>
<pre><code class="language-python">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(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.exec_impl(self, *args, **kwargs)
wrapper.__doc__ = func.__doc__
return wrapper
</code></pre>
<p>自动获取关联的类的属性和方法依赖 <strong>inspect</strong> 标准库，这样子就无需手动关联依赖。</p>
<pre><code class="language-python">execute_model_list = [execute_model for execute_model in ExecutableModel.__subclasses__() if
execute_model.target_entity == cls.__name__]
</code></pre>
<pre><code class="language-python">data_model_list = [data_model for data_model in DataSourceModel.__subclasses__() if
data_model.target_entity == cls.__name__ and data_model.extend is False]
</code></pre>
<p>在 <strong>init_class</strong>()中调用，绑定到类属性</p>
<pre><code class="language-python">def __init_subclass__(cls, **kwargs):
...
</code></pre>
<p>需要注意的是，python中， __init_subclass__() 的作用是在子类创建是被执行的，当你继承该类的时候，被作用于子类上。</p>
<pre><code class="language-python">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
</code></pre>
<p>通过这个方法可以实现和元编程类似的效果，通过动态修改类的原型来实现一些自动化的功能。</p>
<p>以下是示例代码以及实现的效果：</p>
<pre><code class="language-python">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
</code></pre>
<p>最后加上了runner，运行实例，能实现如下的效果，示例代码如下：</p>
<pre><code class="language-python">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
</code></pre>
<p>runner的代码目前有点问题，之前尝试使用<strong>multiprocess</strong>库实现每个容器运行在不同的进程上，但是当我试图把一个entity的实现使用fastapi启动一个web服务时出现了</p>
<p>AttributeError: Can't get local object 'FastAPI.setup.<locals>.openapi'</locals></p>
<p>的报错,所以暂时先不展示了，我改为fastapi直接执行后，我用wrk进行了压测后发现，通过装饰器进行间接函数调用，损失了接近5~8%的QPS。</p>
<hr>
<p>如果大家有什么想法可以添加<strong>我的微信：lry13890140665</strong> ， 我想把这个写成一个分离AI编程和人类编程的AI可以自治（自动扩展）软件框架，能够适用于web开发，c2服务器开发，客户端开发等领域。<br>
目前处于设计和起步阶段，，，，</p><br><br>
来源：https://www.cnblogs.com/whereslow/p/18951004
頁:
[1]