action(s) to execute on each data element

filter(s) used to select data to process

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

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

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

action to execute before processing

action to execute after processing

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

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

additional keyword arguments

21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

def init(
    self,
    action=None,
    select=None,
    unpack=True,
    column=None,
    merge=”hstack”,
    initialize=None,
    finalize=None,
    concurrency=None,
    onetomany=True,
    kwargs,
):
    “””
    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.
    Args:
        action: action(s) to execute on each data element
        select: filter(s) used to select data to process
        unpack: if data elements should be unpacked or unwrapped from (id, data, tag) tuples
        column: column index to select if element is a tuple, defaults to all
        merge: merge mode for joining multi-action outputs, defaults to hstack
        initialize: action to execute before processing
        finalize: action to execute after processing
        concurrency: sets concurrency method when execute instance available
                     valid values: “thread” for thread-based concurrency, “process” for process-based concurrency
        onetomany: if one-to-many data transformations should be enabled, defaults to True
        kwargs: additional keyword arguments
    “””
    # Standardize into list of actions
    if not action:
        action = []
    elif not isinstance(action, list):
        action = [action]
    self.action = action
    self.select = select
    self.unpack = unpack
    self.column = column
    self.merge = merge
    self.initialize = initialize
    self.finalize = finalize
    self.concurrency = concurrency
    self.onetomany = onetomany
    # Check for custom registration. Adds additional instance members and validates required dependencies available.
    if hasattr(self, “register”):
        self.register(kwargs)
    elif kwargs:
        # Raise error if additional keyword arguments passed in without register method
        kwargs = “, “.join(f”‘{kw}’” for kw in kwargs)
        raise TypeError(f”init() got unexpected keyword arguments: {kwargs}”)

local directory used to store retrieved files

flatten input directory structure, defaults to True

19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

def register(self, directory=None, flatten=True):
    “””
    Adds retrieve parameters to task.
    Args:
        directory: local directory used to store retrieved files
        flatten: flatten input directory structure, defaults to True
    “””
    # pylint: disable=W0201
    # Create default temporary directory if not specified
    if not directory:
        # Save tempdir to prevent content from being deleted until this task is out of scope
        # pylint: disable=R1732
        self.tempdir = tempfile.TemporaryDirectory()
        directory = self.tempdir.name
    # Create output directory if necessary
    os.makedirs(directory, exist_ok=True)
    self.directory = directory
    self.flatten = flatten