From 3556909c23e6467461122cbd252b4a21bdbd2e48 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Fri, 10 Apr 2026 13:36:41 +0200 Subject: [PATCH 01/11] Add support for adding unique constrains on TaskParameters, Refactor src folder structure --- examples/lot_sizing/lot_sizing_pipeline.py | 2 +- .../ml_blood_sugar_level.py | 32 ++------- src/cosy_luigi/__init__.py | 3 + src/cosy_luigi/constraints/__init__.py | 3 + src/cosy_luigi/constraints/unique.py | 28 ++++++++ src/cosy_luigi/core/__init__.py | 3 + src/cosy_luigi/{ => core}/combinatorics.py | 67 +++++++++++++------ src/cosy_luigi/utils/__init__.py | 3 + src/cosy_luigi/utils/traversals.py | 39 +++++++++++ tests/test_abstract_variant_expansion.py | 2 +- tests/test_heterogeneous_repo_input.py | 5 +- tests/test_infinite_chain.py | 2 +- 12 files changed, 139 insertions(+), 50 deletions(-) create mode 100644 src/cosy_luigi/constraints/__init__.py create mode 100644 src/cosy_luigi/constraints/unique.py create mode 100644 src/cosy_luigi/core/__init__.py rename src/cosy_luigi/{ => core}/combinatorics.py (58%) create mode 100644 src/cosy_luigi/utils/__init__.py create mode 100644 src/cosy_luigi/utils/traversals.py diff --git a/examples/lot_sizing/lot_sizing_pipeline.py b/examples/lot_sizing/lot_sizing_pipeline.py index da65380..a775326 100644 --- a/examples/lot_sizing/lot_sizing_pipeline.py +++ b/examples/lot_sizing/lot_sizing_pipeline.py @@ -13,7 +13,7 @@ from lot_optimizers.silver_meal_heuristic import SilverMeal from lot_optimizers.wagner_whitin import WagnerWhitin -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class GetCosts(CoSyLuigiTask): diff --git a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py index 0042995..bf47ca2 100644 --- a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py +++ b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py @@ -1,7 +1,6 @@ import json import textwrap from abc import ABC -from collections.abc import Callable, Iterable, Mapping, Sequence from pathlib import Path import luigi @@ -15,7 +14,7 @@ from sklearn.model_selection import train_test_split from sklearn.preprocessing import MinMaxScaler, RobustScaler -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter ninetydegaisle = True @@ -146,7 +145,7 @@ def run(self): class EvaluateRegressionModel(CoSyLuigiTask): regressor = CoSyLuigiTaskParameter(TrainRegressionModel) - scaled_feats = CoSyLuigiTaskParameter(FitTransformScaler) + scaled_feats = CoSyLuigiTaskParameter(FitTransformMinMaxScaler, unique_across_prior_tasks=True) splitted_data = CoSyLuigiTaskParameter(TrainTestSplit) def _get_variant_label(self): @@ -170,30 +169,9 @@ def run(self): y_pred.to_json(self.output().path) - @classmethod - def constraints(cls) -> Sequence[Callable[..., bool]]: - return [lambda vs: check_unique(vs, [FitTransformScaler]), lambda _: ninetydegaisle] - - -def traverse_pipeline(vs: Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: - result = [*vs] - for v in result: - result.extend(traverse_pipeline(v.requires().values())) - return result - - -def check_unique(vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Iterable[type[CoSyLuigiTask]]) -> bool: - classes = [pc.__class__ for pc in traverse_pipeline(vs.values())] - seen_subclasses = {} - for c in classes: - for unique in required_to_be_unique: - if issubclass(c, unique): - if unique in seen_subclasses: - if seen_subclasses[unique] != c: - return False - else: - seen_subclasses[unique] = c - return True + + + def main(): diff --git a/src/cosy_luigi/__init__.py b/src/cosy_luigi/__init__.py index e69de29..ac789a5 100644 --- a/src/cosy_luigi/__init__.py +++ b/src/cosy_luigi/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.core import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter + +__all__ = ["CoSyLuigiRepo", "CoSyLuigiTask", "CoSyLuigiTaskParameter"] diff --git a/src/cosy_luigi/constraints/__init__.py b/src/cosy_luigi/constraints/__init__.py new file mode 100644 index 0000000..8c8ee8b --- /dev/null +++ b/src/cosy_luigi/constraints/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.constraints.unique import is_unique_in_prior_tasks + +__all__ = ["is_unique_in_prior_tasks"] diff --git a/src/cosy_luigi/constraints/unique.py b/src/cosy_luigi/constraints/unique.py new file mode 100644 index 0000000..8bce9b7 --- /dev/null +++ b/src/cosy_luigi/constraints/unique.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from cosy_luigi import CoSyLuigiTask +from cosy_luigi.utils import traverse_pipeline + +if TYPE_CHECKING: + from collections.abc import Iterable, Mapping + + +def _is_unique_in_prior_tasks(vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Iterable[type[CoSyLuigiTask]]) -> bool: + classes = [pc.__class__ for pc in traverse_pipeline(vs.values())] + seen_subclasses = {} + for c in classes: + for unique in required_to_be_unique: + if issubclass(c, unique): + if unique in seen_subclasses: + if seen_subclasses[unique] != c: + return False + else: + seen_subclasses[unique] = c + return True + + +def is_unique_in_prior_tasks(vs: Mapping[str, CoSyLuigiTask], + required_to_be_unique: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]) -> bool: + return _is_unique_in_prior_tasks(vs, [required_to_be_unique] if isinstance(required_to_be_unique, CoSyLuigiTask) else required_to_be_unique) diff --git a/src/cosy_luigi/core/__init__.py b/src/cosy_luigi/core/__init__.py new file mode 100644 index 0000000..160decc --- /dev/null +++ b/src/cosy_luigi/core/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.core.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter + +__all__ = ["CoSyLuigiRepo", "CoSyLuigiTask", "CoSyLuigiTaskParameter"] diff --git a/src/cosy_luigi/combinatorics.py b/src/cosy_luigi/core/combinatorics.py similarity index 58% rename from src/cosy_luigi/combinatorics.py rename to src/cosy_luigi/core/combinatorics.py index 56899b1..9949711 100644 --- a/src/cosy_luigi/combinatorics.py +++ b/src/cosy_luigi/core/combinatorics.py @@ -1,10 +1,9 @@ from __future__ import annotations -import inspect -from abc import ABC +import textwrap from collections import defaultdict -from functools import cache -from typing import TYPE_CHECKING, cast +from functools import cache, partial +from typing import TYPE_CHECKING import luigi from cosy.core import Constructor, SpecificationBuilder @@ -17,9 +16,10 @@ class CoSyLuigiTaskParameter(luigi.TaskParameter): - def __init__(self, required_task: type[CoSyLuigiTask]): + def __init__(self, required_task: type[CoSyLuigiTask], *, unique_across_prior_tasks: bool = False): super().__init__() self.required_task = required_task + self.unique_across_prior_tasks = unique_across_prior_tasks class CoSyLuigiTask(luigi.Task): @@ -60,13 +60,23 @@ def requires(self): @classmethod @cache - def _requirements(cls): + def _requirements(cls) -> Mapping[str, CoSyLuigiTaskParameter]: return { k: v for k, v in cls.get_all_class_attributes().items() if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTaskParameter) } + @classmethod + @cache + def requirements_unique_in_prior_tasks(cls) -> Mapping[str, CoSyLuigiTaskParameter]: + return {k: task_parameter for k, task_parameter in cls._requirements().items() if task_parameter.unique_across_prior_tasks} + + @classmethod + @cache + def unique_required_tasks_in_prior(cls) -> Iterable[type[CoSyLuigiTask]]: + return [task_parameter.required_task for task_parameter in cls.requirements_unique_in_prior_tasks().values()] + @classmethod @cache def target(cls): @@ -76,11 +86,18 @@ def target(cls): def constraints(cls) -> Sequence[Callable[..., bool]]: return [] + @classmethod + def __constraints(cls) -> Sequence[Callable[..., bool]]: + from cosy_luigi.constraints.unique import _is_unique_in_prior_tasks # noqa: PLC0415 + return [partial(_is_unique_in_prior_tasks, required_to_be_unique=cls.unique_required_tasks_in_prior())] + @classmethod def combinator_type(cls): sp = SpecificationBuilder() for name in [v.required_task.__name__ for v in cls._requirements().values()]: sp = sp.argument(name, Constructor(name)) + for constraint in cls.__constraints(): + sp = sp.constraint(constraint) for constraint in cls.constraints(): sp = sp.constraint(constraint) return sp.suffix(cls.target()) @@ -97,23 +114,13 @@ def __init__(self, *tasks: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): Register.disable_instance_cache() # Accepts completely heterogeneous nested collections - def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): - return ( - task - for task_or_task_collection in heterogeneous_task_collection - for task in ( - flatten(*cast("Iterable[type[CoSyLuigiTask]]", task_or_task_collection)) - if isinstance(task_or_task_collection, (tuple, list)) - else cast("type[CoSyLuigiTask]", task_or_task_collection).get_all_variants() - if inspect.isabstract(task_or_task_collection) - or ABC in cast("type[CoSyLuigiTask]", task_or_task_collection).__bases__ - else (task_or_task_collection,) - ) - ) + # This doesn't technically need to unpack as flatten could be typed to accept packed tuples # But performance is equivalent/faster because the first layer doesn't need to be checked this way + from cosy_luigi.utils import flatten # noqa: PLC0415 self.luigi_repo: set[type[CoSyLuigiTask]] = set(flatten(*tasks)) + self.check_unique_in_prior_tasks_sanity() self.taxonomy: Mapping[str, set[str]] = defaultdict(set) self.cls_repo: set[tuple[str, Callable, Specification]] = set() for task in self.luigi_repo: @@ -122,3 +129,25 @@ def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Iterable[type[ if issubclass(tpe, CoSyLuigiTask): # Is a subclass of CosyLuigiTask, but a superclass of task self.taxonomy[task.__name__].add(tpe.__name__) + + def check_unique_in_prior_tasks_sanity(self): + for source_task, param_name, required_type in [(task, k,required_unique_task.required_task) for task in self.luigi_repo for k, required_unique_task in task.requirements_unique_in_prior_tasks().items() if not any(issubclass(task, required_unique_task.required_task) and task is not required_unique_task.required_task for task in self.luigi_repo)]: + print( + textwrap.dedent( + f""" + ================================================================= + WARNING + + Class: {source_task.__name__} + Parameter: {param_name} + Type: {required_type.__name__} + + is required to be unique, but there are no sub- + classes of it present in the repository. Either + you forgot adding sub-classes of it to the rep- + ository or this will not behave as expected. + + Please head over to the documentation: + =================================================================""" + ) + ) diff --git a/src/cosy_luigi/utils/__init__.py b/src/cosy_luigi/utils/__init__.py new file mode 100644 index 0000000..ebd4586 --- /dev/null +++ b/src/cosy_luigi/utils/__init__.py @@ -0,0 +1,3 @@ +from cosy_luigi.utils.traversals import flatten, traverse_pipeline + +__all__ = ["flatten", "traverse_pipeline"] diff --git a/src/cosy_luigi/utils/traversals.py b/src/cosy_luigi/utils/traversals.py new file mode 100644 index 0000000..89e84e8 --- /dev/null +++ b/src/cosy_luigi/utils/traversals.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import inspect +from abc import ABC +from typing import TYPE_CHECKING, cast + +from cosy_luigi import CoSyLuigiTask + +if TYPE_CHECKING: + from collections.abc import Iterable + + +def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): + return ( + task + for task_or_task_collection in heterogeneous_task_collection + for task in ( + flatten(*cast("Iterable[type[CoSyLuigiTask]]", task_or_task_collection)) + if isinstance(task_or_task_collection, (tuple, list)) + else cast("type[CoSyLuigiTask]", task_or_task_collection).get_all_variants() + if inspect.isabstract(task_or_task_collection) + or ABC in cast("type[CoSyLuigiTask]", task_or_task_collection).__bases__ + else (task_or_task_collection,) + ) + ) + +def _traverse_pipeline(vs: Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: + result = [*vs] + for v in vs: + result.extend(traverse_pipeline(v.requires().values())) + return result + + +def traverse_pipeline(to_traverse: CoSyLuigiTask | Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: + return ( + _traverse_pipeline([to_traverse]) + if isinstance(to_traverse, CoSyLuigiTask) + else _traverse_pipeline(to_traverse) + ) diff --git a/tests/test_abstract_variant_expansion.py b/tests/test_abstract_variant_expansion.py index 6bbcfd1..e9f53af 100644 --- a/tests/test_abstract_variant_expansion.py +++ b/tests/test_abstract_variant_expansion.py @@ -1,6 +1,6 @@ from abc import ABC, abstractmethod -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask class ABCInheritedTask(CoSyLuigiTask, ABC): diff --git a/tests/test_heterogeneous_repo_input.py b/tests/test_heterogeneous_repo_input.py index 204b292..f3eea2c 100644 --- a/tests/test_heterogeneous_repo_input.py +++ b/tests/test_heterogeneous_repo_input.py @@ -1,4 +1,5 @@ -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask +from cosy_luigi.utils import flatten class TaskA(CoSyLuigiTask): @@ -48,3 +49,5 @@ class TaskK(CoSyLuigiTask): def test_heterogeneous_repo_input(): repo = CoSyLuigiRepo(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) assert repo.luigi_repo == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} + flattened_collection = set(flatten(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK]))) + assert flattened_collection == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} diff --git a/tests/test_infinite_chain.py b/tests/test_infinite_chain.py index 1d85513..801950a 100644 --- a/tests/test_infinite_chain.py +++ b/tests/test_infinite_chain.py @@ -4,7 +4,7 @@ from cosy.maestro import Maestro from luigi.mock import MockTarget -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class ChainLink(CoSyLuigiTask, ABC): From bc375ed8a6b09776323e865945f264526917f873 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Fri, 10 Apr 2026 13:45:53 +0200 Subject: [PATCH 02/11] Run formatting pass --- examples/getting_started/basic_example.py | 2 +- examples/getting_started/variation_example.py | 2 +- .../ml_blood_sugar_level.py | 4 -- src/cosy_luigi/constraints/unique.py | 13 ++++-- src/cosy_luigi/core/combinatorics.py | 41 +++++++++++++------ src/cosy_luigi/utils/traversals.py | 19 ++++----- tests/test_heterogeneous_repo_input.py | 4 +- 7 files changed, 52 insertions(+), 33 deletions(-) diff --git a/examples/getting_started/basic_example.py b/examples/getting_started/basic_example.py index 59e9e28..a079bc6 100644 --- a/examples/getting_started/basic_example.py +++ b/examples/getting_started/basic_example.py @@ -3,7 +3,7 @@ import luigi from cosy.maestro import Maestro -from src.cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class TaskA(CoSyLuigiTask): diff --git a/examples/getting_started/variation_example.py b/examples/getting_started/variation_example.py index 5b5c3e1..49ab853 100644 --- a/examples/getting_started/variation_example.py +++ b/examples/getting_started/variation_example.py @@ -5,7 +5,7 @@ import luigi from cosy.maestro import Maestro -from src.cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class WriteTemplateTask(CoSyLuigiTask): diff --git a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py index bf47ca2..d76385d 100644 --- a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py +++ b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py @@ -170,10 +170,6 @@ def run(self): y_pred.to_json(self.output().path) - - - - def main(): repo = CoSyLuigiRepo( TrainTestSplit, diff --git a/src/cosy_luigi/constraints/unique.py b/src/cosy_luigi/constraints/unique.py index 8bce9b7..1201a72 100644 --- a/src/cosy_luigi/constraints/unique.py +++ b/src/cosy_luigi/constraints/unique.py @@ -9,7 +9,9 @@ from collections.abc import Iterable, Mapping -def _is_unique_in_prior_tasks(vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Iterable[type[CoSyLuigiTask]]) -> bool: +def _is_unique_in_prior_tasks( + vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Iterable[type[CoSyLuigiTask]] +) -> bool: classes = [pc.__class__ for pc in traverse_pipeline(vs.values())] seen_subclasses = {} for c in classes: @@ -23,6 +25,9 @@ def _is_unique_in_prior_tasks(vs: Mapping[str, CoSyLuigiTask], required_to_be_un return True -def is_unique_in_prior_tasks(vs: Mapping[str, CoSyLuigiTask], - required_to_be_unique: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]) -> bool: - return _is_unique_in_prior_tasks(vs, [required_to_be_unique] if isinstance(required_to_be_unique, CoSyLuigiTask) else required_to_be_unique) +def is_unique_in_prior_tasks( + vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]] +) -> bool: + return _is_unique_in_prior_tasks( + vs, [required_to_be_unique] if isinstance(required_to_be_unique, CoSyLuigiTask) else required_to_be_unique + ) diff --git a/src/cosy_luigi/core/combinatorics.py b/src/cosy_luigi/core/combinatorics.py index 9949711..7530d7d 100644 --- a/src/cosy_luigi/core/combinatorics.py +++ b/src/cosy_luigi/core/combinatorics.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import textwrap from collections import defaultdict from functools import cache, partial @@ -14,6 +15,8 @@ from cosy.core.synthesizer import Specification +logger = logging.getLogger(__name__) + class CoSyLuigiTaskParameter(luigi.TaskParameter): def __init__(self, required_task: type[CoSyLuigiTask], *, unique_across_prior_tasks: bool = False): @@ -70,7 +73,11 @@ def _requirements(cls) -> Mapping[str, CoSyLuigiTaskParameter]: @classmethod @cache def requirements_unique_in_prior_tasks(cls) -> Mapping[str, CoSyLuigiTaskParameter]: - return {k: task_parameter for k, task_parameter in cls._requirements().items() if task_parameter.unique_across_prior_tasks} + return { + k: task_parameter + for k, task_parameter in cls._requirements().items() + if task_parameter.unique_across_prior_tasks + } @classmethod @cache @@ -89,6 +96,7 @@ def constraints(cls) -> Sequence[Callable[..., bool]]: @classmethod def __constraints(cls) -> Sequence[Callable[..., bool]]: from cosy_luigi.constraints.unique import _is_unique_in_prior_tasks # noqa: PLC0415 + return [partial(_is_unique_in_prior_tasks, required_to_be_unique=cls.unique_required_tasks_in_prior())] @classmethod @@ -115,10 +123,10 @@ def __init__(self, *tasks: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): # Accepts completely heterogeneous nested collections - # This doesn't technically need to unpack as flatten could be typed to accept packed tuples # But performance is equivalent/faster because the first layer doesn't need to be checked this way from cosy_luigi.utils import flatten # noqa: PLC0415 + self.luigi_repo: set[type[CoSyLuigiTask]] = set(flatten(*tasks)) self.check_unique_in_prior_tasks_sanity() self.taxonomy: Mapping[str, set[str]] = defaultdict(set) @@ -131,23 +139,32 @@ def __init__(self, *tasks: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): self.taxonomy[task.__name__].add(tpe.__name__) def check_unique_in_prior_tasks_sanity(self): - for source_task, param_name, required_type in [(task, k,required_unique_task.required_task) for task in self.luigi_repo for k, required_unique_task in task.requirements_unique_in_prior_tasks().items() if not any(issubclass(task, required_unique_task.required_task) and task is not required_unique_task.required_task for task in self.luigi_repo)]: - print( + for source_task, param_name, required_type in [ + (task, k, required_unique_task.required_task) + for task in self.luigi_repo + for k, required_unique_task in task.requirements_unique_in_prior_tasks().items() + if not any( + issubclass(task, required_unique_task.required_task) and task is not required_unique_task.required_task + for task in self.luigi_repo + ) + ]: + logger.warning( textwrap.dedent( f""" ================================================================= - WARNING - + WARNING ABOUT POTENTIALLY INCORRECT MODEL + Class: {source_task.__name__} - Parameter: {param_name} + Parameter: {param_name} Type: {required_type.__name__} - + is required to be unique, but there are no sub- classes of it present in the repository. Either you forgot adding sub-classes of it to the rep- - ository or this will not behave as expected. - - Please head over to the documentation: - =================================================================""" + ository or this will not behave as expected. + + Please head over to the documentation: + ================================================================= + """ ) ) diff --git a/src/cosy_luigi/utils/traversals.py b/src/cosy_luigi/utils/traversals.py index 89e84e8..fd2f99e 100644 --- a/src/cosy_luigi/utils/traversals.py +++ b/src/cosy_luigi/utils/traversals.py @@ -15,15 +15,16 @@ def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Iterable[type[ task for task_or_task_collection in heterogeneous_task_collection for task in ( - flatten(*cast("Iterable[type[CoSyLuigiTask]]", task_or_task_collection)) - if isinstance(task_or_task_collection, (tuple, list)) - else cast("type[CoSyLuigiTask]", task_or_task_collection).get_all_variants() - if inspect.isabstract(task_or_task_collection) - or ABC in cast("type[CoSyLuigiTask]", task_or_task_collection).__bases__ - else (task_or_task_collection,) - ) + flatten(*cast("Iterable[type[CoSyLuigiTask]]", task_or_task_collection)) + if isinstance(task_or_task_collection, (tuple, list)) + else cast("type[CoSyLuigiTask]", task_or_task_collection).get_all_variants() + if inspect.isabstract(task_or_task_collection) + or ABC in cast("type[CoSyLuigiTask]", task_or_task_collection).__bases__ + else (task_or_task_collection,) + ) ) + def _traverse_pipeline(vs: Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: result = [*vs] for v in vs: @@ -33,7 +34,5 @@ def _traverse_pipeline(vs: Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: def traverse_pipeline(to_traverse: CoSyLuigiTask | Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: return ( - _traverse_pipeline([to_traverse]) - if isinstance(to_traverse, CoSyLuigiTask) - else _traverse_pipeline(to_traverse) + _traverse_pipeline([to_traverse]) if isinstance(to_traverse, CoSyLuigiTask) else _traverse_pipeline(to_traverse) ) diff --git a/tests/test_heterogeneous_repo_input.py b/tests/test_heterogeneous_repo_input.py index f3eea2c..a61ee36 100644 --- a/tests/test_heterogeneous_repo_input.py +++ b/tests/test_heterogeneous_repo_input.py @@ -49,5 +49,7 @@ class TaskK(CoSyLuigiTask): def test_heterogeneous_repo_input(): repo = CoSyLuigiRepo(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) assert repo.luigi_repo == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} - flattened_collection = set(flatten(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK]))) + flattened_collection = set( + flatten(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) + ) assert flattened_collection == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} From d40afdc6dfbe2a5a1903e653f4bd3aa33df821f8 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Fri, 10 Apr 2026 13:59:59 +0200 Subject: [PATCH 03/11] Fix type-checking --- src/cosy_luigi/constraints/unique.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/cosy_luigi/constraints/unique.py b/src/cosy_luigi/constraints/unique.py index 1201a72..b4d9d3e 100644 --- a/src/cosy_luigi/constraints/unique.py +++ b/src/cosy_luigi/constraints/unique.py @@ -1,19 +1,21 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from collections.abc import Iterable +from typing import TYPE_CHECKING, cast -from cosy_luigi import CoSyLuigiTask from cosy_luigi.utils import traverse_pipeline if TYPE_CHECKING: - from collections.abc import Iterable, Mapping + from collections.abc import Mapping + + from cosy_luigi import CoSyLuigiTask def _is_unique_in_prior_tasks( vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Iterable[type[CoSyLuigiTask]] ) -> bool: classes = [pc.__class__ for pc in traverse_pipeline(vs.values())] - seen_subclasses = {} + seen_subclasses: dict[type[CoSyLuigiTask], type[CoSyLuigiTask]] = {} for c in classes: for unique in required_to_be_unique: if issubclass(c, unique): @@ -29,5 +31,8 @@ def is_unique_in_prior_tasks( vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]] ) -> bool: return _is_unique_in_prior_tasks( - vs, [required_to_be_unique] if isinstance(required_to_be_unique, CoSyLuigiTask) else required_to_be_unique + vs, + required_to_be_unique + if isinstance(required_to_be_unique, Iterable) + else [required_to_be_unique], ) From 9ce8c3d2e8253959499da2492586dd8dd456983c Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 13:56:48 +0200 Subject: [PATCH 04/11] Add tests for new feature --- pyproject.toml | 7 +- src/cosy_luigi/core/combinatorics.py | 4 +- tests/test_abstract_variant_expansion.py | 33 ++++++- tests/test_heterogeneous_repo_input.py | 14 +++ tests/test_infinite_chain.py | 12 --- tests/test_unique_task_parameter.py | 119 +++++++++++++++++++++++ 6 files changed, 171 insertions(+), 18 deletions(-) create mode 100644 tests/test_unique_task_parameter.py diff --git a/pyproject.toml b/pyproject.toml index a18c127..90ff5a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -76,9 +76,12 @@ exclude_lines = [ "no cov", "if __name__ == .__main__.:", "if TYPE_CHECKING:", - "@abstractmethod", - "@abc.abstractmethod" ] +exclude_also = [ + "raise AssertionError", + "raise NotImplementedError", + "@(abc\\.)?abstractmethod", + ] [tool.ruff.lint.extend-per-file-ignores] diff --git a/src/cosy_luigi/core/combinatorics.py b/src/cosy_luigi/core/combinatorics.py index 7530d7d..06d9f2d 100644 --- a/src/cosy_luigi/core/combinatorics.py +++ b/src/cosy_luigi/core/combinatorics.py @@ -130,9 +130,9 @@ def __init__(self, *tasks: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): self.luigi_repo: set[type[CoSyLuigiTask]] = set(flatten(*tasks)) self.check_unique_in_prior_tasks_sanity() self.taxonomy: Mapping[str, set[str]] = defaultdict(set) - self.cls_repo: set[tuple[str, Callable, Specification]] = set() + self.cls_repo: list[tuple[str, Callable, Specification]] = [] for task in self.luigi_repo: - self.cls_repo.add(task.combinator()) + self.cls_repo.append(task.combinator()) for tpe in task.mro()[1:]: if issubclass(tpe, CoSyLuigiTask): # Is a subclass of CosyLuigiTask, but a superclass of task diff --git a/tests/test_abstract_variant_expansion.py b/tests/test_abstract_variant_expansion.py index e9f53af..ff989f0 100644 --- a/tests/test_abstract_variant_expansion.py +++ b/tests/test_abstract_variant_expansion.py @@ -3,6 +3,9 @@ from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask +class ABCInheritedTaskWithNoInheritors(CoSyLuigiTask, ABC): + pass + class ABCInheritedTask(CoSyLuigiTask, ABC): pass @@ -15,11 +18,19 @@ class DeeperConcreteTaskFromABCInherited(ConcreteTaskFromABCInherited): pass + # noinspection PyAbstractClass class AbstractTask(CoSyLuigiTask): @abstractmethod def get_class_name(self): - pass + raise NotImplementedError + + +# noinspection PyAbstractClass +class AbstractTaskWithNoInheritors(CoSyLuigiTask): + @abstractmethod + def get_class_name(self): + raise NotImplementedError class ConcreteTaskFromAbstract(AbstractTask): @@ -32,13 +43,17 @@ def get_class_name(self): return "DeeperConcreteTaskFromAbstract" -def test_abstract_variant_expansion(): +def test_expansion_from_abc(): repo = CoSyLuigiRepo(ABCInheritedTask) assert repo.luigi_repo == {ConcreteTaskFromABCInherited, DeeperConcreteTaskFromABCInherited} + +def test_expansion_from_abstract(): repo = CoSyLuigiRepo(AbstractTask) assert ConcreteTaskFromAbstract().get_class_name() == "ConcreteTaskFromAbstract" assert DeeperConcreteTaskFromAbstract().get_class_name() == "DeeperConcreteTaskFromAbstract" assert repo.luigi_repo == {ConcreteTaskFromAbstract, DeeperConcreteTaskFromAbstract} + +def test_expansion_from_abc_and_abstract(): repo = CoSyLuigiRepo(ABCInheritedTask, AbstractTask) assert repo.luigi_repo == { ConcreteTaskFromABCInherited, @@ -46,5 +61,19 @@ def test_abstract_variant_expansion(): ConcreteTaskFromAbstract, DeeperConcreteTaskFromAbstract, } + +def test_implementation_of_abstract_does_not_expand(): repo = CoSyLuigiRepo(ConcreteTaskFromAbstract) assert repo.luigi_repo == {ConcreteTaskFromAbstract} + +def test_implementation_of_abc_does_not_expand(): + repo = CoSyLuigiRepo(ConcreteTaskFromABCInherited) + assert repo.luigi_repo == {ConcreteTaskFromABCInherited} + +def test_expansion_to_nothing_from_abc_with_no_inheritors(): + repo = CoSyLuigiRepo(ABCInheritedTaskWithNoInheritors) + assert repo.luigi_repo == set() + +def test_expansion_to_nothing_from_abstract_with_no_inheritors(): + repo = CoSyLuigiRepo(AbstractTaskWithNoInheritors) + assert repo.luigi_repo == set() diff --git a/tests/test_heterogeneous_repo_input.py b/tests/test_heterogeneous_repo_input.py index a61ee36..6bc1f78 100644 --- a/tests/test_heterogeneous_repo_input.py +++ b/tests/test_heterogeneous_repo_input.py @@ -49,7 +49,21 @@ class TaskK(CoSyLuigiTask): def test_heterogeneous_repo_input(): repo = CoSyLuigiRepo(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) assert repo.luigi_repo == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} + +def test_heterogeneous_args_input(): flattened_collection = set( flatten(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) ) assert flattened_collection == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} + +def test_heterogeneous_list_input(): + flattened_collection = list( + flatten([TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])]) + ) + assert flattened_collection == [TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK] + +def test_heterogeneous_tuple_input(): + flattened_collection = tuple( + flatten((TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK]))) + ) + assert flattened_collection == (TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK) \ No newline at end of file diff --git a/tests/test_infinite_chain.py b/tests/test_infinite_chain.py index 801950a..ea5a1c4 100644 --- a/tests/test_infinite_chain.py +++ b/tests/test_infinite_chain.py @@ -10,12 +10,6 @@ class ChainLink(CoSyLuigiTask, ABC): chain_link: CoSyLuigiTaskParameter | None - def output(self): - return MockTarget("ChainLink") - - def run(self): - self.output().open("w").write("Ok.") - class StartingLink(ChainLink): chain_link = None @@ -28,12 +22,6 @@ class RepeatingLink(ChainLink): class FinalLink(CoSyLuigiTask): chain_link = CoSyLuigiTaskParameter(ChainLink) - def output(self): - return MockTarget("FinalLink") - - def run(self): - self.output().open("w").write("Ok.") - @pytest.fixture def repo(): diff --git a/tests/test_unique_task_parameter.py b/tests/test_unique_task_parameter.py new file mode 100644 index 0000000..4ca4815 --- /dev/null +++ b/tests/test_unique_task_parameter.py @@ -0,0 +1,119 @@ +import logging +from abc import ABC +from typing import Sequence, Callable + +import pytest +from cosy.maestro import Maestro + +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi.constraints import is_unique_in_prior_tasks + + +class ScaleDataABC(CoSyLuigiTask, ABC): + pass + +class ScaleData(ScaleDataABC): + pass + +class ScaleDataVariantA(ScaleData): + pass + +class ScaleDataVariantB(ScaleData): + pass + +class TrainModel(CoSyLuigiTask, ABC): + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + +class TrainModelVariantA(TrainModel): + pass + +class TrainModelVariantB(TrainModel): + pass + +class EvaluatePipelineWithUniqueScaler(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC, unique_across_prior_tasks=True) + +class EvaluatePipelineWithConstraintUniqueScaler(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + + @classmethod + def constraints(cls) -> Sequence[Callable[..., bool]]: + return [lambda vs: is_unique_in_prior_tasks(vs, ScaleDataABC)] + +class EvaluatePipelineWithUniqueScalerAndNonAbstractSuper(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleData, unique_across_prior_tasks=True) + +class EvaluatePipeline(CoSyLuigiTask): + train_model = CoSyLuigiTaskParameter(TrainModel) + scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + +@pytest.fixture +def repo_without_constraints(): + return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipeline) + +@pytest.fixture +def repo_with_constraints(): + return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipelineWithUniqueScaler) + +@pytest.fixture +def repo_with_manual_constraints(): + return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipelineWithConstraintUniqueScaler) + +@pytest.fixture +def repo_with_non_abstract_super(): + return CoSyLuigiRepo(TrainModel, ScaleData, ScaleDataVariantA, ScaleDataVariantB, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper) + +def test_implementation_is_not_unique_across_prior_tasks(repo_without_constraints): + maestro = Maestro( + repo_without_constraints.cls_repo, + repo_without_constraints.taxonomy, + ) + results = list(maestro.query(EvaluatePipeline.target())) + assert len(results) == 18 + +def test_implementation_is_unique_across_prior_tasks(repo_with_constraints): + maestro = Maestro( + repo_with_constraints.cls_repo, + repo_with_constraints.taxonomy, + ) + results: list[EvaluatePipelineWithUniqueScaler] = list(maestro.query(EvaluatePipelineWithUniqueScaler.target())) + assert len(results) == 6 + for result in results: + assert result.scaled_data == result.train_model.scaled_data + +def test_implementation_is_unique_across_prior_tasks_with_manual_constraint(repo_with_manual_constraints): + maestro = Maestro( + repo_with_manual_constraints.cls_repo, + repo_with_manual_constraints.taxonomy, + ) + results: list[EvaluatePipelineWithConstraintUniqueScaler] = list(maestro.query(EvaluatePipelineWithConstraintUniqueScaler.target())) + assert len(results) == 6 + for result in results: + assert result.scaled_data == result.train_model.scaled_data + +def test_implementation_is_unique_across_prior_tasks_with_non_abstract_super(repo_with_non_abstract_super): + maestro = Maestro( + repo_with_non_abstract_super.cls_repo, + repo_with_non_abstract_super.taxonomy, + ) + results: list[EvaluatePipelineWithUniqueScalerAndNonAbstractSuper] = list(maestro.query(EvaluatePipelineWithUniqueScalerAndNonAbstractSuper.target())) + assert len(results) == 6 + for result in results: + assert result.scaled_data == result.train_model.scaled_data + +def test_warning_if_unique_across_prior_tasks_but_no_variance(caplog): + caplog.set_level(logging.WARNING) + repo_with_constraints_and_no_variance = CoSyLuigiRepo(TrainModel, ScaleData, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper) + assert len(caplog.records) == 1 + assert caplog.records[0].levelname == "WARNING" + assert "ScaleData" in caplog.records[0].message + + maestro = Maestro( + repo_with_constraints_and_no_variance.cls_repo, + repo_with_constraints_and_no_variance.taxonomy, + ) + results = list(maestro.query(EvaluatePipelineWithUniqueScalerAndNonAbstractSuper.target())) + assert len(results) == 2 From ae767868fe8563e5d4b9155c9aa730481cd90cd3 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 13:59:38 +0200 Subject: [PATCH 05/11] Format with ruff --- src/cosy_luigi/constraints/unique.py | 6 ++-- tests/test_abstract_variant_expansion.py | 8 ++++- tests/test_heterogeneous_repo_input.py | 5 +++- tests/test_infinite_chain.py | 1 - tests/test_unique_task_parameter.py | 37 ++++++++++++++++++++---- 5 files changed, 45 insertions(+), 12 deletions(-) diff --git a/src/cosy_luigi/constraints/unique.py b/src/cosy_luigi/constraints/unique.py index b4d9d3e..da42f4d 100644 --- a/src/cosy_luigi/constraints/unique.py +++ b/src/cosy_luigi/constraints/unique.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import Iterable -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING from cosy_luigi.utils import traverse_pipeline @@ -32,7 +32,5 @@ def is_unique_in_prior_tasks( ) -> bool: return _is_unique_in_prior_tasks( vs, - required_to_be_unique - if isinstance(required_to_be_unique, Iterable) - else [required_to_be_unique], + required_to_be_unique if isinstance(required_to_be_unique, Iterable) else [required_to_be_unique], ) diff --git a/tests/test_abstract_variant_expansion.py b/tests/test_abstract_variant_expansion.py index ff989f0..6978448 100644 --- a/tests/test_abstract_variant_expansion.py +++ b/tests/test_abstract_variant_expansion.py @@ -6,6 +6,7 @@ class ABCInheritedTaskWithNoInheritors(CoSyLuigiTask, ABC): pass + class ABCInheritedTask(CoSyLuigiTask, ABC): pass @@ -18,7 +19,6 @@ class DeeperConcreteTaskFromABCInherited(ConcreteTaskFromABCInherited): pass - # noinspection PyAbstractClass class AbstractTask(CoSyLuigiTask): @abstractmethod @@ -47,12 +47,14 @@ def test_expansion_from_abc(): repo = CoSyLuigiRepo(ABCInheritedTask) assert repo.luigi_repo == {ConcreteTaskFromABCInherited, DeeperConcreteTaskFromABCInherited} + def test_expansion_from_abstract(): repo = CoSyLuigiRepo(AbstractTask) assert ConcreteTaskFromAbstract().get_class_name() == "ConcreteTaskFromAbstract" assert DeeperConcreteTaskFromAbstract().get_class_name() == "DeeperConcreteTaskFromAbstract" assert repo.luigi_repo == {ConcreteTaskFromAbstract, DeeperConcreteTaskFromAbstract} + def test_expansion_from_abc_and_abstract(): repo = CoSyLuigiRepo(ABCInheritedTask, AbstractTask) assert repo.luigi_repo == { @@ -62,18 +64,22 @@ def test_expansion_from_abc_and_abstract(): DeeperConcreteTaskFromAbstract, } + def test_implementation_of_abstract_does_not_expand(): repo = CoSyLuigiRepo(ConcreteTaskFromAbstract) assert repo.luigi_repo == {ConcreteTaskFromAbstract} + def test_implementation_of_abc_does_not_expand(): repo = CoSyLuigiRepo(ConcreteTaskFromABCInherited) assert repo.luigi_repo == {ConcreteTaskFromABCInherited} + def test_expansion_to_nothing_from_abc_with_no_inheritors(): repo = CoSyLuigiRepo(ABCInheritedTaskWithNoInheritors) assert repo.luigi_repo == set() + def test_expansion_to_nothing_from_abstract_with_no_inheritors(): repo = CoSyLuigiRepo(AbstractTaskWithNoInheritors) assert repo.luigi_repo == set() diff --git a/tests/test_heterogeneous_repo_input.py b/tests/test_heterogeneous_repo_input.py index 6bc1f78..6ffa6dd 100644 --- a/tests/test_heterogeneous_repo_input.py +++ b/tests/test_heterogeneous_repo_input.py @@ -50,20 +50,23 @@ def test_heterogeneous_repo_input(): repo = CoSyLuigiRepo(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) assert repo.luigi_repo == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} + def test_heterogeneous_args_input(): flattened_collection = set( flatten(TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])) ) assert flattened_collection == {TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK} + def test_heterogeneous_list_input(): flattened_collection = list( flatten([TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK])]) ) assert flattened_collection == [TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK] + def test_heterogeneous_tuple_input(): flattened_collection = tuple( flatten((TaskA, [TaskB, TaskC], (TaskD, TaskE), [TaskF, (TaskG, TaskH)], (TaskI, [TaskJ, TaskK]))) ) - assert flattened_collection == (TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK) \ No newline at end of file + assert flattened_collection == (TaskA, TaskB, TaskC, TaskD, TaskE, TaskF, TaskG, TaskH, TaskI, TaskJ, TaskK) diff --git a/tests/test_infinite_chain.py b/tests/test_infinite_chain.py index ea5a1c4..7e3bc88 100644 --- a/tests/test_infinite_chain.py +++ b/tests/test_infinite_chain.py @@ -2,7 +2,6 @@ import pytest from cosy.maestro import Maestro -from luigi.mock import MockTarget from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter diff --git a/tests/test_unique_task_parameter.py b/tests/test_unique_task_parameter.py index 4ca4815..8cc556f 100644 --- a/tests/test_unique_task_parameter.py +++ b/tests/test_unique_task_parameter.py @@ -1,6 +1,6 @@ import logging from abc import ABC -from typing import Sequence, Callable +from collections.abc import Callable, Sequence import pytest from cosy.maestro import Maestro @@ -12,28 +12,36 @@ class ScaleDataABC(CoSyLuigiTask, ABC): pass + class ScaleData(ScaleDataABC): pass + class ScaleDataVariantA(ScaleData): pass + class ScaleDataVariantB(ScaleData): pass + class TrainModel(CoSyLuigiTask, ABC): scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + class TrainModelVariantA(TrainModel): pass + class TrainModelVariantB(TrainModel): pass + class EvaluatePipelineWithUniqueScaler(CoSyLuigiTask): train_model = CoSyLuigiTaskParameter(TrainModel) scaled_data = CoSyLuigiTaskParameter(ScaleDataABC, unique_across_prior_tasks=True) + class EvaluatePipelineWithConstraintUniqueScaler(CoSyLuigiTask): train_model = CoSyLuigiTaskParameter(TrainModel) scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) @@ -42,29 +50,38 @@ class EvaluatePipelineWithConstraintUniqueScaler(CoSyLuigiTask): def constraints(cls) -> Sequence[Callable[..., bool]]: return [lambda vs: is_unique_in_prior_tasks(vs, ScaleDataABC)] + class EvaluatePipelineWithUniqueScalerAndNonAbstractSuper(CoSyLuigiTask): train_model = CoSyLuigiTaskParameter(TrainModel) scaled_data = CoSyLuigiTaskParameter(ScaleData, unique_across_prior_tasks=True) + class EvaluatePipeline(CoSyLuigiTask): train_model = CoSyLuigiTaskParameter(TrainModel) scaled_data = CoSyLuigiTaskParameter(ScaleDataABC) + @pytest.fixture def repo_without_constraints(): return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipeline) + @pytest.fixture def repo_with_constraints(): return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipelineWithUniqueScaler) + @pytest.fixture def repo_with_manual_constraints(): return CoSyLuigiRepo(TrainModel, ScaleDataABC, EvaluatePipelineWithConstraintUniqueScaler) + @pytest.fixture def repo_with_non_abstract_super(): - return CoSyLuigiRepo(TrainModel, ScaleData, ScaleDataVariantA, ScaleDataVariantB, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper) + return CoSyLuigiRepo( + TrainModel, ScaleData, ScaleDataVariantA, ScaleDataVariantB, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper + ) + def test_implementation_is_not_unique_across_prior_tasks(repo_without_constraints): maestro = Maestro( @@ -74,6 +91,7 @@ def test_implementation_is_not_unique_across_prior_tasks(repo_without_constraint results = list(maestro.query(EvaluatePipeline.target())) assert len(results) == 18 + def test_implementation_is_unique_across_prior_tasks(repo_with_constraints): maestro = Maestro( repo_with_constraints.cls_repo, @@ -84,29 +102,38 @@ def test_implementation_is_unique_across_prior_tasks(repo_with_constraints): for result in results: assert result.scaled_data == result.train_model.scaled_data + def test_implementation_is_unique_across_prior_tasks_with_manual_constraint(repo_with_manual_constraints): maestro = Maestro( repo_with_manual_constraints.cls_repo, repo_with_manual_constraints.taxonomy, ) - results: list[EvaluatePipelineWithConstraintUniqueScaler] = list(maestro.query(EvaluatePipelineWithConstraintUniqueScaler.target())) + results: list[EvaluatePipelineWithConstraintUniqueScaler] = list( + maestro.query(EvaluatePipelineWithConstraintUniqueScaler.target()) + ) assert len(results) == 6 for result in results: assert result.scaled_data == result.train_model.scaled_data + def test_implementation_is_unique_across_prior_tasks_with_non_abstract_super(repo_with_non_abstract_super): maestro = Maestro( repo_with_non_abstract_super.cls_repo, repo_with_non_abstract_super.taxonomy, ) - results: list[EvaluatePipelineWithUniqueScalerAndNonAbstractSuper] = list(maestro.query(EvaluatePipelineWithUniqueScalerAndNonAbstractSuper.target())) + results: list[EvaluatePipelineWithUniqueScalerAndNonAbstractSuper] = list( + maestro.query(EvaluatePipelineWithUniqueScalerAndNonAbstractSuper.target()) + ) assert len(results) == 6 for result in results: assert result.scaled_data == result.train_model.scaled_data + def test_warning_if_unique_across_prior_tasks_but_no_variance(caplog): caplog.set_level(logging.WARNING) - repo_with_constraints_and_no_variance = CoSyLuigiRepo(TrainModel, ScaleData, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper) + repo_with_constraints_and_no_variance = CoSyLuigiRepo( + TrainModel, ScaleData, EvaluatePipelineWithUniqueScalerAndNonAbstractSuper + ) assert len(caplog.records) == 1 assert caplog.records[0].levelname == "WARNING" assert "ScaleData" in caplog.records[0].message From 63962d3b8e8207a59001655049fef4c2a63d6933 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 15:53:59 +0200 Subject: [PATCH 06/11] bugfix: no variance in ml blood sugar --- examples/ml_blood_sugar_level/ml_blood_sugar_level.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py index d76385d..2e5365d 100644 --- a/examples/ml_blood_sugar_level/ml_blood_sugar_level.py +++ b/examples/ml_blood_sugar_level/ml_blood_sugar_level.py @@ -145,7 +145,7 @@ def run(self): class EvaluateRegressionModel(CoSyLuigiTask): regressor = CoSyLuigiTaskParameter(TrainRegressionModel) - scaled_feats = CoSyLuigiTaskParameter(FitTransformMinMaxScaler, unique_across_prior_tasks=True) + scaled_feats = CoSyLuigiTaskParameter(FitTransformScaler, unique_across_prior_tasks=True) splitted_data = CoSyLuigiTaskParameter(TrainTestSplit) def _get_variant_label(self): From 354fe54156a002ac0628b382967e858b74d6618b Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 15:55:30 +0200 Subject: [PATCH 07/11] bugfix: benchmark imports --- benchmarks/test_benchmark_chain_100.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/benchmarks/test_benchmark_chain_100.py b/benchmarks/test_benchmark_chain_100.py index a7ce05c..2fdbdaf 100644 --- a/benchmarks/test_benchmark_chain_100.py +++ b/benchmarks/test_benchmark_chain_100.py @@ -4,7 +4,7 @@ from cosy.maestro import Maestro from luigi.mock import MockTarget -from cosy_luigi.combinatorics import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter +from cosy_luigi import CoSyLuigiRepo, CoSyLuigiTask, CoSyLuigiTaskParameter class ChainLink(CoSyLuigiTask, ABC): From 2017735dc2c0752e640e20c8a3f5509b51869868 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 17:54:01 +0200 Subject: [PATCH 08/11] Additional speedups and safety --- src/cosy_luigi/constraints/unique.py | 8 +++--- src/cosy_luigi/core/combinatorics.py | 41 +++++++++++++++------------- src/cosy_luigi/utils/traversals.py | 12 ++++---- 3 files changed, 33 insertions(+), 28 deletions(-) diff --git a/src/cosy_luigi/constraints/unique.py b/src/cosy_luigi/constraints/unique.py index da42f4d..8f0e9df 100644 --- a/src/cosy_luigi/constraints/unique.py +++ b/src/cosy_luigi/constraints/unique.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Iterable +from collections.abc import Sequence from typing import TYPE_CHECKING from cosy_luigi.utils import traverse_pipeline @@ -12,7 +12,7 @@ def _is_unique_in_prior_tasks( - vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Iterable[type[CoSyLuigiTask]] + vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: Sequence[type[CoSyLuigiTask]] ) -> bool: classes = [pc.__class__ for pc in traverse_pipeline(vs.values())] seen_subclasses: dict[type[CoSyLuigiTask], type[CoSyLuigiTask]] = {} @@ -28,9 +28,9 @@ def _is_unique_in_prior_tasks( def is_unique_in_prior_tasks( - vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]] + vs: Mapping[str, CoSyLuigiTask], required_to_be_unique: type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]] ) -> bool: return _is_unique_in_prior_tasks( vs, - required_to_be_unique if isinstance(required_to_be_unique, Iterable) else [required_to_be_unique], + required_to_be_unique if isinstance(required_to_be_unique, Sequence) else [required_to_be_unique], ) diff --git a/src/cosy_luigi/core/combinatorics.py b/src/cosy_luigi/core/combinatorics.py index 06d9f2d..b797502 100644 --- a/src/cosy_luigi/core/combinatorics.py +++ b/src/cosy_luigi/core/combinatorics.py @@ -11,7 +11,7 @@ from luigi.task_register import Register if TYPE_CHECKING: - from collections.abc import Callable, Iterable, Mapping, Sequence + from collections.abc import Callable, Mapping, Sequence from cosy.core.synthesizer import Specification @@ -26,10 +26,6 @@ def __init__(self, required_task: type[CoSyLuigiTask], *, unique_across_prior_ta class CoSyLuigiTask(luigi.Task): - def __init_subclass__(cls, **kwargs): - super().__init_subclass__(**kwargs) - cls.self = cls - @classmethod @cache def get_all_variants(cls): @@ -56,32 +52,35 @@ def requires(self): :return: A list of other tasks required to run this task """ return { - k: v + k: v.required_task() for k, v in self.get_all_instance_attributes().items() - if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTask) + if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTaskParameter) } @classmethod @cache - def _requirements(cls) -> Mapping[str, CoSyLuigiTaskParameter]: - return { - k: v + def _requirements(cls) -> Sequence[tuple[str, CoSyLuigiTaskParameter]]: + return [ + (k, v) for k, v in cls.get_all_class_attributes().items() if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTaskParameter) - } + ] + + @classmethod + @cache + def get_params(cls): + return cls._requirements() @classmethod @cache def requirements_unique_in_prior_tasks(cls) -> Mapping[str, CoSyLuigiTaskParameter]: return { - k: task_parameter - for k, task_parameter in cls._requirements().items() - if task_parameter.unique_across_prior_tasks + k: task_parameter for k, task_parameter in cls._requirements() if task_parameter.unique_across_prior_tasks } @classmethod @cache - def unique_required_tasks_in_prior(cls) -> Iterable[type[CoSyLuigiTask]]: + def unique_required_tasks_in_prior(cls) -> Sequence[type[CoSyLuigiTask]]: return [task_parameter.required_task for task_parameter in cls.requirements_unique_in_prior_tasks().values()] @classmethod @@ -102,8 +101,8 @@ def __constraints(cls) -> Sequence[Callable[..., bool]]: @classmethod def combinator_type(cls): sp = SpecificationBuilder() - for name in [v.required_task.__name__ for v in cls._requirements().values()]: - sp = sp.argument(name, Constructor(name)) + for name, type_name in [(k, v.required_task.__name__) for k, v in cls._requirements()]: + sp = sp.argument(name, Constructor(type_name)) for constraint in cls.__constraints(): sp = sp.constraint(constraint) for constraint in cls.constraints(): @@ -114,11 +113,15 @@ def combinator_type(cls): def combinator(cls): if len(cls._requirements()) == 0: return cls.__name__, lambda: cls(), cls.combinator_type() - return cls.__name__, lambda *args: cls(*args), cls.combinator_type() + return ( + cls.__name__, + lambda *vals: cls(**dict(zip([k for k, _ in cls._requirements()], vals, strict=False))), + cls.combinator_type(), + ) class CoSyLuigiRepo: - def __init__(self, *tasks: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): + def __init__(self, *tasks: type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]]): Register.disable_instance_cache() # Accepts completely heterogeneous nested collections diff --git a/src/cosy_luigi/utils/traversals.py b/src/cosy_luigi/utils/traversals.py index fd2f99e..f90bdb7 100644 --- a/src/cosy_luigi/utils/traversals.py +++ b/src/cosy_luigi/utils/traversals.py @@ -7,15 +7,15 @@ from cosy_luigi import CoSyLuigiTask if TYPE_CHECKING: - from collections.abc import Iterable + from collections.abc import Iterable, Sequence -def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Iterable[type[CoSyLuigiTask]]): +def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Sequence[type[CoSyLuigiTask]]): return ( task for task_or_task_collection in heterogeneous_task_collection for task in ( - flatten(*cast("Iterable[type[CoSyLuigiTask]]", task_or_task_collection)) + flatten(*cast("Sequence[type[CoSyLuigiTask]]", task_or_task_collection)) if isinstance(task_or_task_collection, (tuple, list)) else cast("type[CoSyLuigiTask]", task_or_task_collection).get_all_variants() if inspect.isabstract(task_or_task_collection) @@ -25,14 +25,16 @@ def flatten(*heterogeneous_task_collection: type[CoSyLuigiTask] | Iterable[type[ ) -def _traverse_pipeline(vs: Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: +def _traverse_pipeline(vs: Sequence[CoSyLuigiTask] | Iterable[CoSyLuigiTask]) -> Sequence[CoSyLuigiTask]: result = [*vs] for v in vs: result.extend(traverse_pipeline(v.requires().values())) return result -def traverse_pipeline(to_traverse: CoSyLuigiTask | Iterable[CoSyLuigiTask]) -> Iterable[CoSyLuigiTask]: +def traverse_pipeline( + to_traverse: CoSyLuigiTask | Sequence[CoSyLuigiTask] | Iterable[CoSyLuigiTask], +) -> Sequence[CoSyLuigiTask]: return ( _traverse_pipeline([to_traverse]) if isinstance(to_traverse, CoSyLuigiTask) else _traverse_pipeline(to_traverse) ) From bbf2796fd8c6ed0b9fcc5d4df94b527ee86948ea Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 18:36:23 +0200 Subject: [PATCH 09/11] Fix benchmark regression, realize 3x performance speedup --- benchmarks/test_benchmark_chain_100.py | 3 +++ src/cosy_luigi/core/combinatorics.py | 37 ++++++++++++++------------ 2 files changed, 23 insertions(+), 17 deletions(-) diff --git a/benchmarks/test_benchmark_chain_100.py b/benchmarks/test_benchmark_chain_100.py index 2fdbdaf..c942930 100644 --- a/benchmarks/test_benchmark_chain_100.py +++ b/benchmarks/test_benchmark_chain_100.py @@ -50,3 +50,6 @@ def create_infinite_chain(repo): def test_benchmark_chain_creation(repo, benchmark): benchmark(create_infinite_chain, repo) + +if __name__ == "__main__": + create_infinite_chain(repo) diff --git a/src/cosy_luigi/core/combinatorics.py b/src/cosy_luigi/core/combinatorics.py index b797502..878dd40 100644 --- a/src/cosy_luigi/core/combinatorics.py +++ b/src/cosy_luigi/core/combinatorics.py @@ -26,6 +26,10 @@ def __init__(self, required_task: type[CoSyLuigiTask], *, unique_across_prior_ta class CoSyLuigiTask(luigi.Task): + def __init_subclass__(cls, **kwargs): + super().__init_subclass__(**kwargs) + cls.self = cls + @classmethod @cache def get_all_variants(cls): @@ -52,30 +56,32 @@ def requires(self): :return: A list of other tasks required to run this task """ return { - k: v.required_task() + k: v for k, v in self.get_all_instance_attributes().items() - if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTaskParameter) + if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTask) } @classmethod @cache - def _requirements(cls) -> Sequence[tuple[str, CoSyLuigiTaskParameter]]: - return [ - (k, v) + def _requirements(cls) -> Mapping[str, CoSyLuigiTaskParameter]: + return { + k: v for k, v in cls.get_all_class_attributes().items() if not k.startswith("__") and not callable(v) and issubclass(v.__class__, CoSyLuigiTaskParameter) - ] + } @classmethod @cache def get_params(cls): - return cls._requirements() + return list(cls._requirements().items()) @classmethod @cache def requirements_unique_in_prior_tasks(cls) -> Mapping[str, CoSyLuigiTaskParameter]: return { - k: task_parameter for k, task_parameter in cls._requirements() if task_parameter.unique_across_prior_tasks + k: task_parameter + for k, task_parameter in cls._requirements().items() + if task_parameter.unique_across_prior_tasks } @classmethod @@ -95,14 +101,15 @@ def constraints(cls) -> Sequence[Callable[..., bool]]: @classmethod def __constraints(cls) -> Sequence[Callable[..., bool]]: from cosy_luigi.constraints.unique import _is_unique_in_prior_tasks # noqa: PLC0415 - - return [partial(_is_unique_in_prior_tasks, required_to_be_unique=cls.unique_required_tasks_in_prior())] + if cls.requirements_unique_in_prior_tasks(): + return [partial(_is_unique_in_prior_tasks, required_to_be_unique=cls.unique_required_tasks_in_prior())] + return [] @classmethod def combinator_type(cls): sp = SpecificationBuilder() - for name, type_name in [(k, v.required_task.__name__) for k, v in cls._requirements()]: - sp = sp.argument(name, Constructor(type_name)) + for name in [v.required_task.__name__ for v in cls._requirements().values()]: + sp = sp.argument(name, Constructor(name)) for constraint in cls.__constraints(): sp = sp.constraint(constraint) for constraint in cls.constraints(): @@ -113,11 +120,7 @@ def combinator_type(cls): def combinator(cls): if len(cls._requirements()) == 0: return cls.__name__, lambda: cls(), cls.combinator_type() - return ( - cls.__name__, - lambda *vals: cls(**dict(zip([k for k, _ in cls._requirements()], vals, strict=False))), - cls.combinator_type(), - ) + return cls.__name__, lambda *args: cls(*args), cls.combinator_type() class CoSyLuigiRepo: From 8d2ffc52d1cc7abaabc089326a08807034fd72e6 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 18:38:39 +0200 Subject: [PATCH 10/11] Format with ruff --- benchmarks/test_benchmark_chain_100.py | 1 + src/cosy_luigi/core/combinatorics.py | 1 + 2 files changed, 2 insertions(+) diff --git a/benchmarks/test_benchmark_chain_100.py b/benchmarks/test_benchmark_chain_100.py index c942930..013666c 100644 --- a/benchmarks/test_benchmark_chain_100.py +++ b/benchmarks/test_benchmark_chain_100.py @@ -51,5 +51,6 @@ def create_infinite_chain(repo): def test_benchmark_chain_creation(repo, benchmark): benchmark(create_infinite_chain, repo) + if __name__ == "__main__": create_infinite_chain(repo) diff --git a/src/cosy_luigi/core/combinatorics.py b/src/cosy_luigi/core/combinatorics.py index 878dd40..fb09bd7 100644 --- a/src/cosy_luigi/core/combinatorics.py +++ b/src/cosy_luigi/core/combinatorics.py @@ -101,6 +101,7 @@ def constraints(cls) -> Sequence[Callable[..., bool]]: @classmethod def __constraints(cls) -> Sequence[Callable[..., bool]]: from cosy_luigi.constraints.unique import _is_unique_in_prior_tasks # noqa: PLC0415 + if cls.requirements_unique_in_prior_tasks(): return [partial(_is_unique_in_prior_tasks, required_to_be_unique=cls.unique_required_tasks_in_prior())] return [] From de06a204e9be95fdb6066ef7dd6a47f4f9c62d47 Mon Sep 17 00:00:00 2001 From: Constantin Chaumet <16308584+Jekannadar@users.noreply.github.com> Date: Mon, 13 Apr 2026 19:09:47 +0200 Subject: [PATCH 11/11] Pytest benchmarks alerts are inverted (smh) --- .github/workflows/benchmarks.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index 5477bb2..c73b346 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -84,7 +84,7 @@ jobs: output-file-path: cosy-luigi-nested/benchmarks.json github-token: ${{ steps.cls-python-workflow-token.outputs.token }} auto-push: ${{ env.PUSH_TO_BENCHMARKS }} - alert-threshold: '300%' + alert-threshold: '50%' comment-on-alert: true fail-on-alert: true comment-always: ${{ env.COMMENT_ALWAYS }}