133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
from sientia_do.observability.metrics_controller import MetricsController
|
|
from temporalio import workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from os import getenv
|
|
from typing import Any
|
|
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.observability.logger import Logger
|
|
from sientia_do.temporal.activities.postgres_sync import Postgres
|
|
|
|
from scouter.activities.api import API
|
|
from scouter.activities.gates import Gates
|
|
from scouter.activities.mongodb import MongoDB
|
|
from scouter.activities.redis import Redis
|
|
|
|
|
|
class Activities(Postgres, Redis, Gates, MongoDB, API):
|
|
"""
|
|
Unified activities class that combines multiple data processing services.
|
|
|
|
This class provides a comprehensive interface for all data processing activities
|
|
by inheriting from specialized service classes. It handles:
|
|
- PostgreSQL operations for data persistence
|
|
- Redis operations for caching and temporary storage
|
|
- Data quality gates and filtering
|
|
- MongoDB operations for data retrieval
|
|
- PI Web API operations for external data ingestion
|
|
- Notification handling and logging
|
|
|
|
The class implements the multiple inheritance pattern to provide a unified
|
|
interface while maintaining separation of concerns across different data services.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
postgres_config: dict[str, Any],
|
|
redis_config: dict[str, Any],
|
|
mongodb_config: dict[str, Any],
|
|
api_config: dict[str, Any],
|
|
logger: Logger,
|
|
notification_handler: NotificationHandler,
|
|
):
|
|
"""
|
|
Initialize the Activities class with all required services.
|
|
|
|
Args:
|
|
postgres_config (dict[str, Any]): PostgreSQL connection configuration.
|
|
Required fields: host, port, user, password, dbname, min_connections, max_connections
|
|
redis_config (dict[str, Any]): Redis connection configuration.
|
|
Required fields: host, port, username, password
|
|
mongodb_config (dict[str, Any]): MongoDB connection configuration.
|
|
Required fields: connection_string, database_name
|
|
api_config (dict[str, Any]): PI Web API configuration.
|
|
Required fields: base_url, auth_type, auth_token
|
|
logger (Logger): Logger instance for application logging
|
|
notification_handler (NotificationHandler): Handler for system notifications
|
|
"""
|
|
metrics_controller = MetricsController(
|
|
logger=logger,
|
|
)
|
|
|
|
# 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,
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
# 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'],
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
# Initialize Gates
|
|
Gates.__init__(
|
|
self,
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
# Initialize MongoDB
|
|
MongoDB.__init__(
|
|
self,
|
|
connection_string=mongodb_config['connection_string'],
|
|
database_name=mongodb_config['database_name'],
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
# Initialize API
|
|
API.__init__(
|
|
self,
|
|
base_url=api_config['base_url'],
|
|
auth_type=api_config['auth_type'],
|
|
auth_token=api_config['auth_token'],
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
self.pod_id = getenv('HOSTNAME', 'localhost')
|
|
|
|
def shutdown(self):
|
|
"""
|
|
Gracefully shutdown all service connections.
|
|
|
|
This method ensures proper cleanup of database connections and resources
|
|
to prevent connection leaks and ensure graceful application termination.
|
|
"""
|
|
Postgres.close(self)
|
|
MongoDB.close(self)
|
|
Redis.close(self)
|
|
Gates.close(self)
|
|
API.close(self)
|