Skip to content

taps.engine.transform

TaskTransformer 

TaskTransformer(
    transformer: Transformer[IdentifierT] | None = None,
    filter_: Filter | None = None,
)

Bases: Generic[IdentifierT]

Task data transformer.

This class combines a simple object Transformer and a Filter into useful methods for transforming the positional arguments, keyword arguments, and results of tasks.

Parameters:

  • transformer (Transformer[IdentifierT] | None, default: None ) –

    Object transformer. If None, no objects will be transformed.

  • filter_ (Filter | None, default: None ) –

    A filter which when called on an object returns True if the object should be transformed. If None, all objects will be transformed.

Source code in taps/engine/transform.py 
50
51
52
53
54
55
56
def __init__(
    self,
    transformer: Transformer[IdentifierT] | None = None,
    filter_: Filter | None = None,
) -> None:
    self.transformer = transformer
    self.filter_ = filter_

close 

close() -> None

Close the transformer.

Source code in taps/engine/transform.py 
75
76
77
78
def close(self) -> None:
    """Close the transformer."""
    if self.transformer is not None:
        self.transformer.close()

transform 

transform(obj: T) -> T | IdentifierT

Transform an object.

Transforms obj into an identifier if it passes the filter check. The identifier can later be used to resolve the object.

Source code in taps/engine/transform.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def transform(self, obj: T) -> T | IdentifierT:
    """Transform an object.

    Transforms `obj` into an identifier if it passes the filter check.
    The identifier can later be used to resolve the object.
    """
    filtered = self.filter_ is None or self.filter_(obj)
    if filtered and self.transformer is not None and not is_future(obj):
        identifier = self.transformer.transform(obj)
        logger.log(
            TRACE_LOG_LEVEL,
            f'Transformed object (type={type(obj).__name__}) into '
            f'identifier (type={type(identifier).__name__})',
        )
        return identifier
    else:
        return obj

transform_iterable 

transform_iterable(
    iterable: Iterable[T],
) -> tuple[T | IdentifierT, ...]

Transform each object in an iterable.

Source code in taps/engine/transform.py 
 98
 99
100
101
102
103
def transform_iterable(
    self,
    iterable: Iterable[T],
) -> tuple[T | IdentifierT, ...]:
    """Transform each object in an iterable."""
    return tuple(self.transform(obj) for obj in iterable)

transform_mapping 

transform_mapping(
    mapping: Mapping[K, T]
) -> dict[K, Any]

Transform each value in a mapping.

Source code in taps/engine/transform.py 
105
106
107
def transform_mapping(self, mapping: Mapping[K, T]) -> dict[K, Any]:
    """Transform each value in a mapping."""
    return {k: self.transform(v) for k, v in mapping.items()}

resolve 

resolve(obj: Any) -> Any

Resolve an object.

Resolves the object if it is an identifier, otherwise returns the passed object.

Source code in taps/engine/transform.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def resolve(self, obj: Any) -> Any:
    """Resolve an object.

    Resolves the object if it is an identifier, otherwise returns the
    passed object.
    """
    if self.transformer is not None and self.transformer.is_identifier(
        obj,
    ):
        result = self.transformer.resolve(obj)
        logger.log(
            TRACE_LOG_LEVEL,
            f'Resolved identifier (type={type(obj).__name__}) into '
            f'object (type={type(result).__name__})',
        )
        return result
    else:
        return obj

resolve_iterable 

resolve_iterable(
    iterable: Iterable[Any],
) -> tuple[Any, ...]

Resolve each object in an iterable.

Source code in taps/engine/transform.py 
128
129
130
def resolve_iterable(self, iterable: Iterable[Any]) -> tuple[Any, ...]:
    """Resolve each object in an iterable."""
    return tuple(self.resolve(obj) for obj in iterable)

resolve_mapping 

resolve_mapping(mapping: Mapping[K, Any]) -> dict[K, Any]

Resolve each value in a mapping.

Source code in taps/engine/transform.py 
132
133
134
def resolve_mapping(self, mapping: Mapping[K, Any]) -> dict[K, Any]:
    """Resolve each value in a mapping."""
    return {k: self.resolve(v) for k, v in mapping.items()}