TapHealth Kafka

A Python library for working with Kafka in TapHealth services. This library provides abstractions over confluent-kafka to simplify producing and consuming Kafka messages.

Requirements

• Python 3.8+
• confluent-kafka >= 2.0.0

Installation

pip install taphealth-kafka

Usage

This package is designed to be installed and used across TapHealth services. The pattern is:

1. Install the package in your project
2. Extend the abstract classes (KafkaProducer/KafkaConsumer) and implement the required abstract methods
3. Use the provided event dataclasses to structure your messages

Connecting to Kafka

from taphealth_kafka import KafkaClient

# Using context manager (recommended — auto-disconnects)
with KafkaClient() as client:
    client.connect(["localhost:9092"])
    # ... use client ...

# Or manually manage the lifecycle
client = KafkaClient()
client.connect(["localhost:9092"])
# ... use client ...
client.disconnect()

# Disable automatic topic creation on connect
client.connect(["localhost:9092"], auto_create_topics=False)

Producing Messages

Extend KafkaProducer and implement the topic property:

from taphealth_kafka import KafkaProducer, Topics

class GlucoseLoggedProducer(KafkaProducer):
    @property
    def topic(self) -> Topics:
        return Topics.GLUCOSE_LOGGED

producer = GlucoseLoggedProducer(kafka_client)

Use event dataclasses to structure your messages:

from datetime import datetime
from taphealth_kafka.events import GlucoseLoggedData, GlucoseType, GlucoseRange

glucose_data = GlucoseLoggedData(
    user_id="user-12345",
    date=datetime.now().strftime("%Y-%m-%d"),
    glucose_type=GlucoseType.FASTING,
    glucose_reading=105.0,
    glucose_range=GlucoseRange.NORMAL,
    is_critical=False,
    timezone="UTC",
)

# Send the dataclass directly — auto-serialized to camelCase JSON
producer.send(glucose_data.to_dict())

Consuming Messages

Extend KafkaConsumer and implement the required abstract methods:

from taphealth_kafka import KafkaConsumer, Topics

class GlucoseLoggedConsumer(KafkaConsumer):
    @property
    def topic(self) -> Topics:
        return Topics.GLUCOSE_LOGGED

    @property
    def group_id(self) -> str:
        return "glucose-processor-group"

    def on_message(self, data: dict, message) -> None:
        # Deserialize back into a dataclass
        from taphealth_kafka.events import GlucoseLoggedData
        glucose = GlucoseLoggedData.from_dict(data)
        print(f"User: {glucose.user_id}, Reading: {glucose.glucose_reading} mg/dL")

consumer = GlucoseLoggedConsumer(kafka_client)
consumer.consume()  # Blocking call

Override consumer_config to customize consumer settings:

class LatestOffsetConsumer(KafkaConsumer):
    @property
    def consumer_config(self) -> dict:
        return {
            "auto_offset_reset": "latest",
            "enable_auto_commit": False,
        }

    # ... implement topic, group_id, on_message ...

Serialization

All event dataclasses support bidirectional serialization between Python snake_case and Kafka camelCase: for common TapHealth contracts, enum fields follow kafka/ numeric enum encoding.

from taphealth_kafka.events import GlucoseLoggedData, GlucoseType, GlucoseRange

# Serialize to camelCase dict
glucose_data = GlucoseLoggedData(
    user_id="user-12345",
    date="2026-01-15",
    glucose_type=GlucoseType.FASTING,
    glucose_reading=105.0,
    glucose_range=GlucoseRange.NORMAL,
    is_critical=False,
    timezone="UTC",
)
data_dict = glucose_data.to_dict()
# {"userId": "user-12345", "date": "2026-01-15", "type": 0, "glucoseRange": 1, ...}

# Deserialize from camelCase dict
glucose = GlucoseLoggedData.from_dict(data_dict)
assert glucose.user_id == "user-12345"

For advanced use cases, the serialization utilities are available directly:

from taphealth_kafka import dataclass_to_dict, dataclass_from_dict

# Convert any dataclass to camelCase dict
data_dict = dataclass_to_dict(my_dataclass)

# Reconstruct from camelCase dict with full type awareness
obj = dataclass_from_dict(MyDataclass, data_dict)

Disconnecting

# Preferred: use the context manager (see "Connecting to Kafka")
# Manual: call disconnect() to flush pending messages and close consumers
kafka_client.disconnect()

Complete Example

Here's a complete example showing a producer and consumer for glucose logging:

"""
Glucose Logging Example

Usage:
    # Terminal 1 - Start the consumer
    python example.py consumer

    # Terminal 2 - Send a message with the producer
    python example.py producer
"""

import sys
from datetime import datetime

from taphealth_kafka import (
    KafkaClient,
    KafkaConsumer,
    KafkaProducer,
    Topics,
)
from taphealth_kafka.events import (
    GlucoseLoggedData,
    GlucoseRange,
    GlucoseType,
)


class GlucoseLoggedProducer(KafkaProducer):
    @property
    def topic(self) -> Topics:
        return Topics.GLUCOSE_LOGGED


class GlucoseLoggedConsumer(KafkaConsumer):
    @property
    def topic(self) -> Topics:
        return Topics.GLUCOSE_LOGGED

    @property
    def group_id(self) -> str:
        return "glucose-example-group"

    def on_message(self, data: dict, message) -> None:
        glucose = GlucoseLoggedData.from_dict(data)
        print(f"Received: user={glucose.user_id}, reading={glucose.glucose_reading}")


def run_producer(client: KafkaClient) -> None:
    producer = GlucoseLoggedProducer(client)

    glucose_data = GlucoseLoggedData(
        user_id="user-12345",
        date=datetime.now().strftime("%Y-%m-%d"),
        glucose_type=GlucoseType.FASTING,
        glucose_reading=105.0,
        glucose_range=GlucoseRange.NORMAL,
        is_critical=False,
        timezone="UTC",
    )

    producer.send(glucose_data.to_dict())
    print("Event sent!")


def run_consumer(client: KafkaClient) -> None:
    consumer = GlucoseLoggedConsumer(client)
    print("Listening for events... (Ctrl+C to stop)")
    consumer.consume()


def main() -> None:
    with KafkaClient() as client:
        client.connect(["localhost:9092"])

        mode = sys.argv[1] if len(sys.argv) > 1 else "consumer"

        if mode == "producer":
            run_producer(client)
        else:
            run_consumer(client)


if __name__ == "__main__":
    main()

API Reference

KafkaClient

The main client for managing Kafka connections. Supports the context manager protocol.

KafkaProducer (Abstract)

Base class for producers. Subclasses must implement:

KafkaConsumer (Abstract)

Base class for consumers. Subclasses must implement:

Topics

Enum containing available Kafka topics:

Deprecated: Topics.UPDATE_WEEEKLY_NUTRTION_PLAN still works but emits a DeprecationWarning. Use Topics.UPDATE_WEEKLY_NUTRITION_PLAN instead.

Use Topics.get_all_topics() to get all topic names as a list of strings.

Serialization Utilities

Event Dataclasses

The package provides typed dataclasses for structuring event messages. All dataclasses have to_dict() and from_dict() methods for bidirectional camelCase serialization:

Supporting types and enums:

Configuration

When creating consumers, you can pass additional configuration options:

consumer = kafka_client.create_consumer(
    group_id="my-group",
    auto_offset_reset="earliest",      # or "latest"
    enable_auto_commit=True,
    session_timeout_ms=6000,
    max_poll_interval_ms=300000
)

Or override consumer_config in your KafkaConsumer subclass (see Consuming Messages).

License

MIT