gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1163

Browse Source 
Update requirements and refactor Ingestor class for MongoDB integration

- Updated sientia-dataops-library dependency version to 1.3.0 in requirements.txt.
- Refactored Ingestor class to use CoreNotificationHandler and added MongoDB connection parameters to improve configuration management.
This commit is contained in:
vitor-aignosi
2025-07-16 08:59:24 -03:00
parent c2f2664622
commit 3e6ab33ef4
2 changed files with 13 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
ingestor/ingestor.py
View File

@@ -3,7 +3,8 @@ from os import getenv
from copy import deepcopy from copy import deepcopy
from typing import Dict, Any from typing import Dict, Any
from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.utils.logger import get_logger
from ingestor.managers.ingestor_manager import IngestorManager from ingestor.managers.ingestor_manager import IngestorManager
@@ -23,6 +24,10 @@ class Ingestor:
HEARTBEAT_TTL (int): Time-to-live for heartbeats in seconds. Defaults to 20. HEARTBEAT_TTL (int): Time-to-live for heartbeats in seconds. Defaults to 20.
HOSTNAME (str): Identifier for the current pod or host. Defaults to "localhost". HOSTNAME (str): Identifier for the current pod or host. Defaults to "localhost".
POLL_INTERVAL (int): Interval in seconds for polling operations. Defaults to 5. POLL_INTERVAL (int): Interval in seconds for polling operations. Defaults to 5.
MONGODB_URL (str): URL of the MongoDB server. Defaults to "localhost:27017".
MONGODB_USERNAME (str): Username for the MongoDB server. Defaults to "sientia".
MONGODB_PASSWORD (str): Password for the MongoDB server. Defaults to "sientia".
MONGODB_DATABASE (str): Name of the MongoDB database. Defaults to "sientia".
Attributes: Attributes:
kafka_servers (list): List of Kafka server addresses. kafka_servers (list): List of Kafka server addresses.
redis_host (str): Hostname of the Redis server. redis_host (str): Hostname of the Redis server.
@@ -50,17 +55,18 @@ class Ingestor:
self.heartbeat_ttl = int(getenv("HEARTBEAT_TTL", "20")) self.heartbeat_ttl = int(getenv("HEARTBEAT_TTL", "20"))
self.pod_id = getenv("HOSTNAME", "localhost") self.pod_id = getenv("HOSTNAME", "localhost")
self.poll_interval = int(getenv("POLL_INTERVAL", "5")) self.poll_interval = int(getenv("POLL_INTERVAL", "5"))
mongo_url = getenv("MONGODB_URL", 'localhost:27017') mongo_url = getenv("MONGODB_URL", "localhost:27017")
mongo_username = getenv("MONGODB_USERNAME", 'sientia') mongo_username = getenv("MONGODB_USERNAME", "sientia")
mongo_password = getenv("MONGODB_PASSWORD", 'sientia') mongo_password = getenv("MONGODB_PASSWORD", "sientia")
self.mongo_database = getenv("MONGODB_DATABASE", 'sientia') self.mongo_database = getenv("MONGODB_DATABASE", "sientia")
self.mongo_connection_string = f"mongodb://{mongo_username}:{mongo_password}@{mongo_url}" self.mongo_connection_string = f"mongodb://{mongo_username}:{mongo_password}@{mongo_url}"
self.kafka_servers = kafka_servers.split(",") self.kafka_servers = kafka_servers.split(",")
self.logger = None self.logger = None
self.init_logger() self.init_logger()
self.notification_handler = NotificationHandler( self.notification_handler = NotificationHandler(
servers=self.kafka_servers, connection_string=self.mongo_connection_string,
database=self.mongo_database,
logger=self.logger, logger=self.logger,
project_name="OPC_INGESTOR" project_name="OPC_INGESTOR"
) )

2
requirements.txt
View File

@@ -1,5 +1,5 @@
asyncua==1.1.5 asyncua==1.1.5
redis redis
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.0 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.0
prometheus_client prometheus_client
pymongo pymongo