Skip to content

taps.transformer

taps/transformer/__init__.py

Transformer

Bases: Protocol[IdentifierT]

Data transformer protocol.

A data transformer is used by the Engine to transform task parameters and results into alternative formats that are more suitable for communication.

An object can be transformed using transform() which returns an identifier. The identifier can then be provided to resolve(), the inverse of transform(), which returns the original object.

Data transformer implementations can implement object identifiers in any manner, provided identifiers are serializable. For example, a simple identifier could be a UUID corresponding to a database entry containing the serialized object.

close() 

close() -> None

Close the transformer.

The transformer is only closed by the client once the application has finished executing (or raised an exception).

Source code in taps/transformer/_protocol.py 
39
40
41
42
43
44
45
def close(self) -> None:
    """Close the transformer.

    The transformer is only closed by the client once the application
    has finished executing (or raised an exception).
    """
    ...

is_identifier() 

is_identifier(obj: T) -> bool

Check if the object is an identifier instance.

Source code in taps/transformer/_protocol.py 
47
48
49
def is_identifier(self, obj: T) -> bool:
    """Check if the object is an identifier instance."""
    ...

transform() 

transform(obj: T) -> IdentifierT

Transform the object into an identifier.

Parameters:

  • obj (T) –

    Object to transform.

Returns:

  • IdentifierT –

    Identifier object that can be used to resolve obj.

Source code in taps/transformer/_protocol.py 
51
52
53
54
55
56
57
58
59
60
def transform(self, obj: T) -> IdentifierT:
    """Transform the object into an identifier.

    Args:
        obj: Object to transform.

    Returns:
        Identifier object that can be used to resolve `obj`.
    """
    ...

resolve() 

resolve(identifier: IdentifierT) -> Any

Resolve an object from an identifier.

Parameters:

  • identifier (IdentifierT) –

    Identifier to an object.

Returns:

  • Any –

    The resolved object.

Source code in taps/transformer/_protocol.py 
62
63
64
65
66
67
68
69
70
71
def resolve(self, identifier: IdentifierT) -> Any:
    """Resolve an object from an identifier.

    Args:
        identifier: Identifier to an object.

    Returns:
        The resolved object.
    """
    ...

TransformerConfig

Bases: BaseModel, ABC

Abstract Transformer plugin configuration.

get_transformer() abstractmethod 

get_transformer() -> Transformer[Any]

Create a transformer from the configuration.

Source code in taps/transformer/_protocol.py 
85
86
87
88
@abc.abstractmethod
def get_transformer(self) -> Transformer[Any]:
    """Create a transformer from the configuration."""
    ...

NullTransformer

No-op transformer.

Rather than transforming an object, this transformer just returns the input object as its own identifier.

close() 

close() -> None

Close the transformer.

Source code in taps/transformer/_null.py 
37
38
39
def close(self) -> None:
    """Close the transformer."""
    pass

is_identifier() 

is_identifier(obj: Any) -> bool

Check if the object is an identifier instance.

Always False in this implementation.

Source code in taps/transformer/_null.py 
41
42
43
44
45
46
def is_identifier(self, obj: Any) -> bool:
    """Check if the object is an identifier instance.

    Always `False` in this implementation.
    """
    return False

transform() 

transform(obj: T) -> T

Transform the object into an identifier.

Parameters:

  • obj (T) –

    Object to transform.

Returns:

  • T –

    Identifier object that can be usd to resolve obj.

Source code in taps/transformer/_null.py 
48
49
50
51
52
53
54
55
56
57
def transform(self, obj: T) -> T:
    """Transform the object into an identifier.

    Args:
        obj: Object to transform.

    Returns:
        Identifier object that can be usd to resolve `obj`.
    """
    return obj

resolve() 

resolve(identifier: Any) -> NoReturn

Resolve an object from an identifier.

Parameters:

  • identifier (Any) –

    Identifier to an object.

Returns:

Source code in taps/transformer/_null.py 
59
60
61
62
63
64
65
66
67
68
69
70
def resolve(self, identifier: Any) -> NoReturn:
    """Resolve an object from an identifier.

    Args:
        identifier: Identifier to an object.

    Returns:
        The resolved object.
    """
    raise NotImplementedError(
        f'{self.__class__.__name__} does not support identifiers',
    )

NullTransformerConfig

Bases: TransformerConfig

NullTransformer plugin configuration.

get_transformer() 

get_transformer() -> NullTransformer

Create a transformer from the configuration.

Source code in taps/transformer/_null.py 
25
26
27
def get_transformer(self) -> NullTransformer:
    """Create a transformer from the configuration."""
    return NullTransformer()

PickleFileTransformer 

PickleFileTransformer(cache_dir: Path | str)

Pickle file object transformer.

Transforms objects by pickling the object and writing the pickled object to a file.

Parameters:

  • cache_dir (Path | str) –

    Directory to store pickled objects in.

Source code in taps/transformer/_file.py 
71
72
73
74
75
def __init__(
    self,
    cache_dir: pathlib.Path | str,
) -> None:
    self.cache_dir = pathlib.Path(cache_dir).resolve()

close() 

close() -> None

Close the transformer.

Source code in taps/transformer/_file.py 
77
78
79
def close(self) -> None:
    """Close the transformer."""
    shutil.rmtree(self.cache_dir, ignore_errors=True)

is_identifier() 

is_identifier(obj: Any) -> bool

Check if the object is an identifier instance.

Source code in taps/transformer/_file.py 
81
82
83
def is_identifier(self, obj: Any) -> bool:
    """Check if the object is an identifier instance."""
    return isinstance(obj, PickleFileIdentifier)

transform() 

transform(obj: T) -> PickleFileIdentifier

Transform the object into an identifier.

Parameters:

  • obj (T) –

    Object to transform.

Returns:

Source code in taps/transformer/_file.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def transform(self, obj: T) -> PickleFileIdentifier:
    """Transform the object into an identifier.

    Args:
        obj: Object to transform.

    Returns:
        Identifier object that can be used to resolve `obj`.
    """
    identifier = PickleFileIdentifier(self.cache_dir, uuid.uuid4())
    filepath = identifier.path()
    filepath.parent.mkdir(parents=True, exist_ok=True)

    with open(filepath, 'wb', buffering=0) as f:
        pickle.dump(obj, f)

    return identifier

resolve() 

resolve(identifier: PickleFileIdentifier) -> Any

Resolve an object from an identifier.

Parameters:

Returns:

  • Any –

    The resolved object.

Source code in taps/transformer/_file.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
def resolve(self, identifier: PickleFileIdentifier) -> Any:
    """Resolve an object from an identifier.

    Args:
        identifier: Identifier to an object.

    Returns:
        The resolved object.
    """
    filepath = identifier.path()
    with open(filepath, 'rb') as f:
        obj = pickle.load(f)
    return obj

PickleFileTransformerConfig

Bases: TransformerConfig

PickleFileTransformer plugin configuration.

Attributes:

  • file_dir (str) –

    Object file directory.

get_transformer() 

get_transformer() -> PickleFileTransformer

Create a transformer from the configuration.

Source code in taps/transformer/_file.py 
35
36
37
def get_transformer(self) -> PickleFileTransformer:
    """Create a transformer from the configuration."""
    return PickleFileTransformer(self.file_dir)

PickleFileIdentifier

Bases: NamedTuple

Identifier type for the PickleFileTransformer.

Attributes:

  • cache_dir (Path) –

    Object directory.

  • obj_id (UUID) –

    Object ID.

path() 

path() -> Path

Get path to the object.

Source code in taps/transformer/_file.py 
56
57
58
def path(self) -> pathlib.Path:
    """Get path to the object."""
    return self.cache_dir / str(self.obj_id)

ProxyTransformer 

ProxyTransformer(
    store: Store[Any], *, extract_target: bool = False
)

Proxy object transformer.

Transforms objects into proxies which act as the identifier.

Parameters:

  • store (Store[Any]) –

    Store instance to use for proxying objects.

  • extract_target (bool, default: False ) –

    When True, resolving an identifier (i.e., a proxy) will return the target object. Otherwise, the proxy is returned since a proxy can act as the target object.

Source code in taps/transformer/_proxy.py 
76
77
78
79
80
81
82
83
def __init__(
    self,
    store: Store[Any],
    *,
    extract_target: bool = False,
) -> None:
    self.store = store
    self.extract_target = extract_target

close() 

close() -> None

Close the transformer.

Source code in taps/transformer/_proxy.py 
 99
100
101
def close(self) -> None:
    """Close the transformer."""
    self.store.close()

is_identifier() 

is_identifier(obj: Any) -> bool

Check if the object is an identifier instance.

Source code in taps/transformer/_proxy.py 
103
104
105
def is_identifier(self, obj: Any) -> bool:
    """Check if the object is an identifier instance."""
    return isinstance(obj, Proxy)

transform() 

transform(obj: T) -> Proxy[T]

Transform the object into an identifier.

Parameters:

  • obj (T) –

    Object to transform.

Returns:

  • Proxy[T] –

    Identifier object that can be used to resolve obj.

Source code in taps/transformer/_proxy.py 
107
108
109
110
111
112
113
114
115
116
def transform(self, obj: T) -> Proxy[T]:
    """Transform the object into an identifier.

    Args:
        obj: Object to transform.

    Returns:
        Identifier object that can be used to resolve `obj`.
    """
    return self.store.proxy(obj)

resolve() 

resolve(identifier: Proxy[T]) -> T | Proxy[T]

Resolve an object from an identifier.

Parameters:

  • identifier (Proxy[T]) –

    Identifier to an object.

Returns:

  • T | Proxy[T] –

    The resolved object or a proxy of the resolved object depending on the setting of extract_target.

Source code in taps/transformer/_proxy.py 
118
119
120
121
122
123
124
125
126
127
128
def resolve(self, identifier: Proxy[T]) -> T | Proxy[T]:
    """Resolve an object from an identifier.

    Args:
        identifier: Identifier to an object.

    Returns:
        The resolved object or a proxy of the resolved object depending \
        on the setting of `extract_target`.
    """
    return extract(identifier) if self.extract_target else identifier

ProxyTransformerConfig

Bases: TransformerConfig

ProxyTransformer plugin configuration.

Attributes:

get_transformer() 

get_transformer() -> ProxyTransformer

Create a transformer from the configuration.

Source code in taps/transformer/_proxy.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
def get_transformer(self) -> ProxyTransformer:
    """Create a transformer from the configuration."""
    connector = self.connector.get_connector()
    return ProxyTransformer(
        store=Store(
            'proxy-transformer',
            connector=connector,
            cache_size=self.cache_size,
            populate_target=self.populate_target,
            register=True,
        ),
        extract_target=self.extract_target,
    )