149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Any, Callable, List, Optional
|
|
|
|
from logging import getLogger
|
|
logger = getLogger(__name__)
|
|
|
|
class MiddlewareInterface(ABC):
|
|
"""中间件接口
|
|
"""
|
|
|
|
@abstractmethod
|
|
def process(self, data: Any) -> Any:
|
|
"""处理数据并返回处理后的结果"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_next(self) -> Optional[List['MiddlewareInterface']]:
|
|
"""获取下一个中间件"""
|
|
pass
|
|
|
|
class EngineInterface(ABC):
|
|
"""引擎接口
|
|
"""
|
|
@abstractmethod
|
|
def execute(self, data: Any) -> Any:
|
|
"""执行引擎逻辑并返回结果"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def transform(self, data: Any) -> Any:
|
|
"""将输出数据转换为标准化的格式"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def to_execute_middleware(self) -> MiddlewareInterface:
|
|
"""将引擎执行转换为遵循MiddlewareInterface的中间件."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def to_transform_middleware(self) -> MiddlewareInterface:
|
|
"""将引擎数据标准化转换为遵循MiddlewareInterface的中间件."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def to_middleware(self) -> MiddlewareInterface:
|
|
"""将引擎转换为遵循MiddlewareInterface的中间件."""
|
|
pass
|
|
|
|
class BaseMiddleware(MiddlewareInterface):
|
|
"""中间件基类
|
|
注意 next_middleware 参数是可选的,如果未提供,可能有两种模式
|
|
|
|
1.如果process返回值为None(只有next_middleware为None返回值才为None)
|
|
则返回数据将在Pipeline里返回,否则继续执行Pipeline的下一个中间件。
|
|
|
|
2.中间件的process返回值为process_logic处理后的值
|
|
"""
|
|
def __init__(self, process_logic: Callable[[Any], Any],
|
|
next_middleware: Optional[List[MiddlewareInterface]] = []):
|
|
self._process_logic = process_logic
|
|
self._next_middleware = next_middleware
|
|
self._data = None
|
|
|
|
def process(self, data: Any) -> Any:
|
|
"""处理数据并返回处理后的结果,支持自动调用下一个中间件
|
|
(如果next_middleware为None则返回None)"""
|
|
try:
|
|
data = self._process_logic(data)
|
|
self._data = data
|
|
if self._next_middleware:
|
|
for middleware in self._next_middleware:
|
|
self._data = middleware.process(self._data)
|
|
return None if self._next_middleware is None else data
|
|
except Exception as e:
|
|
logger.exception(f"Error occurred during processing: {e}")
|
|
raise e
|
|
|
|
def get_next(self) -> Optional[List[MiddlewareInterface]]:
|
|
return self._next_middleware
|
|
|
|
class BaseEngine(EngineInterface):
|
|
"""引擎基类
|
|
"""
|
|
def transform(self, data: Any) -> Any:
|
|
raise NotImplementedError("transform method not implemented")
|
|
|
|
def execute(self, data: Any) -> Any:
|
|
raise NotImplementedError("execute method not implemented")
|
|
|
|
def to_transform_middleware(self, next_middleware: Optional[List[MiddlewareInterface]] = []) -> MiddlewareInterface:
|
|
return BaseMiddleware(self.transform, next_middleware)
|
|
|
|
def to_execute_middleware(self, next_middleware: Optional[List[MiddlewareInterface]] = []) -> MiddlewareInterface:
|
|
return BaseMiddleware(self.execute, next_middleware)
|
|
|
|
def to_middleware(self, next_middleware: Optional[List[MiddlewareInterface]] = []) -> MiddlewareInterface:
|
|
return BaseMiddleware(lambda data: self.transform(self.execute(data)), next_middleware)
|
|
|
|
class ExecutePipeline:
|
|
"""
|
|
执行流 用于按顺序执行中间件
|
|
示例:
|
|
pipeline = ExecutePipeline(middleware1, middleware2)
|
|
pipeline.add(middleware3)
|
|
result = pipeline.execute(data)
|
|
|
|
执行顺序 data -> middleware1 -> middleware2 -> middleware3 -> result
|
|
|
|
pipeline.reorder([0,2,1])
|
|
result = pipeline.execute(data)
|
|
|
|
执行顺序 data -> middleware1 -> middleware3 -> middleware2 -> result
|
|
|
|
pipeline.reorder([0,1])
|
|
result = pipeline.execute(data)
|
|
|
|
执行顺序 data -> middleware1 -> middleware2 -> result
|
|
"""
|
|
def __init__(self, *middlewares:MiddlewareInterface):
|
|
self.middlewares = [*middlewares]
|
|
|
|
def add(self, middleware: MiddlewareInterface):
|
|
"""添加中间件到执行流里"""
|
|
self.middlewares.append(middleware)
|
|
|
|
def reorder(self, new_order: List[int]):
|
|
"""重新排序中间件,按照new_order指定的索引顺序排列,注意可以实现删除功能"""
|
|
self.middlewares = [self.middlewares[i] for i in new_order]
|
|
|
|
def execute(self, data):
|
|
"""执行中间件
|
|
|
|
注意中间件的next不参与主执行流
|
|
|
|
注意只有中间件的next为None,返回值才不会被输入到执行流中,即停止执行后续内容
|
|
|
|
TODO: 1. 添加中间件执行失败时的处理机制
|
|
"""
|
|
try:
|
|
for middleware in self.middlewares:
|
|
if data is None:
|
|
break
|
|
data = middleware.process(data)
|
|
return data
|
|
except ValueError:
|
|
return None
|
|
except Exception as e:
|
|
raise e
|