diff --git a/src/pypelines/__init__.py b/src/pypelines/__init__.py index 9a7e7e02027551a0f12c4aa9744bdfca03965688..688ab703060372be225bd300bc01f07dcdd36e0c 100644 --- a/src/pypelines/__init__.py +++ b/src/pypelines/__init__.py @@ -1,4 +1,4 @@ -__version__ = "0.0.42" +__version__ = "0.0.43" from . import loggs from .pipes import * diff --git a/src/pypelines/celery_tasks.py b/src/pypelines/celery_tasks.py index efb3a9559efa9ab6aa7cf7bd8b0c578d1704905b..1d0bacd688e3c5d113d2165504621bea5f9ad05a 100644 --- a/src/pypelines/celery_tasks.py +++ b/src/pypelines/celery_tasks.py @@ -379,6 +379,7 @@ def create_celery_app(conf_path, app_name="pypelines", v_host=None) -> "Celery | tasks_dynamic_data = {} pipelines = getattr(app, "pipelines", {}) for pipeline in pipelines.values(): + pipeline.resolve() for pipe in pipeline.pipes.values(): for step in pipe.steps.values(): if step.complete_name in app.tasks.keys(): @@ -390,10 +391,8 @@ def create_celery_app(conf_path, app_name="pypelines", v_host=None) -> "Celery | "step_name": step.step_name, "pipe_name": step.pipe_name, "pipeline_name": step.pipeline_name, - "requires": [ - item if isinstance(item, str) else item.complete_name for item in step.requires - ], - "step_level_in_pipe": step.get_level(selfish=True), + "requires": [item.complete_name for item in step.requires], + "step_level_in_pipe": step.get_level(selfish=False), } tasks_dynamic_data[step.complete_name] = task_data return tasks_dynamic_data