taps.data.config

taps/data/config.py

FilterConfig

Bases: Config

Filter configuration.

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_filter()

get_filter() -> Filter

Get a filter according to the config.

Source code in taps/data/config.py

def get_filter(self) -> Filter:
    """Get a filter according to the config."""
    if self.filter_type is None:
        return NullFilter()
    elif self.filter_type == 'object-size':
        return ObjectSizeFilter(
            min_bytes=self.filter_min_size,
            max_bytes=self.filter_max_size,
        )
    elif self.filter_type == 'pickle-size':
        return PickleSizeFilter(
            min_bytes=self.filter_min_size,
            max_bytes=self.filter_max_size,
        )
    else:
        raise AssertionError(f'Unknown filter type {self.filter_type}.')

DataTransformerChoicesConfig

Bases: Config

Data transformer choice configuration.

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/data/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.
    """
    configs = get_registered()

    group = parser.add_argument_group(cls.__name__)
    group.add_argument(
        '--transformer',
        choices=sorted(configs.keys()),
        default='null',
        help='executor to use',
    )

    executor_type: str | None = None
    if argv is not None and '--executor' in argv:  # pragma: no cover
        executor_type = argv[argv.index('--executor') + 1]

    for name, config_type in configs.items():
        config_type.add_argument_group(
            parser,
            argv=argv,
            required=name == executor_type,
        )