CedricAnover / CedricAnover/marsh
[FEATURE] DAG - Add Node factory method from a list of tuples
- Dominant language
- Python
- Stars
- 1
- Forks
- 0
- PR merge metrics
- No merged PRs in 30d
Description
## Is your feature request related to a problem? Please describe.
Add factory method to create a Node from a list of tuples representing the ingredients of conveyor and command runners.
## Describe the solution you'd like
```python
@classmethod
def from_cmd_runners(cls, name: str, *tuples: tuple) -> "Node":
assert tuples, "The ingredients for constructing command runners and conveyor cannot be empty."
# (cmd_runner, args, kwargs, cmd_runner_decorator)
# Initialize Conveyor
conveyor = Conveyor()
for tuple_ in tuples:
match tuple_:
case (cmd_runner,) if callable(cmd_runner):
conveyor = conveyor.add_cmd_runner(cmd_runner)
case (cmd_runner, args, kwargs, cmd_runner_decorator) \
if callable(cmd_runner) and isinstance(args, tuple) \
and isinstance(kwargs, dict) and isinstance(cmd_runner_decorator, CmdRunDecorator):
conveyor = conveyor.add_cmd_runner(
cmd_runner, *args, cmd_runner_decorator=cmd_runner_decorator, **kwargs
)
case (cmd_runner, cmd_runner_decorator) \
if callable(cmd_runner) and isinstance(cmd_runner_decorator, CmdRunDecorator):
conveyor = conveyor.add_cmd_runner(
cmd_runner, cmd_runner_decorator=cmd_runner_decorator
)
case (cmd_runner, args, kwargs) \
if callable(cmd_runner) and isinstance(args, tuple) and isinstance(kwargs, dict):
conveyor = conveyor.add_cmd_runner(cmd_runner, *args, **kwargs)
case (cmd_runner, args) \
if callable(cmd_runner) and isinstance(args, tuple):
conveyor = conveyor.add_cmd_runner(cmd_runner, *args)
case (cmd_runner, args, cmd_runner_decorator) \
if callable(cmd_runner) and isinstance(args, tuple) \
and isinstance(cmd_runner_decorator, CmdRunDecorator):
conveyor = conveyor.add_cmd_runner(
cmd_runner, *args, cmd_runner_decorator=cmd_runner_decorator
)
case (cmd_runner, kwargs) \
if callable(cmd_runner) and isinstance(kwargs, dict):
conveyor = conveyor.add_cmd_runner(cmd_runner, **kwargs)
case (cmd_runner, kwargs, cmd_runner_decorator) \
if callable(cmd_runner) and isinstance(kwargs, dict) \
and isinstance(cmd_runner_decorator, CmdRunDecorator):
conveyor = conveyor.add_cmd_runner(
cmd_runner, cmd_runner_decorator=cmd_runner_decorator, **kwargs
)
case _:
raise ValueError(f"{tuple_} is invalid.")
return cls(name, conveyor)
```
## Describe alternatives you've considered
...
## Additional context
...
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.