SIENTIAPDE-1478
Refactor PIWebAPIClient and Update Configuration Imports - Removed outdated PostgreSQL, Redis, MongoDB, and API configuration functions from connectors_config.py. - Deleted the pi_web_api_client.py file as part of the refactor. - Updated imports in worker.py and api.py to use the new repository structure. - Cleaned up scenarios.md by removing obsolete scenarios related to unique constraint violations. - Commented out the specific version of the sientia-dataops-library in requirements.txt for flexibility.
This commit is contained in:
vitor-aignosi
@@ -10,8 +10,7 @@ with workflow.unsafe.imports_passed_through():
|
||||
from sientia_do.observability.metrics_controller import MetricsController
|
||||
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
||||
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
|
||||
|
||||
from scouter.utils.clients.pi_web_api_client import PIWebAPIClient
|
||||
from sientia_do.repository.pi_web_api_client import PIWebAPIClient
|
||||
|
||||
|
||||
class API(SientiaMonitoring):
|
||||
|
||||
@@ -23,24 +23,4 @@ TAG_CHANGES_MONITOR = Gauge(
|
||||
'scouter_tag_changes_monitor',
|
||||
'Current value change of each tag',
|
||||
[*CORE_LABELS, 'tag_name'],
|
||||
)
|
||||
|
||||
|
||||
# Generic REST client metrics
|
||||
GENERIC_REST_CLIENT_LAG = Histogram(
|
||||
'scouter_generic_rest_client_lag',
|
||||
'Lag time for a request to a generic REST client',
|
||||
SIENTIA_CORE_LABELS,
|
||||
)
|
||||
|
||||
GENERIC_REST_READ_COUNT = Counter(
|
||||
'scouter_generic_rest_client_read_count',
|
||||
'Number of reads from a generic REST client',
|
||||
SIENTIA_CORE_LABELS,
|
||||
)
|
||||
|
||||
GENERIC_REST_READ_ERROR_COUNT = Counter(
|
||||
'scouter_generic_rest_client_read_error_count',
|
||||
'Number of read errors from a generic REST client',
|
||||
SIENTIA_CORE_LABELS,
|
||||
)
|
||||
)
|
||||
@@ -1,340 +0,0 @@
|
||||
import io
|
||||
import json
|
||||
import time
|
||||
import warnings
|
||||
from typing import Any
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import pandas as pd
|
||||
import pycurl
|
||||
from sientia_do.notifications.handlers import NotificationHandler
|
||||
from sientia_do.observability.logger import Logger
|
||||
from sientia_do.observability.metrics_controller import MetricsController
|
||||
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
||||
|
||||
from scouter import metrics
|
||||
|
||||
warnings.simplefilter('ignore') # Ignore warnings such as 'verify=False'
|
||||
|
||||
|
||||
class PIMSRequestError(Exception):
|
||||
"""Generic error for failed requests to PI Web API using pycurl."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class PIWebAPIClient(SientiaMonitoring):
|
||||
"""
|
||||
Client for interacting with the PI Web API.
|
||||
|
||||
This class provides a robust interface for querying historical and real-time
|
||||
data from OSIsoft PI systems through the PI Web API. It implements:
|
||||
- Asynchronous HTTP requests using pycurl
|
||||
- Authentication support (Basic and Bearer)
|
||||
- Automatic data normalization and timestamp handling
|
||||
- Comprehensive error handling and monitoring
|
||||
- Metrics collection for observability
|
||||
|
||||
The client is designed for high-performance data retrieval with proper
|
||||
connection management and error recovery mechanisms.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str,
|
||||
auth_config: dict[str, Any],
|
||||
logger: Logger,
|
||||
notification_handler: NotificationHandler,
|
||||
metrics_controller: MetricsController,
|
||||
headers_config: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Initialize the PI Web API client with connection parameters.
|
||||
|
||||
Args:
|
||||
base_url (str): Base URL of the PI Web API server
|
||||
auth_config (dict[str, Any]): Authentication configuration.
|
||||
Required fields:
|
||||
- type (str): Authentication type ('basic' or 'bearer')
|
||||
- token (str): Authentication token
|
||||
logger (Logger): Logger instance for operation logging
|
||||
notification_handler (NotificationHandler): Handler for system notifications
|
||||
metrics_controller (MetricsController): Controller for metrics collection
|
||||
headers_config (dict[str, Any], optional): HTTP headers configuration.
|
||||
Default headers include content-type, accept, and x-requested-with
|
||||
max_concurrency (int, optional): Maximum number of concurrent requests. Defaults to 8
|
||||
"""
|
||||
if headers_config is None:
|
||||
headers_config = {
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json',
|
||||
'x-requested-with': 'XMLHttpRequest',
|
||||
}
|
||||
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
|
||||
|
||||
self.base_url = base_url.rstrip('/')
|
||||
|
||||
# auth_config spec:
|
||||
# 'type': 'basic' or 'bearer',
|
||||
# 'token': 'token',
|
||||
self.auth_config = auth_config
|
||||
self.auth_config['type'] = self.auth_config['type'].lower()
|
||||
self.headers: dict[str, str] = headers_config
|
||||
|
||||
self.authenticate()
|
||||
|
||||
def close(self) -> None:
|
||||
"""
|
||||
Close the client and shutdown monitoring services.
|
||||
"""
|
||||
SientiaMonitoring.shutdown(self)
|
||||
|
||||
def _to_clean_timestamp(self, series: pd.Series) -> pd.Series:
|
||||
"""
|
||||
Convert a Series of timestamps to datetime, UTC, and round to the nearest second.
|
||||
|
||||
Args:
|
||||
series (pd.Series): Series containing timestamp values
|
||||
|
||||
Returns:
|
||||
pd.Series: Cleaned timestamp series in UTC, floored to seconds
|
||||
"""
|
||||
series = pd.to_datetime(series, utc=True, errors='coerce')
|
||||
return series.dt.floor('s')
|
||||
|
||||
def _extract_numeric(self, value: Any) -> float | None:
|
||||
"""
|
||||
Normalize a value (potentially nested) to float.
|
||||
|
||||
This method handles PI Web API response values that may be nested
|
||||
in dictionaries or other structures, extracting the numeric value.
|
||||
|
||||
Args:
|
||||
value (Any): Value to extract and normalize
|
||||
|
||||
Returns:
|
||||
float | None: Numeric value as float, or None if conversion fails
|
||||
"""
|
||||
if isinstance(value, dict):
|
||||
value = value.get('Value', value)
|
||||
return pd.to_numeric(value, errors='coerce')
|
||||
|
||||
def authenticate(self):
|
||||
"""
|
||||
Configure authentication headers based on auth_config.
|
||||
|
||||
This method sets up the Authorization header using either Basic or Bearer
|
||||
authentication based on the configured authentication type.
|
||||
|
||||
Raises:
|
||||
ValueError: If authentication type is not 'basic' or 'bearer'
|
||||
"""
|
||||
self.logger.info(f'Authenticating with {self.auth_config["type"]} authentication')
|
||||
|
||||
if self.auth_config['type'] == 'basic':
|
||||
self.headers['Authorization'] = f'Basic {self.auth_config["token"]}'
|
||||
elif self.auth_config['type'] == 'bearer':
|
||||
self.headers['Authorization'] = f'Bearer {self.auth_config["token"]}'
|
||||
else:
|
||||
raise ValueError(f'Invalid authentication type: {self.auth_config["type"]}')
|
||||
|
||||
async def _curl_get_json(
|
||||
self,
|
||||
url: str,
|
||||
params: list[tuple[str, str]] | None = None,
|
||||
timeout: int = 30,
|
||||
verify: bool = True,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Perform a GET request using pycurl and return the decoded JSON response.
|
||||
|
||||
This method executes an asynchronous HTTP GET request with proper error handling,
|
||||
metrics collection, and timeout management. It automatically tracks request
|
||||
latency and emits monitoring metrics.
|
||||
|
||||
Args:
|
||||
url (str): Target URL for the GET request
|
||||
params (list[tuple[str, str]], optional): Query parameters as list of tuples.
|
||||
Each tuple contains (parameter_name, parameter_value)
|
||||
timeout (int, optional): Request timeout in seconds. Defaults to 30
|
||||
verify (bool, optional): Verify SSL certificates. Defaults to True
|
||||
metadata (dict[str, Any], optional): Workflow execution metadata for tracking
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Parsed JSON response body
|
||||
|
||||
Raises:
|
||||
PIMSRequestError: If HTTP error, connection error, or JSON parsing error occurs
|
||||
"""
|
||||
if metadata is None:
|
||||
metadata = {}
|
||||
|
||||
buffer = io.BytesIO()
|
||||
c = pycurl.Curl()
|
||||
|
||||
core_labels = self.get_core_labels(metadata=metadata, operation_type='get_json')
|
||||
|
||||
try:
|
||||
if params:
|
||||
query_string = urlencode(params, doseq=True)
|
||||
full_url = f'{url}?{query_string}'
|
||||
else:
|
||||
full_url = url
|
||||
|
||||
c.setopt(pycurl.URL, full_url.encode('utf-8'))
|
||||
c.setopt(pycurl.WRITEDATA, buffer)
|
||||
|
||||
# Configure HTTP headers
|
||||
header_list = [f'{k}: {v}' for k, v in self.headers.items()]
|
||||
if header_list:
|
||||
c.setopt(pycurl.HTTPHEADER, header_list)
|
||||
|
||||
# Set request timeout
|
||||
c.setopt(pycurl.TIMEOUT, timeout)
|
||||
|
||||
# Configure SSL verification
|
||||
if not verify:
|
||||
c.setopt(pycurl.SSL_VERIFYPEER, 0)
|
||||
c.setopt(pycurl.SSL_VERIFYHOST, 0)
|
||||
|
||||
start_time = time.time()
|
||||
try:
|
||||
c.perform()
|
||||
except Exception as e:
|
||||
await self.emit_metric(
|
||||
metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT,
|
||||
tags=core_labels,
|
||||
)
|
||||
raise e
|
||||
await self.observe_lag(
|
||||
start_time=start_time,
|
||||
metric_object=metrics.GENERIC_REST_CLIENT_LAG,
|
||||
tags=core_labels,
|
||||
)
|
||||
|
||||
status_code = c.getinfo(pycurl.RESPONSE_CODE)
|
||||
body = buffer.getvalue().decode('utf-8', errors='replace')
|
||||
|
||||
if status_code >= 400:
|
||||
await self.emit_metric(
|
||||
metric_object=metrics.GENERIC_REST_READ_ERROR_COUNT,
|
||||
tags=core_labels,
|
||||
)
|
||||
raise PIMSRequestError(f"HTTP {status_code} calling '{full_url}': {body[:200]}")
|
||||
|
||||
await self.emit_metric(
|
||||
metric_object=metrics.GENERIC_REST_READ_COUNT,
|
||||
tags=core_labels,
|
||||
)
|
||||
try:
|
||||
return json.loads(body)
|
||||
except json.JSONDecodeError as e:
|
||||
raise PIMSRequestError(
|
||||
f"Error decoding JSON response from '{full_url}': {e}; body: {body[:200]}"
|
||||
) from e
|
||||
|
||||
except pycurl.error as e:
|
||||
raise PIMSRequestError(f"Connection error calling '{url}': {e}") from e
|
||||
finally:
|
||||
c.close()
|
||||
|
||||
async def get_latest_values_df(
|
||||
self,
|
||||
web_ids: dict[str, dict[str, str]],
|
||||
endpoint: str,
|
||||
timeout: int = 30,
|
||||
start_time: str = '*-1d',
|
||||
end_time: str = '*',
|
||||
max_count: int | None = 1,
|
||||
metadata: dict[str, Any] | None = None,
|
||||
) -> pd.DataFrame:
|
||||
"""
|
||||
Retrieve historical values for multiple WebIds using PI Web API streamsets.
|
||||
|
||||
This method queries the PI Web API's /streamsets/recorded endpoint to fetch
|
||||
historical data for multiple tags simultaneously. It returns a normalized
|
||||
DataFrame with timestamps, tag names, values, and WebIds.
|
||||
|
||||
Args:
|
||||
web_ids (dict[str, str | None]): Dictionary mapping tag names to their WebIds.
|
||||
None values are filtered out before querying
|
||||
endpoint (str): PI Web API endpoint path (e.g., '/streamsets/recorded')
|
||||
timeout (int, optional): Request timeout in seconds. Defaults to 30
|
||||
start_time (str, optional): Start time in PI Web API format (e.g., "*-50d").
|
||||
Defaults to "*-1d" (1 day ago)
|
||||
end_time (str, optional): End time in PI Web API format (e.g., "*").
|
||||
Defaults to "*" (current time)
|
||||
max_count (int, optional): Maximum number of data points per series.
|
||||
Defaults to 1. If None, maxCount parameter is not sent
|
||||
metadata (dict[str, Any], optional): Workflow execution metadata for tracking
|
||||
|
||||
Returns:
|
||||
pd.DataFrame: DataFrame with columns:
|
||||
- timestamp: Cleaned timestamp (UTC, floored to seconds)
|
||||
- name: Tag name
|
||||
- value: Numeric value (normalized)
|
||||
- tag: WebId of the tag
|
||||
Returns empty DataFrame if no data is found
|
||||
|
||||
Raises:
|
||||
PIMSRequestError: If API request fails or returns invalid data
|
||||
"""
|
||||
if metadata is None:
|
||||
metadata = {}
|
||||
|
||||
url = f'{self.base_url}{endpoint}'
|
||||
|
||||
# Build query parameters with WebIds (filtering out None values)
|
||||
params: list[tuple[str, str]] = [('webid', web_id['webid']) for web_id in web_ids.values()]
|
||||
# Invert startTime and endTime to get descending order (most recent first)
|
||||
# PI Web API returns descending order when endTime < startTime
|
||||
params.extend(
|
||||
[
|
||||
('startTime', end_time), # Use end_time as startTime (inverted)
|
||||
('endtime', start_time), # Use start_time as endTime (inverted)
|
||||
('selectedFields', 'Items.Name;Items.Items.Timestamp;Items.Items.Value'),
|
||||
]
|
||||
)
|
||||
|
||||
if max_count is not None:
|
||||
params.append(('maxCount', str(max_count)))
|
||||
|
||||
data = await self._curl_get_json(
|
||||
url=url,
|
||||
params=params,
|
||||
timeout=timeout,
|
||||
verify=False,
|
||||
metadata=metadata,
|
||||
)
|
||||
|
||||
self.debug(f'Raw data from PI Web API: {data}', metadata=metadata)
|
||||
|
||||
raw_data = data.get('Items', [])
|
||||
|
||||
records = []
|
||||
for entry in raw_data:
|
||||
tag_name = entry.get('Name')
|
||||
series_items = entry.get('Items', [])
|
||||
for it in series_items:
|
||||
if isinstance(it, dict) and 'Timestamp' in it and 'Value' in it:
|
||||
ts = it.get('Timestamp')
|
||||
val = it.get('Value')
|
||||
web_id = web_ids[tag_name]['webid']
|
||||
if ts is not None:
|
||||
records.append(
|
||||
{
|
||||
'timestamp': ts,
|
||||
'name': tag_name,
|
||||
'value': self._extract_numeric(val),
|
||||
'tag': web_id,
|
||||
}
|
||||
)
|
||||
|
||||
if not records:
|
||||
return pd.DataFrame()
|
||||
|
||||
df = pd.DataFrame.from_records(records)
|
||||
df['timestamp'] = self._to_clean_timestamp(df['timestamp'])
|
||||
|
||||
return df
|
||||
@@ -2,31 +2,6 @@ from os import getenv
|
||||
from typing import Any
|
||||
|
||||
|
||||
def build_postgres_config() -> dict[str, Any]:
|
||||
"""
|
||||
Build PostgreSQL connection configuration from environment variables.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: PostgreSQL configuration dictionary with keys:
|
||||
- host: Database hostname (default: localhost)
|
||||
- port: Database port (default: 5432)
|
||||
- user: Database username (default: sientia)
|
||||
- password: Database password (default: sientia)
|
||||
- dbname: Database name (default: sientia)
|
||||
- min_connections: Minimum connection pool size (default: 5)
|
||||
- max_connections: Maximum connection pool size (default: 20)
|
||||
"""
|
||||
return {
|
||||
'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_kafka_config() -> dict[str, Any]:
|
||||
"""
|
||||
Build Kafka configuration from environment variables.
|
||||
@@ -42,74 +17,3 @@ def build_kafka_config() -> dict[str, Any]:
|
||||
'polling_time': int(getenv('KAFKA_POLLING_TIME', '1000')),
|
||||
'group_id': 'scouter-group',
|
||||
}
|
||||
|
||||
|
||||
def build_redis_config() -> dict[str, Any]:
|
||||
"""
|
||||
Build Redis connection configuration from environment variables.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Redis configuration dictionary with keys:
|
||||
- host: Redis server hostname (default: localhost)
|
||||
- port: Redis server port (default: 6379)
|
||||
- username: Redis authentication username (default: None)
|
||||
- password: Redis authentication password (default: None)
|
||||
"""
|
||||
return {
|
||||
'host': getenv('REDIS_HOST', 'localhost'),
|
||||
'port': int(getenv('REDIS_PORT', '6379')),
|
||||
'username': getenv('REDIS_USERNAME', None),
|
||||
'password': getenv('REDIS_PASSWORD', None),
|
||||
}
|
||||
|
||||
|
||||
def build_mongodb_config() -> dict[str, Any]:
|
||||
"""
|
||||
Build MongoDB connection configuration from environment variables.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: MongoDB configuration dictionary with keys:
|
||||
- connection_string: Complete MongoDB connection URI
|
||||
- database_name: Target database name (default: sientia)
|
||||
"""
|
||||
username = getenv('MONGODB_USERNAME', 'sientia')
|
||||
password = getenv('MONGODB_PASSWORD', 'sientia')
|
||||
uri = getenv('MONGODB_URL', 'localhost:27017')
|
||||
|
||||
connection_string = f'mongodb://{username}:{password}@{uri}'
|
||||
return {
|
||||
'connection_string': connection_string,
|
||||
'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
|
||||
}
|
||||
|
||||
|
||||
def build_api_config() -> dict[str, Any]:
|
||||
"""
|
||||
Build API connection configuration from environment variables.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: API configuration dictionary with keys:
|
||||
- base_url: API base URL (default: https://pi.example.com)
|
||||
- auth_type: API authentication type (default: basic)
|
||||
- auth_token: API authentication token (default: None)
|
||||
"""
|
||||
return {
|
||||
'base_url': getenv('PI_WEB_API_BASE_URL', 'https://pi.example.com'),
|
||||
'auth_type': getenv('PI_WEB_API_AUTH_TYPE', 'basic'),
|
||||
'auth_token': getenv('PI_WEB_API_AUTH_TOKEN', None),
|
||||
}
|
||||
|
||||
|
||||
def build_druid_config() -> dict[str, Any]:
|
||||
"""
|
||||
Build Apache Druid connection configuration from environment variables.
|
||||
|
||||
Returns:
|
||||
dict[str, Any]: Druid configuration dictionary with keys:
|
||||
- host: Druid server hostname (default: localhost)
|
||||
- port: Druid server port (default: 8082)
|
||||
"""
|
||||
return {
|
||||
'host': getenv('DRUID_HOST', 'localhost'),
|
||||
'port': int(getenv('DRUID_PORT', '8082')),
|
||||
}
|
||||
|
||||
@@ -14,7 +14,7 @@ with workflow.unsafe.imports_passed_through():
|
||||
|
||||
from scouter import metrics
|
||||
from scouter.activities.activities import Activities
|
||||
from scouter.utils.connectors_config import (
|
||||
from sientia_do.connectors_config import (
|
||||
build_api_config,
|
||||
build_mongodb_config,
|
||||
build_postgres_config,
|
||||
|
||||
@@ -109,9 +109,7 @@ class CoreScouter:
|
||||
'schema': input_data['schema'],
|
||||
'table_name': input_data['table_name'],
|
||||
'data': held_data,
|
||||
'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ},
|
||||
'on_conflict': 'ignore',
|
||||
'unique_columns': ['model_id', 'timestamp', 'variable'],
|
||||
'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ}
|
||||
},
|
||||
retry_policy=retry_policy,
|
||||
start_to_close_timeout=timedelta(seconds=60),
|
||||
|
||||
Reference in New Issue
Block a user