API Reference¶

Auto-generated from source docstrings.

avro_datagen¶

`avro_datagen` ¶

Schema-driven data generator for Avro schemas with arg.properties hints.

`RecordResolver(schema, seed=None)` ¶

Resolves a single Avro record schema into generated dicts.

Maintains pools (for pool hints) across multiple generate() calls so that pooled values are reused across records.

Source code in src/avro_datagen/resolver.py

def __init__(self, schema: dict[str, Any], seed: int | None = None):
    self.schema = schema
    self.fields: list[dict[str, Any]] = schema["fields"]
    self.seed = seed
    # Pools: cache key -> list of pre-generated values. Process-RNG pools
    # are keyed by field name; deterministic pools by their seed basis
    # (see _resolve_pool).
    self.pools: dict[str, list[Any]] = {}
    # Capture "now" once so timestamps are reproducible with a seed
    self.now_ts: float = datetime.now(UTC).timestamp()
    # Named record types encountered during resolution (for recursive records)
    self.named_types: dict[str, dict[str, Any]] = {}
    self._indexnamed_types(schema)
    # Cache for locale-specific Faker instances (seeded consistently)
    self._locale_fakers: dict[str, Faker] = {}
    # Cache for foreign key source files: (path, field) -> list of values
    self._fk_cache: dict[tuple[str, str], list[Any]] = {}

`generate()` ¶

Generate one record. Fields resolved top-to-bottom.

Source code in src/avro_datagen/resolver.py

def generate(self) -> dict:
    """Generate one record. Fields resolved top-to-bottom."""
    record: dict[str, Any] = {}
    for field in self.fields:
        name = field["name"]
        record[name] = self._resolve_field(field, record)
    return record

`SchemaValidationError(errors)` ¶

Bases: ValueError

Raised when a schema fails validation.

Contains a list of error messages, one per issue found. The first message is also used as the exception's main message.

Source code in src/avro_datagen/validator.py

def __init__(self, errors: list[str]):
    self.errors = errors
    message = errors[0] if errors else "Schema validation failed"
    if len(errors) > 1:
        message += f" ({len(errors) - 1} more issue(s))"
    super().__init__(message)

`generate(schema_path, count, seed=None, now=None)` ¶

Generate count fake records from an Avro schema file.

Parameters:

Name	Type	Description	Default
`schema_path`	`str \| Path`	Path to a .avsc file.	required
`count`	`int`	Number of records to generate. 0 means infinite.	required
`seed`	`int \| None`	Optional seed for reproducible output.	`None`
`now`	`datetime \| float \| None`	Optional clock anchor for relative bounds ("now", "-30d", "today", ...) and un-hinted timestamp/date fields. A timezone-aware datetime or epoch seconds. Defaults to 2026-01-01T00:00:00Z when seeded and the real clock otherwise, so reproducible output is a function of the pair (seed, now).	`None`

Yields:

Type	Description
`dict`	dict — one record per iteration.

Source code in src/avro_datagen/generator.py

def generate(
    schema_path: str | Path,
    count: int,
    seed: int | None = None,
    now: datetime | float | None = None,
) -> Iterator[dict]:
    """Generate `count` fake records from an Avro schema file.

    Args:
        schema_path: Path to a .avsc file.
        count: Number of records to generate. 0 means infinite.
        seed: Optional seed for reproducible output.
        now: Optional clock anchor for relative bounds ("now", "-30d",
            "today", ...) and un-hinted timestamp/date fields. A timezone-aware
            datetime or epoch seconds. Defaults to 2026-01-01T00:00:00Z when
            seeded and the real clock otherwise, so reproducible output is a
            function of the pair (seed, now).

    Yields:
        dict — one record per iteration.
    """
    if seed is not None:
        random.seed(seed)
        _faker.seed_instance(seed)

    schema = load_schema(schema_path)
    resolver = RecordResolver(schema, seed=seed)

    # An explicit anchor wins; otherwise pin the clock when seeded so
    # timestamps are reproducible
    if now is not None:
        resolver.now_ts = _to_epoch(now)
    elif seed is not None:
        resolver.now_ts = _FIXED_EPOCH

    if count == 0:
        while True:
            yield resolver.generate()
    else:
        for _ in range(count):
            yield resolver.generate()

`load_schema(path)` ¶

Load an .avsc file and return the parsed JSON.

Source code in src/avro_datagen/resolver.py

def load_schema(path: str | Path) -> dict:
    """Load an .avsc file and return the parsed JSON."""
    with open(path, encoding="utf-8") as f:
        return json.load(f)

`to_human_json(record, schema)` ¶

Return a copy of record with temporal logical types in human form.

Walks the record alongside its schema, converting timestamp-millis, timestamp-micros, date, time-millis and time-micros to ISO-8601 strings. Every other value is passed through unchanged.

Parameters:

Name	Type	Description	Default
`record`	`dict`	A generated record (wire form).	required
`schema`	`dict`	The Avro record schema it was generated from.	required

Returns:

Type	Description
`dict`	A new dict — `record` is not mutated.

Source code in src/avro_datagen/json_format.py

def to_human_json(record: dict, schema: dict) -> dict:
    """Return a copy of `record` with temporal logical types in human form.

    Walks the record alongside its schema, converting `timestamp-millis`,
    `timestamp-micros`, `date`, `time-millis` and `time-micros` to ISO-8601
    strings. Every other value is passed through unchanged.

    Args:
        record: A generated record (wire form).
        schema: The Avro record schema it was generated from.

    Returns:
        A new dict — `record` is not mutated.
    """
    if not isinstance(schema, dict) or schema.get("type") != "record":
        return record
    return _convert_record(record, schema)

`validate(schema)` ¶

Validate an Avro schema and return a list of warnings.

Errors (structural problems) raise SchemaValidationError. Warnings (unknown hint keys, etc.) are returned as a list.

Parameters:

Name	Type	Description	Default
`schema`	`dict \| str \| Path`	Parsed schema dict or path to an .avsc file.	required

Returns:

Type	Description
`list[str]`	List of warning strings (empty if the schema is clean).

Raises:

Type	Description
`SchemaValidationError`	if the schema has structural errors.

Source code in src/avro_datagen/validator.py

def validate(schema: dict | str | Path) -> list[str]:
    """Validate an Avro schema and return a list of warnings.

    Errors (structural problems) raise SchemaValidationError.  Warnings
    (unknown hint keys, etc.) are returned as a list.

    Args:
        schema: Parsed schema dict or path to an .avsc file.

    Returns:
        List of warning strings (empty if the schema is clean).

    Raises:
        SchemaValidationError: if the schema has structural errors.
    """
    if isinstance(schema, (str, Path)):
        schema = load_schema(schema)

    errors: list[str] = []
    warnings: list[str] = []
    _validate_record(schema, path="", errors=errors, warnings=warnings)

    if errors:
        raise SchemaValidationError(errors)
    return warnings

generator¶

`avro_datagen.generator` ¶

Core generator — loads an Avro schema and yields fake records.

`generate(schema_path, count, seed=None, now=None)` ¶

Generate count fake records from an Avro schema file.

Parameters:

Name	Type	Description	Default
`schema_path`	`str \| Path`	Path to a .avsc file.	required
`count`	`int`	Number of records to generate. 0 means infinite.	required
`seed`	`int \| None`	Optional seed for reproducible output.	`None`
`now`	`datetime \| float \| None`	Optional clock anchor for relative bounds ("now", "-30d", "today", ...) and un-hinted timestamp/date fields. A timezone-aware datetime or epoch seconds. Defaults to 2026-01-01T00:00:00Z when seeded and the real clock otherwise, so reproducible output is a function of the pair (seed, now).	`None`

Yields:

Type	Description
`dict`	dict — one record per iteration.

Source code in src/avro_datagen/generator.py

def generate(
    schema_path: str | Path,
    count: int,
    seed: int | None = None,
    now: datetime | float | None = None,
) -> Iterator[dict]:
    """Generate `count` fake records from an Avro schema file.

    Args:
        schema_path: Path to a .avsc file.
        count: Number of records to generate. 0 means infinite.
        seed: Optional seed for reproducible output.
        now: Optional clock anchor for relative bounds ("now", "-30d",
            "today", ...) and un-hinted timestamp/date fields. A timezone-aware
            datetime or epoch seconds. Defaults to 2026-01-01T00:00:00Z when
            seeded and the real clock otherwise, so reproducible output is a
            function of the pair (seed, now).

    Yields:
        dict — one record per iteration.
    """
    if seed is not None:
        random.seed(seed)
        _faker.seed_instance(seed)

    schema = load_schema(schema_path)
    resolver = RecordResolver(schema, seed=seed)

    # An explicit anchor wins; otherwise pin the clock when seeded so
    # timestamps are reproducible
    if now is not None:
        resolver.now_ts = _to_epoch(now)
    elif seed is not None:
        resolver.now_ts = _FIXED_EPOCH

    if count == 0:
        while True:
            yield resolver.generate()
    else:
        for _ in range(count):
            yield resolver.generate()

resolver¶

`avro_datagen.resolver` ¶

Field resolver — maps Avro types + arg.properties to generated values.

This is the core engine. It walks an Avro schema top-to-bottom, resolving each field's value based on its type, logical type, and arg.properties hints. Fields are resolved in declaration order so that later fields can reference earlier ones via ref and rules.

`RecordResolver(schema, seed=None)` ¶

Resolves a single Avro record schema into generated dicts.

Maintains pools (for pool hints) across multiple generate() calls so that pooled values are reused across records.

Source code in src/avro_datagen/resolver.py

def __init__(self, schema: dict[str, Any], seed: int | None = None):
    self.schema = schema
    self.fields: list[dict[str, Any]] = schema["fields"]
    self.seed = seed
    # Pools: cache key -> list of pre-generated values. Process-RNG pools
    # are keyed by field name; deterministic pools by their seed basis
    # (see _resolve_pool).
    self.pools: dict[str, list[Any]] = {}
    # Capture "now" once so timestamps are reproducible with a seed
    self.now_ts: float = datetime.now(UTC).timestamp()
    # Named record types encountered during resolution (for recursive records)
    self.named_types: dict[str, dict[str, Any]] = {}
    self._indexnamed_types(schema)
    # Cache for locale-specific Faker instances (seeded consistently)
    self._locale_fakers: dict[str, Faker] = {}
    # Cache for foreign key source files: (path, field) -> list of values
    self._fk_cache: dict[tuple[str, str], list[Any]] = {}

`generate()` ¶

Generate one record. Fields resolved top-to-bottom.

Source code in src/avro_datagen/resolver.py

def generate(self) -> dict:
    """Generate one record. Fields resolved top-to-bottom."""
    record: dict[str, Any] = {}
    for field in self.fields:
        name = field["name"]
        record[name] = self._resolve_field(field, record)
    return record

`load_schema(path)` ¶

Load an .avsc file and return the parsed JSON.

Source code in src/avro_datagen/resolver.py

def load_schema(path: str | Path) -> dict:
    """Load an .avsc file and return the parsed JSON."""
    with open(path, encoding="utf-8") as f:
        return json.load(f)

json_format¶

`avro_datagen.json_format` ¶

Post-processing: render wire-form records as human-readable JSON.

The generator emits the Avro wire representation of logical types — a timestamp-millis is an int of epoch milliseconds. That is what a schema-registry serde expects to receive after it encodes, but Kafka UIs with a produce form (kafbat-ui, for one) want the human representation and do the conversion themselves. Pasting a generated sample into such a form therefore means hand-converting every timestamp.

to_human_json() is a pure post-processing walker over an already-generated record. It never touches the generator or the producer path, both of which stay wire-form.

This module deliberately imports nothing from the rest of the package — resolver imports iso_utc from here, not the other way round.

`iso_utc(dt, timespec='auto')` ¶

Format a UTC datetime as ISO-8601 with a Z suffix.

Source code in src/avro_datagen/json_format.py

def iso_utc(dt: datetime, timespec: str = "auto") -> str:
    """Format a UTC datetime as ISO-8601 with a `Z` suffix."""
    return dt.isoformat(timespec=timespec).replace("+00:00", "Z")

`to_human_json(record, schema)` ¶

Return a copy of record with temporal logical types in human form.

Walks the record alongside its schema, converting timestamp-millis, timestamp-micros, date, time-millis and time-micros to ISO-8601 strings. Every other value is passed through unchanged.

Parameters:

Name	Type	Description	Default
`record`	`dict`	A generated record (wire form).	required
`schema`	`dict`	The Avro record schema it was generated from.	required

Returns:

Type	Description
`dict`	A new dict — `record` is not mutated.

Source code in src/avro_datagen/json_format.py

def to_human_json(record: dict, schema: dict) -> dict:
    """Return a copy of `record` with temporal logical types in human form.

    Walks the record alongside its schema, converting `timestamp-millis`,
    `timestamp-micros`, `date`, `time-millis` and `time-micros` to ISO-8601
    strings. Every other value is passed through unchanged.

    Args:
        record: A generated record (wire form).
        schema: The Avro record schema it was generated from.

    Returns:
        A new dict — `record` is not mutated.
    """
    if not isinstance(schema, dict) or schema.get("type") != "record":
        return record
    return _convert_record(record, schema)

validator¶

`avro_datagen.validator` ¶

Schema validator -- checks Avro schemas for structural issues and common mistakes.

Validates: - Required fields (name, type, fields for records) - Valid primitive and logical types - Logical type compatibility with base types (e.g. uuid must be on string) - Known arg.properties hint keys (warns on unknown keys) - Hints that are misplaced rather than misspelled: args/kwargs/locale without a sibling faker, and hints nested inside the type object - faker specs: the method exists, the spec shape is known - range bounds match the field's (logical) type - options entries can be encoded as the field's Avro type - Rule conditions reference declared fields - ref targets exist and appear before the referencing field - pool hints: scalar or {size, per, seed} object form, per targets appear before the pooled field, and pool/foreign_key mutual exclusivity

Validation is not mandatory -- generate() still works with any schema the resolver accepts. Use validate() before generation for clearer errors, or call it from the avro-datagen validate CLI subcommand.

`SchemaValidationError(errors)` ¶

Bases: ValueError

Raised when a schema fails validation.

Contains a list of error messages, one per issue found. The first message is also used as the exception's main message.

Source code in src/avro_datagen/validator.py

def __init__(self, errors: list[str]):
    self.errors = errors
    message = errors[0] if errors else "Schema validation failed"
    if len(errors) > 1:
        message += f" ({len(errors) - 1} more issue(s))"
    super().__init__(message)

`validate(schema)` ¶

Validate an Avro schema and return a list of warnings.

Errors (structural problems) raise SchemaValidationError. Warnings (unknown hint keys, etc.) are returned as a list.

Parameters:

Name	Type	Description	Default
`schema`	`dict \| str \| Path`	Parsed schema dict or path to an .avsc file.	required

Returns:

Type	Description
`list[str]`	List of warning strings (empty if the schema is clean).

Raises:

Type	Description
`SchemaValidationError`	if the schema has structural errors.

Source code in src/avro_datagen/validator.py

def validate(schema: dict | str | Path) -> list[str]:
    """Validate an Avro schema and return a list of warnings.

    Errors (structural problems) raise SchemaValidationError.  Warnings
    (unknown hint keys, etc.) are returned as a list.

    Args:
        schema: Parsed schema dict or path to an .avsc file.

    Returns:
        List of warning strings (empty if the schema is clean).

    Raises:
        SchemaValidationError: if the schema has structural errors.
    """
    if isinstance(schema, (str, Path)):
        schema = load_schema(schema)

    errors: list[str] = []
    warnings: list[str] = []
    _validate_record(schema, path="", errors=errors, warnings=warnings)

    if errors:
        raise SchemaValidationError(errors)
    return warnings

producer¶

`avro_datagen.producer` ¶

Kafka producer — generates records from an Avro schema and publishes to a topic.

`build_producer_config(bootstrap_servers, security_protocol='', sasl_mechanism='', sasl_username='', sasl_password='', acks='all', linger_ms=5, batch_size=16384, compression_type='none')` ¶

Build a confluent-kafka producer config dict from explicit params.

Source code in src/avro_datagen/producer.py

def build_producer_config(
    bootstrap_servers: str,
    security_protocol: str = "",
    sasl_mechanism: str = "",
    sasl_username: str = "",
    sasl_password: str = "",
    acks: str = "all",
    linger_ms: int = 5,
    batch_size: int = 16384,
    compression_type: str = "none",
) -> dict[str, str | int]:
    """Build a confluent-kafka producer config dict from explicit params."""
    config: dict[str, str | int] = {
        "bootstrap.servers": bootstrap_servers,
        "acks": acks,
        "linger.ms": linger_ms,
        "batch.size": batch_size,
        "compression.type": compression_type,
    }

    if security_protocol:
        config["security.protocol"] = security_protocol
    if sasl_mechanism:
        config["sasl.mechanism"] = sasl_mechanism
    if sasl_username:
        config["sasl.username"] = sasl_username
    if sasl_password:
        config["sasl.password"] = sasl_password

    return config

`produce(schema_path, topic, producer_config, count=10, rate=None, seed=None, key_fn=None, on_delivery=None, on_progress=None)` ¶

Generate records and publish them to a Kafka topic.

Parameters:

Name	Type	Description	Default
`schema_path`	`str \| Path`	Path to the .avsc schema file.	required
`topic`	`str`	Kafka topic to publish to.	required
`producer_config`	`dict[str, str \| int]`	confluent-kafka Producer config dict.	required
`count`	`int`	Number of records (0 = infinite).	`10`
`rate`	`float \| None`	Records per second limit. None = unlimited.	`None`
`seed`	`int \| None`	Random seed for reproducibility.	`None`
`key_fn`	`Callable[[dict[str, Any]], str \| None] \| None`	Callable to extract a message key from the record.	`None`
`on_delivery`	`Callable[[Any, Any], None] \| None`	Per-message delivery callback (err, msg).	`None`
`on_progress`	`Callable[[int, dict[str, Any]], None] \| None`	Called after each record with (index, record).	`None`

Returns:

Type	Description
`dict[str, int \| float]`	dict with produced, errors, elapsed_s keys.

Source code in src/avro_datagen/producer.py

def produce(
    schema_path: str | Path,
    topic: str,
    producer_config: dict[str, str | int],
    count: int = 10,
    rate: float | None = None,
    seed: int | None = None,
    key_fn: Callable[[dict[str, Any]], str | None] | None = None,
    on_delivery: Callable[[Any, Any], None] | None = None,
    on_progress: Callable[[int, dict[str, Any]], None] | None = None,
) -> dict[str, int | float]:
    """Generate records and publish them to a Kafka topic.

    Args:
        schema_path: Path to the .avsc schema file.
        topic: Kafka topic to publish to.
        producer_config: confluent-kafka Producer config dict.
        count: Number of records (0 = infinite).
        rate: Records per second limit. None = unlimited.
        seed: Random seed for reproducibility.
        key_fn: Callable to extract a message key from the record.
        on_delivery: Per-message delivery callback (err, msg).
        on_progress: Called after each record with (index, record).

    Returns:
        dict with produced, errors, elapsed_s keys.
    """
    if key_fn is None:
        key_fn = _default_key

    producer = Producer(producer_config)
    interval = 1.0 / rate if rate else 0.0

    produced = 0
    errors = 0
    start_time = time.monotonic()

    def _delivery_cb(err, msg):
        nonlocal errors
        if err is not None:
            errors += 1
        if on_delivery:
            on_delivery(err, msg)

    try:
        for i, record in enumerate(generate(schema_path, count, seed)):
            tick = time.monotonic()

            value = json.dumps(record).encode("utf-8")
            key = key_fn(record)
            key_bytes = key.encode("utf-8") if key else None

            try:
                producer.produce(
                    topic=topic,
                    value=value,
                    key=key_bytes,
                    callback=_delivery_cb,
                )
                produced += 1
            except BufferError:
                producer.flush(timeout=5)
                producer.produce(
                    topic=topic,
                    value=value,
                    key=key_bytes,
                    callback=_delivery_cb,
                )
                produced += 1
            except KafkaException:
                errors += 1

            producer.poll(0)

            if on_progress:
                on_progress(i, record)

            if interval:
                elapsed = time.monotonic() - tick
                sleep_time = interval - elapsed
                if sleep_time > 0:
                    time.sleep(sleep_time)

    except KeyboardInterrupt:
        pass
    finally:
        producer.flush(timeout=30)

    return {
        "produced": produced,
        "errors": errors,
        "elapsed_s": round(time.monotonic() - start_time, 2),
    }

cli¶

`avro_datagen.cli` ¶

CLI entry point — generate fake data to stdout as JSON lines.

API Reference¶

avro_datagen¶

avro_datagen ¶

RecordResolver(schema, seed=None) ¶

generate() ¶

SchemaValidationError(errors) ¶

generate(schema_path, count, seed=None, now=None) ¶

load_schema(path) ¶

to_human_json(record, schema) ¶

validate(schema) ¶

generator¶

avro_datagen.generator ¶

generate(schema_path, count, seed=None, now=None) ¶

resolver¶

avro_datagen.resolver ¶

RecordResolver(schema, seed=None) ¶

generate() ¶

load_schema(path) ¶

json_format¶

avro_datagen.json_format ¶

iso_utc(dt, timespec='auto') ¶

to_human_json(record, schema) ¶

validator¶

avro_datagen.validator ¶

SchemaValidationError(errors) ¶

validate(schema) ¶

producer¶

avro_datagen.producer ¶

build_producer_config(bootstrap_servers, security_protocol='', sasl_mechanism='', sasl_username='', sasl_password='', acks='all', linger_ms=5, batch_size=16384, compression_type='none') ¶

produce(schema_path, topic, producer_config, count=10, rate=None, seed=None, key_fn=None, on_delivery=None, on_progress=None) ¶

cli¶

avro_datagen.cli ¶

`avro_datagen` ¶

`RecordResolver(schema, seed=None)` ¶

`generate()` ¶

`SchemaValidationError(errors)` ¶

`generate(schema_path, count, seed=None, now=None)` ¶

`load_schema(path)` ¶

`to_human_json(record, schema)` ¶

`validate(schema)` ¶

`avro_datagen.generator` ¶

`generate(schema_path, count, seed=None, now=None)` ¶

`avro_datagen.resolver` ¶

`RecordResolver(schema, seed=None)` ¶

`generate()` ¶

`load_schema(path)` ¶

`avro_datagen.json_format` ¶

`iso_utc(dt, timespec='auto')` ¶

`to_human_json(record, schema)` ¶

`avro_datagen.validator` ¶

`SchemaValidationError(errors)` ¶

`validate(schema)` ¶

`avro_datagen.producer` ¶

`build_producer_config(bootstrap_servers, security_protocol='', sasl_mechanism='', sasl_username='', sasl_password='', acks='all', linger_ms=5, batch_size=16384, compression_type='none')` ¶

`produce(schema_path, topic, producer_config, count=10, rate=None, seed=None, key_fn=None, on_delivery=None, on_progress=None)` ¶

`avro_datagen.cli` ¶