taps.data.proxy

taps/data/proxy.py

ProxyFileTransformerConfig

Bases: DataTransformerConfig

Proxy file transformer config.

add_argument_group() classmethod

add_argument_group(
    parser: ArgumentParser,
    *,
    argv: Sequence[str] | None = None,
    required: bool = True
) -> None

Add model fields as arguments of an argument group on the parser.

Parameters:

• parser (ArgumentParser) –

  Parser to add a new argument group to.

• argv (Sequence[str] | None, default: None ) –

  Optional sequence of string arguments.

• required (bool, default: True ) –

  Mark arguments without defaults as required.

Source code in taps/config.py

@classmethod
def add_argument_group(
    cls,
    parser: argparse.ArgumentParser,
    *,
    argv: Sequence[str] | None = None,
    required: bool = True,
) -> None:
    """Add model fields as arguments of an argument group on the parser.

    Args:
        parser: Parser to add a new argument group to.
        argv: Optional sequence of string arguments.
        required: Mark arguments without defaults as required.
    """
    group = parser.add_argument_group(cls.__name__)
    for field_name, field_info in cls.model_fields.items():
        arg_name = field_name.replace('_', '-').lower()
        group.add_argument(
            f'--{arg_name}',
            dest=field_name,
            # type=field_info.annotation,
            default=field_info.get_default(),
            required=field_info.is_required() and required,
            help=field_info.description,
        )

get_transformer()

get_transformer() -> ProxyTransformer

Create a transformer instance from the config.

Source code in taps/data/proxy.py

def get_transformer(self) -> ProxyTransformer:
    """Create a transformer instance from the config."""
    connector: Connector[Any]
    if self.ps_type == 'file':
        if self.ps_file_dir is None:  # pragma: no cover
            raise ValueError(
                'Option --ps-file-dir is required when --ps-type file.',
            )
        connector = FileConnector(self.ps_file_dir)
    elif self.ps_type == 'redis':
        if self.ps_redis_addr is None:
            raise ValueError(  # pragma: no cover
                'Option --ps-redis-addr is required when --ps-type redis.',
            )
        parts = self.ps_redis_addr.split(':')
        host, port = parts[0], int(parts[1])
        connector = RedisConnector(host, port)
    else:
        raise AssertionError(
            f'Unknown proxy transformer type: {self.ps_type}.',
        )

    return ProxyTransformer(
        store=Store(
            'transformer',
            connector=connector,
            register=True,
            populate_target=True,
        ),
        extract_target=self.ps_extract_target,
    )

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/data/proxy.py

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/data/proxy.py

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/data/proxy.py

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/data/proxy.py

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/data/proxy.py

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