API Reference¶

def add_validation_errors(
    *,
    record: DataRecordDict,
    errors: ErrorMessages,
) -> DataRecordDict:
    """Return a copy of a record with validation errors attached.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        record: A dictionary representing one data record.
        errors: A list of validation error messages.

    Returns:
        A copy of the record with a validation_errors field appended.
    """
    output = dict(record)
    output["validation_errors"] = " | ".join(errors)
    return output

def validate_boolean_text(value: str, *, field_name: str) -> list[str]:
    """Return errors for an invalid boolean text value.

    All boolean values must be represented as
    "true" or "false" (case-insensitive).

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        value: The text value to validate.
        field_name: The name of the field being validated, for error messages.

    Returns:
        A list of errors, or an empty list if the value is valid.
    """
    allowed_values = {"true", "false"}

    if value.lower() not in allowed_values:
        return [f"{field_name} must be true or false: {value}"]

    return []

def validate_datetime(value: str) -> list[str]:
    """Return errors for an invalid datetime value.

    All datetime values must be in ISO 8601 format.

    Arguments:
        value: The text value to validate.

    Returns:
        A list of errors, or an empty list if the value is valid.
    """
    try:
        datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return [f"Invalid datetime: {value}"]

    return []

def validate_positive_integer(value: str) -> list[str]:
    """Return errors for an invalid positive integer value.

    All positive integer values must be integers greater than or equal to 1.

    Arguments:
        value: The text value to validate.

    Returns:
        A list of errors, or an empty list if the value is valid.
    """
    try:
        number = int(value)
    except ValueError:
        return [f"Value must be an integer: {value}"]

    if number < 1:
        return [f"Value must be at least 1: {value}"]

    return []

def validate_required_fields(
    *,
    record: DataRecordDict,
    required_fields: list[str],
) -> list[str]:
    """Return errors for missing or blank required fields.

    All required fields must be present and not blank.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        record: A dictionary representing one data record / row.
        required_fields: A list of field names that are required.

    Returns:
        A list of errors, or
        an empty list if all required fields are present.
    """
    errors: list[str] = []

    for field_name in required_fields:
        if field_name not in record:
            errors.append(f"Missing required field: {field_name}")
        elif not record[field_name].strip():
            errors.append(f"Required field is blank: {field_name}")

    return errors

def make_lookup_set(records: DataRecordDictList, key_field: str) -> AllowedValuesSet:
    """Create a set of allowed values for a field in a reference table.

    Arguments:
        records: A list of row dictionaries from a reference CSV file.
        key_field: The field to use as the key for allowed values.

    Returns:
        A set of allowed values for the specified key field.
    """
    values: AllowedValuesSet = set()
    for record in records:
        value: str = record.get(key_field, "").strip()
        if value:
            values.add(value)
    return values

def validate_reference_records(
    *,
    records: DataRecordDictList,
    required_fields: list[str],
    label: str,
) -> list[str]:
    """Validate reference records and return file-level errors.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        records: Reference data records to validate.
        required_fields: Field names required in each record.
        label: Label for this reference file, used in error messages.

    Returns:
        A list of errors, or an empty list if all records are valid.
    """
    errors: list[str] = []
    for record_number, record in enumerate(records, start=1):
        for error in validate_required_fields(
            record=record, required_fields=required_fields
        ):
            errors.append(f"{label} record {record_number}: {error}")
    return errors

def reference_validation_failed_message(*, label: str, error_count: int) -> str:
    """Return help text when a reference data file fails validation."""
    return f"""
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
A reference data file failed validation.
File:   {label}
Errors: {error_count} problem(s) found.

The producer cannot run until all reference files are valid.
Fix the reference file before retrying.

CHECK:
1. Open data/{label} and inspect the header row.
2. Confirm all required fields are present and spelled correctly.
3. Confirm no rows have blank values in required fields.
4. See data_contract_case.py for the list of required fields.
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
""".strip()

def append_csv_row(path: Path, row: dict[str, Any], fieldnames: list[str]) -> None:
    """Append one row to a CSV file, writing the header first if needed."""
    path.parent.mkdir(parents=True, exist_ok=True)
    file_exists = path.exists()

    with path.open(mode="a", encoding="utf-8", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)

        if not file_exists:
            writer.writeheader()

        writer.writerow(row)

def format_message_for_log(message: dict[str, Any]) -> str:
    """Format one message dictionary for readable log output."""
    lines = ["{"]

    for key, value in message.items():
        lines.append(f"  {key}: {value}")

    lines.append("}")
    return "\n".join(lines)

def read_csv_as_lookup(
    path: Path,
    *,
    key_field: str,
    value_field: str,
) -> dict[str, Any]:
    """Read a CSV file into a key-value lookup dictionary.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        path:        Path to the CSV file.
        key_field:   The column to use as the dictionary key.
        value_field: The column to use as the dictionary value.

    Returns:
        A dict mapping each key_field value to its value_field value.

    Example:
        region_lookup = read_csv_as_lookup(
            REGIONS_CSV, key_field="region_id", value_field="tax_rate_pct"
        )
        tax_rate = float(region_lookup["US-MO"]) / 100.0
    """
    rows = read_csv_rows(path)
    return {row[key_field]: row[value_field] for row in rows}

def read_csv_rows(path: Path) -> list[dict[str, str]]:
    """Read a CSV file into a list of string dictionaries."""
    if not path.exists():
        msg = missing_csv_file_message(path=path.as_posix())
        raise FileNotFoundError(msg)

    with path.open(mode="r", encoding="utf-8", newline="") as file:
        reader = csv.DictReader(file)

        if reader.fieldnames is None:
            msg = f"CSV file has no header row: {path.as_posix()}"
            raise ValueError(msg)

        return list(reader)

def row_from_json(text: str) -> dict[str, Any]:
    """Convert JSON text to a row dictionary."""
    value = json.loads(text)

    if not isinstance(value, dict):
        msg = "Expected JSON object."
        raise ValueError(msg)

    return value

def row_to_json(row: dict[str, Any]) -> str:
    """Convert a row dictionary to compact JSON text."""
    return json.dumps(row, sort_keys=True, separators=(",", ":"))

def missing_csv_field_message(*, field: str, available_fields: list[str]) -> str:
    """Return help text for a missing CSV field.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        field: The name of the missing field.
        available_fields: A list of field names that were found in the CSV file.

    Returns:
        A help message with troubleshooting steps.
    """
    fields = ", ".join(available_fields)

    return f"""
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
The project read the CSV file,
but an expected column was not present.
Required CSV field missing:
    {field}

Available fields were:
    {fields}

CHECK:
1. Open data/sales.csv.
2. Confirm the header row includes: {field}
3. Header names must match exactly.
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
""".strip()

def missing_csv_file_message(*, path: str) -> str:
    """Return help text for a missing CSV file.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        path: The file path that was not found.

    Returns:
        A help message with troubleshooting steps.
    """
    return f"""
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
This project needs a CSV file to generate messages.
Required CSV file not found:
    {path}

CHECK:
1. Confirm you are running the command from the project root folder.
2. Confirm the data folder exists.
3. Confirm data/sales.csv exists.
4. If the file was deleted, restore it from the repository.
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
""".strip()

def create_producer(settings: KafkaSettings) -> Producer:
    """Create a Kafka producer.

    Arguments:
        settings: KafkaSettings object with producer configuration.

    Returns:
        A confluent_kafka.Producer instance.
    """
    return Producer(
        settings.producer_config(),
        logger=logging.getLogger("rdkafka.producer"),
    )

def prepare_producer_topic(settings: KafkaSettings) -> None:
    """Prepare the Kafka topic before producing messages.

    If settings.clear_topic_on_start is true, delete and recreate the topic
    so the producer starts with an empty topic.

    If settings.clear_topic_on_start is false, keep an existing topic.

    If the topic does not exist, create it.
    """
    admin = create_admin_client(settings)

    if topic_exists(admin, settings.topic):
        if settings.clear_topic_on_start:
            delete_topic(admin, settings.topic)
        else:
            return

    if not topic_exists(admin, settings.topic):
        create_topic(admin, settings.topic)

def produce_kafka_message(
    *,
    producer: Producer,
    topic: str,
    key: str,
    message: dict[str, Any],
) -> None:
    """Produce one dictionary message to Kafka as JSON.

    This function sends one message and waits for delivery before returning.
    That makes producer examples reliable and easy to reason about.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        producer: A confluent_kafka.Producer instance.
        topic: The Kafka topic to produce to.
        key: The Kafka message key.
        message: The message dictionary to produce.

    Raises:
        RuntimeError: If Kafka reports a delivery failure.
    """
    delivery_errors: list[str] = []

    def delivery_report(error: Any, _delivered_message: Any) -> None:
        """Record Kafka delivery failure details."""
        if error is not None:
            delivery_errors.append(str(error))

    producer.produce(
        topic=topic,
        key=key.encode("utf-8"),
        value=row_to_json(message).encode("utf-8"),
        callback=delivery_report,
    )

    remaining = producer.flush(timeout=10)

    if remaining > 0:
        detail = f"{remaining} Kafka message(s) were not delivered before timeout."
        msg = kafka_delivery_failed_message(detail=detail)
        raise RuntimeError(msg)

    if delivery_errors:
        detail = "; ".join(delivery_errors)
        msg = kafka_delivery_failed_message(detail=detail)
        raise RuntimeError(msg)

def consume_kafka_message(
    *,
    consumer: Any,
    timeout_seconds: float,
) -> dict[str, Any] | None:
    """Consume one Kafka message and return it as a row dictionary.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        consumer: A confluent_kafka.Consumer instance.
        timeout_seconds: How long to wait for a message before giving up.

    Returns:
        A dictionary representing the message, or None if no message was received.

    """
    message = consumer.poll(timeout_seconds)

    if message is None:
        return None

    if message.error():
        msg = kafka_consume_failed_message(detail=str(message.error()))
        raise RuntimeError(msg)

    raw_value = message.value()

    if raw_value is None:
        return None

    row = row_from_json(raw_value.decode("utf-8"))

    raw_key = message.key()
    row["_kafka_key"] = raw_key.decode("utf-8") if raw_key else ""
    row["_kafka_partition"] = message.partition()
    row["_kafka_offset"] = message.offset()

    return row

def create_consumer(settings: KafkaSettings) -> Consumer:
    """Create a Kafka consumer."""
    return Consumer(
        settings.consumer_config(),
        logger=logging.getLogger("rdkafka.consumer"),
    )

def create_consumer_from_beginning(settings: KafkaSettings) -> Any:
    """Create a Kafka consumer subscribed to the topic from the beginning.

    This is useful for learning examples where each run should read all
    available messages from the topic.

    Arguments:
        settings: KafkaSettings instance with the topic and consumer config.

    Returns:
        A confluent_kafka.Consumer instance subscribed to the topic from the beginning.
    """
    consumer = create_consumer(settings)

    consumer.subscribe(
        [settings.topic],
        on_assign=lambda c, partitions: c.assign(
            [
                TopicPartition(
                    partition.topic,
                    partition.partition,
                    OFFSET_BEGINNING,
                )
                for partition in partitions
            ]
        ),
    )

    return consumer

def verify_consumer_topic_ready(settings: KafkaSettings) -> None:
    """Verify the Kafka topic exists and has messages.

    Arguments:
        settings: KafkaSettings instance with the topic and consumer config.

    Returns:
        None

    Raises:
        RuntimeError: If the topic does not exist or has no messages.
    """
    admin = create_admin_client(settings)

    if not topic_exists(admin, settings.topic):
        msg = kafka_topic_not_found_message(
            topic=settings.topic,
            bootstrap_servers=settings.bootstrap_servers,
        )
        raise RuntimeError(msg)

    message_count = get_topic_message_count(admin, settings.topic, settings)

    if message_count == 0:
        msg = kafka_topic_empty_message(topic=settings.topic)
        raise RuntimeError(msg)

def create_admin_client(settings: KafkaSettings) -> AdminClient:
    """Create a Kafka AdminClient."""
    return AdminClient({"bootstrap.servers": settings.bootstrap_servers})

def create_topic(
    admin: AdminClient,
    topic: str,
    *,
    num_partitions: int = 1,
    replication_factor: int = 1,
) -> None:
    """Create a Kafka topic if it does not already exist.

    All arguments after the asterisk must be passed as keyword arguments.

    Arguments:
        admin: An AdminClient instance.
        topic: The topic name to create.
        num_partitions: Number of partitions (default 1 for local dev).
        replication_factor: Replication factor (default 1 for local dev).

    Raises:
        RuntimeError: If topic creation fails.
    """
    if topic_exists(admin, topic):
        return

    new_topic = NewTopic(
        topic,
        num_partitions=num_partitions,
        replication_factor=replication_factor,
    )

    futures = admin.create_topics([new_topic])

    for topic_name, future in futures.items():
        try:
            future.result()
        except KafkaException as error:
            msg = (
                f"Failed to create topic {topic_name!r}.\n"
                f"Kafka reported: {error}\n\n"
                "Check that Kafka is running and that you have permission to create topics."
            )
            raise RuntimeError(msg) from error

def delete_topic(admin: AdminClient, topic: str) -> None:
    """Delete a Kafka topic if it exists.

    Deleting a topic removes all its messages. Run the producer again
    after deleting to repopulate the topic.

    Arguments:
        admin: An AdminClient instance.
        topic: The topic name to delete.

    Raises:
        RuntimeError: If topic deletion fails.
    """
    if not topic_exists(admin, topic):
        return

    futures = admin.delete_topics([topic])

    for topic_name, future in futures.items():
        try:
            future.result()
        except KafkaException as error:
            msg = (
                f"Failed to delete topic {topic_name!r}.\n"
                f"Kafka reported: {error}\n\n"
                "Check that Kafka is running and that you have permission to delete topics."
            )
            raise RuntimeError(msg) from error

def get_topic_message_count(
    admin: AdminClient, topic: str, settings: KafkaSettings
) -> int:
    """Return the total number of messages available in a topic.

    Sums the high-water offset across all partitions.
    This reflects the total messages ever produced to the topic,
    not the number of unread messages for a specific consumer group.

    Arguments:
        admin: An AdminClient instance.
        topic: The topic name to inspect.
        settings: KafkaSettings instance containing configuration.

    Returns:
        Total message count across all partitions, or 0 if topic is empty.

    Raises:
        RuntimeError: If topic metadata cannot be retrieved.
    """
    try:
        metadata = admin.list_topics(topic=topic, timeout=5)
    except KafkaException as error:
        msg = kafka_admin_failed_message(
            operation="list_topics",
            topic=topic,
            detail=str(error),
        )
        raise RuntimeError(msg) from error

    topic_metadata = metadata.topics.get(topic)
    if topic_metadata is None:
        return 0

    bootstrap_servers = settings.bootstrap_servers
    temp_consumer = Consumer(
        {
            "bootstrap.servers": bootstrap_servers,
            "broker.address.family": settings.broker_address_family,
            "group.id": "_offset_inspector",
            "enable.auto.commit": "false",
        }
    )

    total = 0
    try:
        for partition_id in topic_metadata.partitions:
            tp = TopicPartition(topic, partition_id)
            low, high = temp_consumer.get_watermark_offsets(tp, timeout=5)
            total += max(0, high - low)
    finally:
        temp_consumer.close()

    return total

def list_topics(admin: AdminClient) -> list[str]:
    """Return a sorted list of topic names currently in Kafka.

    Retries several times to allow rdkafka's broker handshake to complete.
    On Windows, the handshake can take up to ~4 seconds after the AdminClient
    is created, which causes an immediate call to fail with a transport error.

    Arguments:
        admin: An AdminClient instance.

    Returns:
        A sorted list of topic name strings.

    Raises:
        RuntimeError: If Kafka is unreachable after all retries.
    """
    last_error: Exception | None = None

    for attempt in range(1, ADMIN_READY_RETRIES + 1):
        try:
            metadata = admin.list_topics(timeout=5)
            return sorted(metadata.topics.keys())
        except KafkaException as error:
            last_error = error
            if attempt < ADMIN_READY_RETRIES:
                time.sleep(ADMIN_READY_DELAY_SECONDS)

    msg = kafka_admin_failed_message(
        operation="list_topics",
        topic="(all)",
        detail=(
            f"Kafka did not respond after {ADMIN_READY_RETRIES} attempts.\n"
            f"    Last error: {last_error}"
        ),
    )
    raise RuntimeError(msg) from last_error

def topic_exists(admin: AdminClient, topic: str) -> bool:
    """Return True if the topic already exists in Kafka."""
    return topic in list_topics(admin)

def verify_kafka_connection(settings: KafkaSettings) -> None:
    """Verify that the Kafka bootstrap server is reachable."""
    bootstrap_server = settings.bootstrap_servers.split(",")[0].strip()

    if ":" not in bootstrap_server:
        msg = (
            "KAFKA_BOOTSTRAP_SERVERS must include host and port, "
            f"but got {bootstrap_server!r}."
        )
        raise ConnectionError(msg)

    host, port_text = bootstrap_server.rsplit(":", 1)

    try:
        port = int(port_text)
    except ValueError as error:
        msg = f"KAFKA_BOOTSTRAP_SERVERS has an invalid port. Got {bootstrap_server!r}."
        raise ConnectionError(msg) from error

    try:
        with socket.create_connection((host, port), timeout=5):
            return
    except OSError as error:
        msg = kafka_not_reachable_message(
            bootstrap_servers=settings.bootstrap_servers,
        )
        raise ConnectionError(msg) from error