main

main

Classes:

Name	Description
CliOpts	CLI options for anonymizer commands.

Functions:

Name	Description
run	Run the full anonymization pipeline (detection + replacement or rewrite).
preview	Run the pipeline on a subset of records for quick inspection.
validate	Validate that the active config is compatible with model selections.
main	Entry point for the anonymizer CLI.

CliOpts(replace=None, rewrite=False, detect=Detect(), model_configs=None, model_providers=None, artifact_path=None, verbose=False, debug=False, format_template=None, normalize_label=True, algorithm='sha256', digest_length=12, protect=None, preserve=None, risk_tolerance='low', max_repair_iterations=2, instructions=None, _REPLACE_ONLY_FLAGS=('format_template',), _REWRITE_ONLY_FLAGS=('protect', 'preserve')) dataclass

CLI options for anonymizer commands.

Exactly one of --replace or --rewrite must be provided.

Replace-specific flags (format_template, normalize_label, algorithm, digest_length) only apply when --replace is set.

Rewrite-specific flags (protect, preserve, risk_tolerance, max_repair_iterations) only apply when --rewrite is set.

run(*, data, opts=CliOpts(), output=None)

Run the full anonymization pipeline (detection + replacement or rewrite).

Source code in src/anonymizer/interface/cli/main.py

@app.command
@_cli_error_handler
def run(
    *,
    data: Annotated[AnonymizerInput, cyclopts.Parameter(name="*")],
    opts: Annotated[CliOpts, cyclopts.Parameter(name="*")] = CliOpts(),
    output: str | None = None,
) -> None:
    """Run the full anonymization pipeline (detection + replacement or rewrite)."""
    if output is None:
        source = Path(data.source)
        suffix = "_rewritten" if opts.rewrite else "_anonymized"
        output = str(source.parent / f"{source.stem}{suffix}{source.suffix}")
    output_path = Path(output).resolve()
    if output_path.suffix.lower() not in SUPPORTED_IO_FORMATS:
        raise InvalidConfigError(
            f"Unsupported output format: {output_path.suffix!r}. Use one of {SUPPORTED_IO_FORMATS}"
        )
    if output_path == Path(data.source).resolve():
        raise InvalidConfigError(f"Output path must differ from source: {output_path}")
    _configure_logging(opts)
    config, anonymizer = _build_config_and_anonymizer(opts)
    result = anonymizer.run(config=config, data=data)
    written = write_result(result, output)
    print(f"Output written to: {written}")

preview(*, data, opts=CliOpts(), num_records=10)

Run the pipeline on a subset of records for quick inspection.

Source code in src/anonymizer/interface/cli/main.py

@app.command
@_cli_error_handler
def preview(
    *,
    data: Annotated[AnonymizerInput, cyclopts.Parameter(name="*")],
    opts: Annotated[CliOpts, cyclopts.Parameter(name="*")] = CliOpts(),
    num_records: int = 10,
) -> None:
    """Run the pipeline on a subset of records for quick inspection."""
    _configure_logging(opts)
    config, anonymizer = _build_config_and_anonymizer(opts)
    result = anonymizer.preview(config=config, data=data, num_records=num_records)
    print(result.dataframe.to_string(max_colwidth=80))

validate(*, data, opts=CliOpts())

Validate that the active config is compatible with model selections.

Source code in src/anonymizer/interface/cli/main.py

@app.command
@_cli_error_handler
def validate(
    *,
    data: Annotated[AnonymizerInput, cyclopts.Parameter(name="*")],
    opts: Annotated[CliOpts, cyclopts.Parameter(name="*")] = CliOpts(),
) -> None:
    """Validate that the active config is compatible with model selections."""
    _configure_logging(opts)
    config, anonymizer = _build_config_and_anonymizer(opts)
    anonymizer.validate_config(config)
    print("Config is valid.")

main()

Entry point for the anonymizer CLI.

Source code in src/anonymizer/interface/cli/main.py

def main() -> None:
    """Entry point for the anonymizer CLI."""
    app()