SIENTIAPDE-1084
Refactor connectors_config.py and conditional_filters.py for improved configuration management and data filtering - Updated PostgreSQL and MLFlow configuration functions to enhance default values and environment variable handling. - Simplified OPC server configuration logic and improved MongoDB connection string construction. - Refactored conditional filters to streamline null value checks and empty data validation, removing unnecessary comments and examples for clarity. - Removed extensive module docstrings to enhance code readability.
This commit is contained in:
vitor-aignosi
@@ -1,205 +1,99 @@
|
||||
"""
|
||||
Connectors Configuration Module
|
||||
|
||||
This module provides configuration management for all external service connectors
|
||||
used by the Sientia DataOps Laborious system. It centralizes configuration
|
||||
for databases, MLFlow servers, OPC servers, and other external dependencies.
|
||||
|
||||
The module implements configuration builders for:
|
||||
1. PostgreSQL database connections
|
||||
2. MLFlow model serving endpoints
|
||||
3. OPC server configurations
|
||||
4. MongoDB notification systems
|
||||
|
||||
Key Features:
|
||||
- Environment variable-based configuration
|
||||
- Default value management for development
|
||||
- Connection pool configuration
|
||||
- Security credential management
|
||||
- Configuration validation and error handling
|
||||
- Support for multiple service instances
|
||||
|
||||
Configuration Sources:
|
||||
- Environment variables for production deployment
|
||||
- Default values for local development
|
||||
- Kubernetes secrets integration
|
||||
- Configurable connection parameters
|
||||
|
||||
Environment Variables:
|
||||
- POSTGRES_*: PostgreSQL connection parameters
|
||||
- MLFLOW_*: MLFlow server parameters
|
||||
- OPC_*: OPC server configuration
|
||||
- MONGODB_*: MongoDB connection parameters
|
||||
|
||||
Dependencies:
|
||||
- os: Environment variable access
|
||||
- typing: Type hints and annotations
|
||||
"""
|
||||
|
||||
import os
|
||||
from os import getenv
|
||||
import json
|
||||
from typing import Dict, Any
|
||||
|
||||
|
||||
def build_postgres_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build PostgreSQL database configuration from environment variables.
|
||||
|
||||
|
||||
This function constructs a PostgreSQL configuration dictionary from
|
||||
environment variables with sensible defaults for local development.
|
||||
It handles connection pool configuration and security parameters.
|
||||
|
||||
|
||||
Environment Variables:
|
||||
POSTGRES_HOST: Database hostname (default: localhost)
|
||||
POSTGRES_PORT: Database port (default: 5432)
|
||||
POSTGRES_USER: Database username (default: sientia)
|
||||
POSTGRES_PASSWORD: Database password (default: sientia)
|
||||
POSTGRES_DBNAME: Database name (default: sientia)
|
||||
POSTGRES_MIN_CONNECTIONS: Minimum connection pool size (default: 1)
|
||||
POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 10)
|
||||
|
||||
POSTGRES_MIN_CONNECTIONS: Minimum connection pool size (default: 5)
|
||||
POSTGRES_MAX_CONNECTIONS: Maximum connection pool size (default: 20)
|
||||
|
||||
Returns:
|
||||
dict: PostgreSQL configuration dictionary with all required parameters
|
||||
|
||||
Example:
|
||||
>>> config = build_postgres_config()
|
||||
>>> print(config)
|
||||
{
|
||||
'host': 'localhost',
|
||||
'port': 5432,
|
||||
'user': 'sientia',
|
||||
'password': 'sientia',
|
||||
'dbname': 'sientia',
|
||||
'min_connections': 1,
|
||||
'max_connections': 10
|
||||
}
|
||||
|
||||
Note:
|
||||
In production, ensure all required environment variables are set
|
||||
with appropriate values for your database environment.
|
||||
"""
|
||||
return {
|
||||
'host': os.getenv('POSTGRES_HOST', 'localhost'),
|
||||
'port': int(os.getenv('POSTGRES_PORT', '5432')),
|
||||
'user': os.getenv('POSTGRES_USER', 'sientia'),
|
||||
'password': os.getenv('POSTGRES_PASSWORD', 'sientia'),
|
||||
'dbname': os.getenv('POSTGRES_DBNAME', 'sientia'),
|
||||
'min_connections': int(os.getenv('POSTGRES_MIN_CONNECTIONS', '1')),
|
||||
'max_connections': int(os.getenv('POSTGRES_MAX_CONNECTIONS', '10'))
|
||||
'host': getenv('POSTGRES_HOST', 'localhost'),
|
||||
'port': int(getenv('POSTGRES_PORT', '5432')),
|
||||
'user': getenv('POSTGRES_USER', 'sientia'),
|
||||
'password': getenv('POSTGRES_PASSWORD', 'sientia'),
|
||||
'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
|
||||
'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
|
||||
'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20'))
|
||||
}
|
||||
|
||||
|
||||
def build_mlflow_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build MLFlow server configuration from environment variables.
|
||||
|
||||
|
||||
This function constructs an MLFlow configuration dictionary from
|
||||
environment variables with sensible defaults for local development.
|
||||
It handles server connection and authentication parameters.
|
||||
|
||||
|
||||
Environment Variables:
|
||||
MLFLOW_HOST: MLFlow server hostname (default: localhost)
|
||||
MLFLOW_PORT: MLFlow server port (default: 5000)
|
||||
MLFLOW_USERNAME: MLFlow username (default: admin)
|
||||
MLFLOW_PASSWORD: MLFlow password (default: admin)
|
||||
|
||||
MLFLOW_HOST: MLFlow server hostname (default: http://localhost)
|
||||
MLFLOW_PORT: MLFlow server port (default: 5080)
|
||||
MLFLOW_USERNAME: MLFlow username (default: aignosi)
|
||||
MLFLOW_PASSWORD: MLFlow password (default: aignosi)
|
||||
|
||||
Returns:
|
||||
dict: MLFlow configuration dictionary with all required parameters
|
||||
|
||||
Example:
|
||||
>>> config = build_mlflow_config()
|
||||
>>> print(config)
|
||||
{
|
||||
'host': 'localhost',
|
||||
'port': 5000,
|
||||
'username': 'admin',
|
||||
'password': 'admin'
|
||||
}
|
||||
|
||||
Note:
|
||||
In production, ensure all required environment variables are set
|
||||
with appropriate values for your MLFlow server environment.
|
||||
Consider using secure authentication methods for production deployments.
|
||||
"""
|
||||
return {
|
||||
'host': os.getenv('MLFLOW_HOST', 'localhost'),
|
||||
'port': int(os.getenv('MLFLOW_PORT', '5000')),
|
||||
'username': os.getenv('MLFLOW_USERNAME', 'admin'),
|
||||
'password': os.getenv('MLFLOW_PASSWORD', 'admin')
|
||||
'host': getenv('MLFLOW_HOST', 'http://localhost'),
|
||||
'port': int(getenv('MLFLOW_PORT', '5080')),
|
||||
'username': getenv('MLFLOW_USERNAME', 'aignosi'),
|
||||
'password': getenv('MLFLOW_PASSWORD', 'aignosi')
|
||||
}
|
||||
|
||||
|
||||
def build_opc_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build OPC server configuration from environment variables.
|
||||
|
||||
|
||||
This function constructs an OPC server configuration dictionary from
|
||||
environment variables. It supports both single server and multi-server
|
||||
configurations with flexible parameter handling.
|
||||
|
||||
|
||||
Environment Variables:
|
||||
OPC_CONFIG: JSON string containing multiple OPC server configurations
|
||||
OPC_URL: Single OPC server URL (fallback)
|
||||
OPC_NAME: Single OPC server name (fallback)
|
||||
OPC_SERVER_URI: Single OPC server URI (fallback)
|
||||
OPC_CERT_PATH: Client certificate path (fallback)
|
||||
OPC_PRIVATE_KEY_PATH: Client private key path (fallback)
|
||||
OPC_SERVER_CERT_PATH: Server certificate path (fallback)
|
||||
OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback)
|
||||
|
||||
OPC_ID: OPC server ID (fallback, default: 1)
|
||||
OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840)
|
||||
OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840)
|
||||
OPC_CERT_PATH: Client certificate path (fallback, default: None)
|
||||
OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None)
|
||||
OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None)
|
||||
OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback, default: 120)
|
||||
|
||||
Returns:
|
||||
dict: OPC server configuration dictionary
|
||||
|
||||
Configuration Modes:
|
||||
1. Multi-server: Use OPC_CONFIG environment variable with JSON string
|
||||
2. Single server: Use individual OPC_* environment variables
|
||||
|
||||
Example Multi-server Configuration:
|
||||
>>> # Set OPC_CONFIG environment variable
|
||||
>>> os.environ['OPC_CONFIG'] = '''
|
||||
... {
|
||||
... "opc_server_1": {
|
||||
... "url": "opc.tcp://server1:4840",
|
||||
... "name": "Server1",
|
||||
... "server_uri": "urn:server1:opcua",
|
||||
... "cert_path": "/path/to/cert.pem",
|
||||
... "private_key_path": "/path/to/key.pem",
|
||||
... "server_cert_path": "/path/to/server_cert.pem",
|
||||
... "reconnection_interval": 5000
|
||||
... }
|
||||
... }
|
||||
... '''
|
||||
>>> config = build_opc_config()
|
||||
|
||||
Example Single Server Configuration:
|
||||
>>> # Set individual environment variables
|
||||
>>> os.environ['OPC_URL'] = 'opc.tcp://localhost:4840'
|
||||
>>> os.environ['OPC_NAME'] = 'LocalServer'
|
||||
>>> config = build_opc_config()
|
||||
|
||||
Note:
|
||||
For production deployments, prefer the OPC_CONFIG approach for
|
||||
multiple servers and ensure all certificate paths are properly configured.
|
||||
"""
|
||||
# Check for multi-server configuration
|
||||
opc_config = os.getenv('OPC_CONFIG')
|
||||
if opc_config:
|
||||
try:
|
||||
import json
|
||||
return json.loads(opc_config)
|
||||
except (json.JSONDecodeError, ImportError) as e:
|
||||
# Fall back to single server configuration if JSON parsing fails
|
||||
pass
|
||||
|
||||
# Single server configuration fallback
|
||||
opc_raw = getenv('OPC_CONFIG', None)
|
||||
|
||||
if opc_raw:
|
||||
return json.loads(opc_raw)
|
||||
|
||||
return {
|
||||
'default': {
|
||||
'url': os.getenv('OPC_URL', 'opc.tcp://localhost:4840'),
|
||||
'name': os.getenv('OPC_NAME', 'DefaultServer'),
|
||||
'server_uri': os.getenv('OPC_SERVER_URI', 'urn:default:opcua'),
|
||||
'cert_path': os.getenv('OPC_CERT_PATH', ''),
|
||||
'private_key_path': os.getenv('OPC_PRIVATE_KEY_PATH', ''),
|
||||
'server_cert_path': os.getenv('OPC_SERVER_CERT_PATH', ''),
|
||||
'reconnection_interval': int(os.getenv('OPC_RECONNECTION_INTERVAL', '5000'))
|
||||
getenv('OPC_ID', '1'): {
|
||||
'id': getenv('OPC_ID', '1'),
|
||||
'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'),
|
||||
'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'),
|
||||
'cert_path': getenv('OPC_CERT_PATH', None),
|
||||
'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None),
|
||||
'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None),
|
||||
'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120'))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -207,32 +101,29 @@ def build_opc_config() -> Dict[str, Any]:
|
||||
def build_mongodb_config() -> Dict[str, Any]:
|
||||
"""
|
||||
Build MongoDB configuration from environment variables.
|
||||
|
||||
|
||||
This function constructs a MongoDB configuration dictionary from
|
||||
environment variables with sensible defaults for local development.
|
||||
It handles connection string and database name configuration.
|
||||
|
||||
|
||||
Environment Variables:
|
||||
MONGODB_URL: MongoDB connection URI (default: localhost:27017)
|
||||
MONGODB_DATABASE: MongoDB database name (default: sientia)
|
||||
|
||||
MONGODB_USERNAME: MongoDB username (default: root)
|
||||
MONGODB_PASSWORD: MongoDB password (default: wKZDbMNU1c)
|
||||
MONGODB_URL: MongoDB connection URI (default: localhost:27018)
|
||||
MONGODB_DATABASE_NAME: MongoDB database name (default: sientia)
|
||||
MONGODB_TTL_INDEX_HOURS: TTL index duration in hours (default: 1)
|
||||
|
||||
Returns:
|
||||
dict: MongoDB configuration dictionary with connection parameters
|
||||
|
||||
Example:
|
||||
>>> config = build_mongodb_config()
|
||||
>>> print(config)
|
||||
{
|
||||
'connection_string': 'localhost:27017',
|
||||
'database_name': 'sientia'
|
||||
}
|
||||
|
||||
Note:
|
||||
In production, ensure the MONGODB_URL environment variable is set
|
||||
with a proper MongoDB connection string including authentication
|
||||
if required by your MongoDB deployment.
|
||||
"""
|
||||
username = getenv('MONGODB_USERNAME', 'root')
|
||||
password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
|
||||
uri = getenv('MONGODB_URL', 'localhost:27018')
|
||||
|
||||
connection_string = f'mongodb://{username}:{password}@{uri}'
|
||||
|
||||
return {
|
||||
'connection_string': os.getenv('MONGODB_URL', 'localhost:27017'),
|
||||
'database_name': os.getenv('MONGODB_DATABASE', 'sientia')
|
||||
'connection_string': connection_string,
|
||||
'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
|
||||
'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600
|
||||
}
|
||||
|
||||
@@ -1,197 +1,45 @@
|
||||
"""
|
||||
Conditional Data Filters Module
|
||||
|
||||
This module provides conditional data filtering functions for the Sientia DataOps Laborious system.
|
||||
It implements data quality validation filters that can be applied to input data before
|
||||
ML operations to ensure data integrity and quality.
|
||||
|
||||
The module implements filters for:
|
||||
1. Empty data detection and validation
|
||||
2. Specific variable null value checking
|
||||
3. Configurable data quality rules
|
||||
4. Flexible filter configuration
|
||||
|
||||
Key Features:
|
||||
- Configurable filter policies and thresholds
|
||||
- Multiple data quality validation rules
|
||||
- Flexible configuration options
|
||||
- Comprehensive error handling
|
||||
- Performance-optimized filtering
|
||||
|
||||
Filter Types:
|
||||
- EMPTY_DATA: Detects empty or insufficient data sets
|
||||
- SPECIFIC_VARIABLES_NULL_VALUES: Validates specific variable null values
|
||||
- Custom filters can be added for specific validation needs
|
||||
|
||||
Dependencies:
|
||||
- pandas.DataFrame: Data manipulation and processing
|
||||
- typing: Type hints and annotations
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List
|
||||
from pandas import DataFrame
|
||||
|
||||
|
||||
def filter_empty_data(data: DataFrame, config: Dict[str, Any]) -> bool:
|
||||
def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool:
|
||||
"""
|
||||
Filter data based on empty data conditions.
|
||||
|
||||
This function checks if the input data meets minimum requirements for
|
||||
processing. It can validate data size, completeness, and other quality
|
||||
metrics to ensure sufficient data is available for ML operations.
|
||||
|
||||
The filter implements multiple validation criteria:
|
||||
1. Data frame size validation
|
||||
2. Row count validation
|
||||
3. Column completeness validation
|
||||
4. Configurable threshold checking
|
||||
|
||||
Filter to check if specific variables contain null values.
|
||||
|
||||
This function examines a DataFrame to determine if any of the specified variables
|
||||
contain null (NaN) values. It returns True if null values are found for any of
|
||||
the specified variables, False otherwise.
|
||||
|
||||
Args:
|
||||
data: Input data as pandas DataFrame
|
||||
config: Filter configuration dictionary
|
||||
Required keys:
|
||||
- min_rows (int, optional): Minimum number of rows required
|
||||
- min_columns (int, optional): Minimum number of columns required
|
||||
- min_data_points (int, optional): Minimum total data points required
|
||||
|
||||
data (DataFrame): The pandas DataFrame to be examined. Must contain columns
|
||||
named 'variable' and 'value'.
|
||||
config (dict): Configuration dictionary containing the following key:
|
||||
- variables (list): List of variable names to check for null values
|
||||
|
||||
Returns:
|
||||
bool: True if data should be filtered (fails quality check), False otherwise
|
||||
|
||||
Filter Logic:
|
||||
- Returns True (filter) if data is empty or below thresholds
|
||||
- Returns False (pass) if data meets quality requirements
|
||||
- Handles missing configuration gracefully with defaults
|
||||
|
||||
Example:
|
||||
>>> import pandas as pd
|
||||
>>> df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})
|
||||
>>> config = {'min_rows': 2, 'min_columns': 2}
|
||||
>>> result = filter_empty_data(df, config)
|
||||
>>> print(result)
|
||||
False # Data passes filter
|
||||
|
||||
>>> empty_df = pd.DataFrame()
|
||||
>>> result = filter_empty_data(empty_df, config)
|
||||
>>> print(result)
|
||||
True # Data fails filter
|
||||
|
||||
Default Thresholds:
|
||||
- min_rows: 1 (at least one row required)
|
||||
- min_columns: 1 (at least one column required)
|
||||
- min_data_points: 1 (at least one data point required)
|
||||
bool: True if any of the specified variables contain null values,
|
||||
False if none of the specified variables contain null values.
|
||||
|
||||
"""
|
||||
# Check if data is completely empty
|
||||
if data.empty:
|
||||
return True
|
||||
|
||||
# Get configuration with defaults
|
||||
min_rows = config.get('min_rows', 1)
|
||||
min_columns = config.get('min_columns', 1)
|
||||
min_data_points = config.get('min_data_points', 1)
|
||||
|
||||
# Check row count
|
||||
if len(data) < min_rows:
|
||||
return True
|
||||
|
||||
# Check column count
|
||||
if len(data.columns) < min_columns:
|
||||
return True
|
||||
|
||||
# Check total data points
|
||||
if data.size < min_data_points:
|
||||
return True
|
||||
|
||||
# Data passes all quality checks
|
||||
return False
|
||||
return not data[
|
||||
data['variable'].isin(config['variables']) & data['value'].isna()].empty
|
||||
|
||||
|
||||
def filter_specific_variables_null_values(data: DataFrame, config: Dict[str, Any]) -> bool:
|
||||
def filter_empty_data(data: DataFrame, _config: dict) -> bool:
|
||||
"""
|
||||
Filter data based on null values in specific variables.
|
||||
|
||||
This function checks for null values in specified variables and determines
|
||||
if the data quality is sufficient for processing. It can validate
|
||||
individual columns or groups of columns for data completeness.
|
||||
|
||||
The filter implements variable-specific validation:
|
||||
1. Individual variable null value checking
|
||||
2. Configurable null value thresholds
|
||||
3. Multiple variable validation
|
||||
4. Flexible threshold configuration
|
||||
|
||||
Filter to check if the DataFrame is empty.
|
||||
|
||||
This function determines whether the provided DataFrame contains any data.
|
||||
It's a simple utility function that can be used in conditional logic to
|
||||
handle cases where no data is available.
|
||||
|
||||
Args:
|
||||
data: Input data as pandas DataFrame
|
||||
config: Filter configuration dictionary
|
||||
Required keys:
|
||||
- variables (list): List of variable names to check
|
||||
- max_null_ratio (float, optional): Maximum allowed null value ratio (0.0 to 1.0)
|
||||
- max_null_count (int, optional): Maximum allowed null value count
|
||||
|
||||
data (DataFrame): The pandas DataFrame to be checked for emptiness.
|
||||
_config (dict): Configuration dictionary (unused in this function).
|
||||
The underscore prefix indicates this parameter is required for
|
||||
interface consistency but not used in the implementation.
|
||||
|
||||
Returns:
|
||||
bool: True if data should be filtered (fails quality check), False otherwise
|
||||
|
||||
Filter Logic:
|
||||
- Returns True (filter) if null value thresholds are exceeded
|
||||
- Returns False (pass) if null values are within acceptable limits
|
||||
- Handles missing variables gracefully
|
||||
- Supports both ratio and count-based thresholds
|
||||
|
||||
Example:
|
||||
>>> import pandas as pd
|
||||
>>> df = pd.DataFrame({
|
||||
... 'temperature': [25.5, None, 27.0, 26.5],
|
||||
... 'humidity': [60.0, 65.0, None, 62.0]
|
||||
... })
|
||||
>>> config = {
|
||||
... 'variables': ['temperature', 'humidity'],
|
||||
... 'max_null_ratio': 0.25
|
||||
... }
|
||||
>>> result = filter_specific_variables_null_values(df, config)
|
||||
>>> print(result)
|
||||
False # Data passes filter (null ratio = 0.25, which equals max)
|
||||
|
||||
>>> config = {
|
||||
... 'variables': ['temperature', 'humidity'],
|
||||
... 'max_null_ratio': 0.20
|
||||
... }
|
||||
>>> result = filter_specific_variables_null_values(df, config)
|
||||
>>> print(result)
|
||||
True # Data fails filter (null ratio = 0.25, exceeds max of 0.20)
|
||||
|
||||
Default Thresholds:
|
||||
- max_null_ratio: 0.5 (50% null values allowed)
|
||||
- max_null_count: None (no count-based limit by default)
|
||||
|
||||
Note:
|
||||
If both max_null_ratio and max_null_count are specified, the filter
|
||||
will trigger if either threshold is exceeded.
|
||||
bool: True if the DataFrame is empty (has no rows), False if it contains data.
|
||||
|
||||
"""
|
||||
# Get configuration
|
||||
variables = config.get('variables', [])
|
||||
max_null_ratio = config.get('max_null_ratio', 0.5)
|
||||
max_null_count = config.get('max_null_count', None)
|
||||
|
||||
# Check if variables exist in data
|
||||
if not variables:
|
||||
return False # No variables specified, pass filter
|
||||
|
||||
# Validate each specified variable
|
||||
for variable in variables:
|
||||
if variable not in data.columns:
|
||||
continue # Skip variables that don't exist in data
|
||||
|
||||
# Calculate null value statistics
|
||||
null_count = data[variable].isnull().sum()
|
||||
total_count = len(data[variable])
|
||||
null_ratio = null_count / total_count if total_count > 0 else 0.0
|
||||
|
||||
# Check ratio threshold
|
||||
if null_ratio > max_null_ratio:
|
||||
return True
|
||||
|
||||
# Check count threshold (if specified)
|
||||
if max_null_count is not None and null_count > max_null_count:
|
||||
return True
|
||||
|
||||
# All variables pass null value checks
|
||||
return False
|
||||
return data.empty
|
||||
|
||||
Reference in New Issue
Block a user