I am using the following code to chain multiple actions dynamically.
from abc import ABC, abstractmethod
from typing import Union, Any
class Pipeline(ABC):
def __init__(self, *actions: "Action"):
self.actions = actions
def __add__(self, other: "Pipeline") -> "Pipeline":
new_pipeline = Pipeline()
new_pipeline.actions = self.actions + other.actions
return new_pipeline
def __getitem__(self, index: Union[int, slice]) -> "Pipeline":
if isinstance(index, int):
return self.actions[index]
else:
new_pipeline = Pipeline()
new_pipeline.actions = self.actions[index]
return new_pipeline
def __len__(self) -> int:
return len(self.actions)
def run(self, pipeline_input: Any = None) -> Any:
value = pipeline_input
for action in self.actions:
value = action.run(value)
return value
class Action(Pipeline):
def __init__(self):
super().__init__(self) # note that it's "super().__init__(self)" instead of "super().__init__()"
@abstractmethod
def run(self, pipeline_input: Any = None) -> Any:
return
class Add(Action):
def __init__(self, value: int):
super().__init__()
self.value = value
def run(self, pipeline_input: int) -> int:
return pipeline_input + self.value
class Multiply(Action):
def __init__(self, value: int):
super().__init__()
self.value = value
def run(self, pipeline_input: int) -> int:
return pipeline_input * self.value
I can run the actions like the following
action1 = Add(3)
action2 = Multiply(4)
action1.run(2) # result 5
action2.run(5) # result 20
pipeline = action1 + action2
pipeline.run(5) # result 40
pipeline[0].run(3) # result 6
pipeline[0:2].run(5) # result 40
The problem is that this structure inherently requires the run method of Action's subclasses to allow only specific types as inputs, and therefore raises the following warning for run.
Method "run" overrides class "Action" in an incompatible manner
Parameter 2 mismatch: base parameter has default argument value, override parameter does not Pylancereport IncompatibleMethodOverride
How would i fix the type hint problem while also preserving the original spirit of the pipeline-action structure?
(The input and output type checking before chaining actions is done outside the python code, and the actions are chained dynamically, by name.)