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: field_name -> list of pre-generated values
    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)

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

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,
) -> 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.

    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)

    # Pin the clock when seeded so timestamps are reproducible
    if 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)

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)

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

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,
) -> 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.

    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)

    # Pin the clock when seeded so timestamps are reproducible
    if 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: field_name -> list of pre-generated values
    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)

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.