This class represents the specification of a workflow.
| 21 | |
| 22 | |
| 23 | class WorkflowSpec(object): |
| 24 | |
| 25 | """ |
| 26 | This class represents the specification of a workflow. |
| 27 | """ |
| 28 | |
| 29 | def __init__(self, name=None, filename=None, addstart=False): |
| 30 | """ |
| 31 | Constructor. |
| 32 | """ |
| 33 | self.name = name or '' |
| 34 | self.description = '' |
| 35 | self.file = filename |
| 36 | self.task_specs = dict() |
| 37 | self.start = None |
| 38 | if addstart: |
| 39 | self.start = StartTask(self) |
| 40 | |
| 41 | def _add_notify(self, task_spec): |
| 42 | """ |
| 43 | Called by a task spec when it was added into the workflow. |
| 44 | """ |
| 45 | if task_spec.name in self.task_specs: |
| 46 | raise KeyError('Duplicate task spec name: ' + task_spec.name) |
| 47 | self.task_specs[task_spec.name] = task_spec |
| 48 | |
| 49 | def get_task_spec_from_name(self, name): |
| 50 | """ |
| 51 | Returns the task with the given name. |
| 52 | |
| 53 | :type name: str |
| 54 | :param name: The name of the task spec. |
| 55 | :rtype: TaskSpec |
| 56 | :returns: The task spec with the given name. |
| 57 | """ |
| 58 | return self.task_specs.get(name) |
| 59 | |
| 60 | def validate(self): |
| 61 | """Checks integrity of workflow and reports any problems with it. |
| 62 | |
| 63 | Detects: |
| 64 | - loops (tasks that wait on each other in a loop) |
| 65 | :returns: empty list if valid, a list of errors if not |
| 66 | """ |
| 67 | results = [] |
| 68 | from ..specs.Join import Join |
| 69 | |
| 70 | def recursive_find_loop(task, history): |
| 71 | current = history[:] |
| 72 | current.append(task) |
| 73 | if isinstance(task, Join): |
| 74 | if task in history: |
| 75 | msg = "Found loop with '%s': %s then '%s' again" % ( |
| 76 | task.name, '->'.join([p.name for p in history]), |
| 77 | task.name) |
| 78 | raise Exception(msg) |
| 79 | for predecessor in task.inputs: |
| 80 | recursive_find_loop(predecessor, current) |
no outgoing calls