Workflow Task

task task

The Workflow Task runs a workflow. Allows creating workflows of workflows.

Example

The following shows a simple example using this task as part of a workflow.

  1. from txtai.workflow import WorkflowTask, Workflow
  3. workflow = Workflow([WorkflowTask(otherworkflow)])
  4. workflow(["input data"])

Methods

Python documentation for the task.

Creates a new task. A task defines two methods, type of data it accepts and the action to execute for each data element. Action is a callable function or list of callable functions.

Parameters:

NameTypeDescriptionDefault
action

action(s) to execute on each data element

None
select

filter(s) used to select data to process

None
unpack

if data elements should be unpacked or unwrapped from (id, data, tag) tuples

True
column

column index to select if element is a tuple, defaults to all

None
merge

merge mode for joining multi-action outputs, defaults to hstack

‘hstack’
initialize

action to execute before processing

None
finalize

action to execute after processing

None
concurrency

sets concurrency method when execute instance available valid values: “thread” for thread-based concurrency, “process” for process-based concurrency

None
onetomany

if one-to-many data transformations should be enabled, defaults to True

True
kwargs

additional keyword arguments

{}

Source code in txtai/workflow/task/base.py

  1. 21
  2. 22
  3. 23
  4. 24
  5. 25
  6. 26
  7. 27
  8. 28
  9. 29
  10. 30
  11. 31
  12. 32
  13. 33
  14. 34
  15. 35
  16. 36
  17. 37
  18. 38
  19. 39
  20. 40
  21. 41
  22. 42
  23. 43
  24. 44
  25. 45
  26. 46
  27. 47
  28. 48
  29. 49
  30. 50
  31. 51
  32. 52
  33. 53
  34. 54
  35. 55
  36. 56
  37. 57
  38. 58
  39. 59
  40. 60
  41. 61
  42. 62
  43. 63
  44. 64
  45. 65
  46. 66
  47. 67
  48. 68
  49. 69
  50. 70
  51. 71
  52. 72
  53. 73
  54. 74
  1. def init(
  2.     self,
  3.     action=None,
  4.     select=None,
  5.     unpack=True,
  6.     column=None,
  7.     merge=”hstack”,
  8.     initialize=None,
  9.     finalize=None,
  10.     concurrency=None,
  11.     onetomany=True,
  12.     kwargs,
  13. ):
  14.     “””
  15.     Creates a new task. A task defines two methods, type of data it accepts and the action to execute
  16.     for each data element. Action is a callable function or list of callable functions.
  18.     Args:
  19.         action: action(s) to execute on each data element
  20.         select: filter(s) used to select data to process
  21.         unpack: if data elements should be unpacked or unwrapped from (id, data, tag) tuples
  22.         column: column index to select if element is a tuple, defaults to all
  23.         merge: merge mode for joining multi-action outputs, defaults to hstack
  24.         initialize: action to execute before processing
  25.         finalize: action to execute after processing
  26.         concurrency: sets concurrency method when execute instance available
  27.                      valid values: “thread” for thread-based concurrency, “process” for process-based concurrency
  28.         onetomany: if one-to-many data transformations should be enabled, defaults to True
  29.         kwargs: additional keyword arguments
  30.     “””
  32.     # Standardize into list of actions
  33.     if not action:
  34.         action = []
  35.     elif not isinstance(action, list):
  36.         action = [action]
  38.     self.action = action
  39.     self.select = select
  40.     self.unpack = unpack
  41.     self.column = column
  42.     self.merge = merge
  43.     self.initialize = initialize
  44.     self.finalize = finalize
  45.     self.concurrency = concurrency
  46.     self.onetomany = onetomany
  48.     # Check for custom registration. Adds additional instance members and validates required dependencies available.
  49.     if hasattr(self, “register”):
  50.         self.register(kwargs)
  51.     elif kwargs:
  52.         # Raise error if additional keyword arguments passed in without register method
  53.         kwargs = “, “.join(f”‘{kw}’” for kw in kwargs)
  54.         raise TypeError(f”init() got unexpected keyword arguments: {kwargs}”)