Files
sientia-dataops-laborious-l…/laborious/utils/connectors_config.py
vitor-aignosi 57b0005c94 SIENTIAPDE-1712
Update MinIO configuration by changing access key in values.yaml and removing region_name references in multiple files to streamline configuration.
2026-03-20 15:06:23 -03:00

93 lines
3.7 KiB
Python

import json
from os import getenv
from typing import Any
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: 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
"""
return {
'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_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 seconds (fallback, default: 120)
Returns:
dict: OPC server configuration dictionary
"""
opc_raw = getenv('OPC_CONFIG', None)
if opc_raw:
return json.loads(opc_raw)
return {
getenv('OPC_ID', '1'): {
'id': getenv('OPC_ID', '1'),
'server_name': getenv('OPC_SERVER_NAME', 'default_server'),
'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')),
}
}
def build_minio_config() -> dict[str, Any]:
"""
Build MinIO (S3-compatible) configuration from environment variables.
Environment Variables:
MINIO_ENDPOINT: MinIO endpoint including scheme (default: http://localhost:9000)
MINIO_ACCESS_KEY: Access key (default: minioadmin)
MINIO_SECRET_KEY: Secret key (default: minioadmin)
MINIO_REGION: Region name for S3 client (default: us-east-1)
MINIO_BUCKET_DEFAULT: Default bucket for uploads (default: laborious)
MINIO_SECURE: Whether to use HTTPS (default: false)
Returns:
dict: MinIO configuration dictionary
"""
return {
'endpoint_url': getenv('MINIO_ENDPOINT_URL', 'http://localhost:9000'),
'access_key': getenv('MINIO_ACCESS_KEY', 'minioadmin'),
'secret_key': getenv('MINIO_SECRET_KEY', 'minioadmin'),
'default_bucket': getenv('MINIO_DEFAULT_BUCKET', 'laborious'),
'retention_hours': int(getenv('MINIO_RETENTION_HOURS', '24')),
'secure': getenv('MINIO_SECURE', 'false') == 'true',
}