Питоническое решение декорированного объекта конвейера ⇐ Python

[Гость]

Я создаю систему с несколькими конвейерами, имеющими примерно одинаковую структуру; экземпляр конвейера будет обрабатывать некоторые входящие данные с сохранением состояния и возвращать словарь, содержащий исходное значение, результат и все промежуточные значения.

Для моих различных целей было полезно создать декоратор для обозначения шагов конвейера. (На самом деле это фабрика декораторов, которая принимает различные параметры журналирования, но здесь это не имеет значения.) В качестве первой версии упрощенная версия выглядит так:

Class StepRegistry: защита __init__(сам): self._steps = [] шаг def (сам, отмеченная_функция): self._steps.append(marked_function) вернуть отмеченную_функцию защита __iter__(сам): выход из себя._шаги класс Трубопровод1: конвейер_шаги = StepRegistry() def __init__(self, config_data: dict = {}): для имени, значения в config_data.items(): setattr(я, имя, значение) def run(сам, текст): self.text = текст результат = {} результат["raw_text"] = self.text для step_function в self.pipeline_steps: результат[step_function.__name__] = Step_function(self) результат["final_text"] = self.text вернуть результат @pipeline_steps.step защита шаг_1 (сам): self.text = self.text[::-1] вернуть self.text @pipeline_steps.step защита шаг_2 (сам): self.text = self.text.replace("плохое слово", "эвфемизм") вернуть self.text класс Трубопровод2: конвейер_шаги = StepRegistry() def __init__(self, config_data: dict = {}): для имени, значения в config_data.items(): setattr(я, имя, значение) def run(сам, номер): self.number = номер результат = {} результат["raw_number"] = self.number для step_function в self.pipeline_steps: результат[step_function.__name__] = Step_function(self) результат["final_number"] = self.number вернуть результат @pipeline_steps.step защита шаг_1 (сам): собственный.номер = собственный.номер + 1 вернуть собственный номер @pipeline_steps.step защита шаг_2 (сам): собственный.номер = собственный.номер * 2 вернуть собственный номер если __name__ == "__main__": print("Выполняем тестовый запуск.") труба = Трубопровод1({"многословный": True}) result = Pipe.run("Здесь спрятан дроу!") печать (результат) труба = Pipeline2({"math_enabled": True}) результат = труба.run(42) печать (результат) Здесь повторяется инициализация конвейеров и использование шагов, поэтому я надеялся сделать что-то вроде этого:

импортировать abc класс StepRegistry: защита __init__(сам): self._steps = [] шаг def (сам, отмеченная_функция): self._steps.append(marked_function) вернуть отмеченную_функцию защита __iter__(сам): выход из себя._шаги класс AbstractPipeline: конвейер_шаги = StepRegistry() def __init__(self, config_data: dict = {}): для имени, значения в config_data.items(): setattr(я, имя, значение) def run(self, input_data): self.input_data = input_data self.result = {} self.pre_run() для step_function в self.pipeline_steps: self.result[step_function.__name__] = Step_function(self) self.post_run() вернуть self.result @abc.abstractmethod защита pre_run (сам): проходить @abc.abstractmethod защита post_run (сам): проходить класс Pipeline1 (Абстрактный конвейер): защита pre_run (сам): self.text = self.input_data self.result["raw_text"] = self.text защита post_run (сам): self.data = self.text self.result["final_text"] = self.text @pipeline_steps.step защита шаг_1 (сам): self.text = self.text[::-1] вернуть self.text @pipeline_steps.step защита шаг_2 (сам): self.text = self.text.replace("плохое слово", "эвфемизм") вернуть self.text класс Pipeline2 (AbstractPipeline): защита pre_run (сам): self.number = self.input_data self.result["raw_number"] = self.number защита post_run (сам): self.data = self.number self.result["final_number"] = self.number @pipeline_steps.step защита шаг_1 (сам): собственный.номер = собственный.номер + 1 вернуть собственный номер @pipeline_steps.step защита шаг_2 (сам): собственный.номер = собственный.номер * 2 вернуть собственный номер если __name__ == "__main__": print("Выполняем тестовый запуск.") труба = Трубопровод1({"многословный": True}) result = Pipe.run("Здесь спрятан дроу!") печать (результат) труба = Pipeline2({"math_enabled": True}) результат = труба.run(42) печать (результат) Это не сработает, потому что, например. при настройке класс Pipeline1 не знает, что такое pipeline_steps, поскольку он находится в суперклассе. Изменение @pipeline_steps.step на @AbstractPipeline.pipeline_steps.step вызывает другие проблемы, поскольку тогда конвейеры используют один и тот же экземпляр StepRegistry, где шаги будут содержать несовместимые шаги из обоих конвейеров.

Одно из предложений, которые я получил, заключалось в том, чтобы определить фактические шаги конвейера только внутри инициализации конвейера следующим образом:

импортировать abc класс StepRegistry: защита __init__(сам): self._steps = [] шаг def (сам, отмеченная_функция): self._steps.append(marked_function) вернуть отмеченную_функцию защита __iter__(сам): выход из себя._шаги класс AbstractPipeline: def __init__(self, config_data: dict = {}): self.pipeline_steps = StepRegistry() для имени, значения в config_data.items(): setattr(я, имя, значение) self.result = {} def run(self, input_data): self.input_data = input_data self.result = {} self.pre_run() для step_function в self.pipeline_steps: self.result[step_function.__name__] = Step_function(self) self.post_run() вернуть self.result @abc.abstractmethod защита pre_run (сам): проходить @abc.abstractmethod защита post_run (сам): проходить класс Pipeline1 (Абстрактный конвейер): def __init__(self, config_data: dict = {}): супер().__init__(config_data) @self.pipeline_steps.step защита шаг_1 (конвейер): конвейер.текст = конвейер.текст[::-1] вернуть конвейер.текст @self.pipeline_steps.step защита шаг_2 (конвейер): Pipeline.text = Pipeline.text.replace("плохое слово", "эвфемизм") вернуть конвейер.текст защита pre_run (сам): self.text = self.input_data self.result["raw_text"] = self.text защита post_run (сам): self.result["final_text"] = self.text класс Pipeline2 (AbstractPipeline): def __init__(self, config_data: dict = {}): супер().__init__(config_data) @self.pipeline_steps.step защита шаг_1 (конвейер): номер конвейера = номер конвейера + 1 возврат конвейера.номер @self.pipeline_steps.step защита шаг_2 (конвейер): номер конвейера = номер конвейера * 2 возврат конвейера.номер защита pre_run (сам): self.number = self.input_data self.result["raw_number"] = self.number защита post_run (сам): self.result["final_number"] = self.number если __name__ == "__main__": print("Выполняем тестовый запуск.") труба = Трубопровод1({"многословный": True}) result = Pipe.run("Здесь спрятан дроу!") печать (результат) труба = Pipeline2({"math_enabled": True}) результат = труба.run(42) печать (результат) Это действительно работает, но мне кажется странным. Более того, мне кажется, что я подхожу ко всему этому как-то не с той стороны.

Каким будет Pythonic-решение для этого?