Get field set for a collection of tasks.
(tasks)
| 181 | |
| 182 | @staticmethod |
| 183 | def get_fields(tasks): |
| 184 | """Get field set for a collection of tasks.""" |
| 185 | fields = OrderedSet() |
| 186 | for task in tasks: |
| 187 | fields.update(task['operator'].atoms(Field)) |
| 188 | # Drop locked fields |
| 189 | locked = [f for f in fields if isinstance(f, LockedField)] |
| 190 | for field in locked: |
| 191 | fields.pop(field) |
| 192 | return fields |
| 193 | |
| 194 | @staticmethod |
| 195 | def attempt_tasks(tasks, **kw): |
no test coverage detected