config

config

Classes:

Name	Description
MeasurementConfig	Configuration for writing structured measurement records around a run.

MeasurementConfig(output_path, output_format='jsonl', record_level=True, streaming=False, keep_records=True, dd_trace='none', dd_trace_path=None, dd_task_trace_path=None, run_id=None, record_hash_key=None, run_tags=None, fail_on_write_error=False) dataclass

Configuration for writing structured measurement records around a run.

Methods:

Name	Description
from_env	Build measurement config from environment variables, or None if output is unset.
from_sources	Resolve measurement config from explicit config first, then optional env.
write_collector	Write a collector using this config's output format.

from_env(*, prefix=DEFAULT_MEASUREMENT_ENV_PREFIX) classmethod

Build measurement config from environment variables, or None if output is unset.

This is intentionally opt-in. Anonymizer API and CLI calls do not read measurement environment variables unless benchmark/tooling code calls this helper explicitly.

Source code in src/anonymizer/measurement/config.py

@classmethod
def from_env(cls, *, prefix: str = DEFAULT_MEASUREMENT_ENV_PREFIX) -> MeasurementConfig | None:
    """Build measurement config from environment variables, or None if output is unset.

    This is intentionally opt-in. Anonymizer API and CLI calls do not read
    measurement environment variables unless benchmark/tooling code calls this
    helper explicitly.
    """
    try:
        settings = _load_measurement_env_settings(prefix=prefix)
    except (SettingsError, ValidationError) as exc:
        raise ValueError(_measurement_env_error_message(exc, prefix=prefix)) from None

    if settings.output_path is None:
        return None
    return cls(
        output_path=settings.output_path,
        output_format=settings.output_format,
        record_level=settings.record_level,
        streaming=settings.streaming,
        keep_records=settings.keep_records,
        dd_trace=settings.dd_trace,
        dd_trace_path=settings.dd_trace_path,
        dd_task_trace_path=settings.dd_task_trace_path,
        run_id=settings.run_id,
        run_tags=settings.run_tags,
        fail_on_write_error=settings.fail_on_write_error,
    )

from_sources(explicit=None, *, env=False, prefix=DEFAULT_MEASUREMENT_ENV_PREFIX) classmethod

Resolve measurement config from explicit config first, then optional env.

Source code in src/anonymizer/measurement/config.py

@classmethod
def from_sources(
    cls,
    explicit: MeasurementConfig | None = None,
    *,
    env: bool = False,
    prefix: str = DEFAULT_MEASUREMENT_ENV_PREFIX,
) -> MeasurementConfig | None:
    """Resolve measurement config from explicit config first, then optional env."""
    if explicit is not None:
        return explicit
    if env:
        return cls.from_env(prefix=prefix)
    return None

write_collector(collector)

Write a collector using this config's output format.

Source code in src/anonymizer/measurement/config.py

def write_collector(self, collector: MeasurementCollector) -> None:
    """Write a collector using this config's output format."""
    _writer_for_format(self.output_format).write(collector.records, self.output_path)