7dbb9a29ea
Update environment configuration and refactor activity imports - Changed Kafka, Redis, and Temporal host configurations to use localhost. - Updated the version reference for the sientia-dataops-library in requirements.txt. - Refactored import paths for activities to align with new module structure. - Removed unused base.py and postgres.py files. - Updated logger and policies imports to reflect new module locations. - Adjusted values.yaml for branch and log level settings.
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
from temporalio import workflow, activity
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from logging import Logger
|
|
from sientia_do.notifications.handlers import NotificationHandler
|
|
from sientia_do.temporal.activities.base import BaseActivity
|
|
from typing import Any
|
|
from kafka import KafkaConsumer
|
|
from pandas import DataFrame
|
|
import json
|
|
|
|
|
|
class Kafka(BaseActivity):
|
|
def __init__(self, bootstrap_servers: str, polling_time: int,
|
|
group_id: str, logger: Logger, notification_handler: NotificationHandler):
|
|
self.polling_time = polling_time
|
|
|
|
self.kafka_connector = KafkaConsumer(
|
|
bootstrap_servers=bootstrap_servers,
|
|
auto_offset_reset="earliest",
|
|
enable_auto_commit=True,
|
|
group_id=group_id,
|
|
value_deserializer=lambda x: json.loads(x.decode("utf-8"))
|
|
)
|
|
|
|
BaseActivity.__init__(self, logger, notification_handler)
|
|
|
|
@activity.defn(name="load_from_kafka")
|
|
async def load_from_kafka(self, input_data: dict[str, Any]) -> dict[str, Any]:
|
|
"""
|
|
Loads data from a kafka topic. Polls the topic for a given time and returns the data.
|
|
|
|
Args:
|
|
input_data (dict[str, Any]): The data to load. Contains:
|
|
topic (str): The topic to load data from.
|
|
Returns:
|
|
dict[str, Any]: The data loaded from the topic.
|
|
"""
|
|
|
|
self.logger.debug(f"Loading data from topic: {input_data['topic']}")
|
|
|
|
topic = input_data["topic"]
|
|
|
|
# Subscribe to the specified topic
|
|
self.kafka_connector.subscribe([topic])
|
|
|
|
# List to store message values
|
|
message_values = []
|
|
|
|
# Poll for messages
|
|
records = self.kafka_connector.poll(timeout_ms=self.polling_time)
|
|
|
|
self.logger.debug(f"Polled {len(records)} records from topic: {topic}")
|
|
|
|
# Process the polled records
|
|
for _topic_partition, msgs in records.items():
|
|
for msg in msgs:
|
|
message_values.append(msg.value)
|
|
|
|
# Return empty dict if no messages were received
|
|
if not message_values:
|
|
return {}
|
|
|
|
self.logger.debug(
|
|
f"Loaded {len(message_values)} messages from topic: {topic}")
|
|
|
|
self.logger.debug(
|
|
f"Loaded data: {message_values}")
|
|
|
|
return DataFrame(message_values).to_dict()
|