53337de74a
SIENTIAPDE-1110 Add MongoDB integration and enhance Redis activity with timestamp management functions.
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
from temporalio import activity, workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from sientia_do.temporal.activities.postgres import Postgres
|
|
from sientia_do.notifications.handlers import NotificationHandler
|
|
from sientia_do.temporal.utils.logger import Logger
|
|
from scouter.activities.redis import Redis
|
|
from scouter.activities.kafka import Kafka
|
|
from scouter.activities.gates import Gates
|
|
from scouter.activities.mongodb import MongoDB
|
|
from typing import Any
|
|
|
|
|
|
class Activities(Postgres, Redis, Kafka, Gates, MongoDB):
|
|
"""Activities class that combines multiple services with proper initialization."""
|
|
|
|
def __init__(self,
|
|
postgres_config: dict[str, Any],
|
|
redis_config: dict[str, Any],
|
|
kafka_config: dict[str, Any],
|
|
mongodb_config: dict[str, Any],
|
|
logger: Logger,
|
|
notification_handler: NotificationHandler):
|
|
|
|
# Initialize Postgres
|
|
Postgres.__init__(
|
|
self,
|
|
host=postgres_config['host'],
|
|
port=postgres_config['port'],
|
|
user=postgres_config['user'],
|
|
password=postgres_config['password'],
|
|
dbname=postgres_config['dbname'],
|
|
min_connections=postgres_config['min_connections'],
|
|
max_connections=postgres_config['max_connections'],
|
|
logger=logger,
|
|
notification_handler=notification_handler
|
|
)
|
|
|
|
# Initialize Redis
|
|
Redis.__init__(
|
|
self,
|
|
host=redis_config['host'],
|
|
port=redis_config['port'],
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
username=redis_config['username'],
|
|
password=redis_config['password']
|
|
)
|
|
|
|
# Initialize Kafka
|
|
Kafka.__init__(
|
|
self,
|
|
bootstrap_servers=kafka_config['bootstrap_servers'],
|
|
polling_time=kafka_config['polling_time'],
|
|
group_id=kafka_config['group_id'],
|
|
logger=logger,
|
|
notification_handler=notification_handler
|
|
)
|
|
|
|
# Initialize Gates
|
|
Gates.__init__(
|
|
self,
|
|
logger=logger,
|
|
notification_handler=notification_handler
|
|
)
|
|
|
|
# Initialize MongoDB
|
|
MongoDB.__init__(
|
|
self,
|
|
connection_string=mongodb_config['connection_string'],
|
|
database_name=mongodb_config['database_name'],
|
|
logger=logger,
|
|
notification_handler=notification_handler
|
|
)
|
|
|
|
def shutdown(self):
|
|
Postgres.close(self)
|
|
Kafka.close(self)
|
|
MongoDB.close(self)
|