b203b7d22c
Implement workflows for fake data generation, scouter processing, and core scouter operations - Added `FakeData` workflow to generate random data and send it to a Kafka topic. - Implemented `Scouter` workflow to load data from Kafka and trigger the core scouter workflow. - Created `CoreScouter` workflow to process data through quality gates, aggregation, and export to PostgreSQL. - Developed comprehensive unit tests for activities and workflows, ensuring proper functionality and error handling. - Enhanced Redis and Postgres activities with robust testing for data handling and error notifications. - Introduced quality filters for data validation and implemented tests to verify their functionality.
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 scouter.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()
|