"""Refresh of BusinessObjects dataproviders by re-running their queries with the applied parameters and filters."""
import os
from typing import List, Optional, Dict, TYPE_CHECKING
if TYPE_CHECKING:
from .client import BOAPIClient
[docs]
class RefreshManager:
"""Manager for dataprovider refresh.
Re-runs dataprovider queries with applied filter parameters.
"""
def __init__(self, client: 'BOAPIClient'):
"""Initialize the manager.
:param client: BO client instance
:type client: BOAPIClient
"""
self.client = client
[docs]
def refresh_dataprovider(self, doc_id: str, dp_id: str, parameters: Optional[Dict] = None):
"""Refresh a dataprovider with optional parameters.
Re-runs the dataprovider query applying the given filter parameters.
LOV parameters are handled automatically.
:param doc_id: Document identifier
:type doc_id: str
:param dp_id: Dataprovider identifier
:type dp_id: str
:param parameters: Filter parameters as {"param_id": ["value1", "value2"]}
:type parameters: Optional[Dict]
:raises Exception: If the refresh fails
"""
put_url = f"{self.client.base_url}/raylight/v1/documents/{doc_id}/dataproviders/{dp_id}/parameters"
if parameters:
# LOV mapping (value -> @id) for this dataprovider
lov_map = self.client.dataproviders.build_lov_map(doc_id, dp_id)
# Parameter metadata, used to resolve value types
param_metadata = self.client.dataproviders.get_dataprovider_parameters(doc_id, dp_id)
param_types = {}
for meta in param_metadata:
meta_id = str(meta.get('id', ''))
if 'answer' in meta:
param_types[meta_id] = meta['answer'].get('@type', 'String')
params_list = []
for param_id, values in parameters.items():
if not isinstance(values, list):
values = [values]
param_type = param_types.get(param_id, 'String')
formatted_values = []
for value in values:
if isinstance(value, dict):
# Already in the expected format
formatted_values.append(value)
continue
value_str = str(value)
if param_id in lov_map and value_str in lov_map[param_id]:
# LOV parameter: use the full format with the value id
formatted_values.append({
"@type": "String",
"@id": lov_map[param_id][value_str],
"$": value_str
})
elif param_type == "DateTime":
formatted_values.append({
"@type": "DateTime",
"$": value_str
})
else:
formatted_values.append({
"@type": param_type,
"$": value_str
})
params_list.append({
"id": param_id,
"answer": {
"values": {
"value": formatted_values
}
}
})
put_data = {
"parameters": {
"parameter": params_list
}
}
else:
# No parameters: plain refresh
put_data = {"parameters": {}}
if os.getenv('BO_DEBUG') == '1':
import json
print(f"DEBUG - PUT URL: {put_url}")
print(f"DEBUG - Payload: {json.dumps(put_data, indent=2)}")
response = self.client.session.put(
put_url,
json=put_data,
headers=self.client.get_headers(use_quotes=False)
)
if response.status_code not in [200, 204]:
import json
print(f"DEBUG - Sent payload: {json.dumps(put_data, indent=2)}")
raise Exception(f"Dataprovider refresh failed: {response.status_code} - {response.text}")
[docs]
def refresh_document(self, doc_id: str, dataprovider_names: Optional[List[str]] = None,
refresh_all: bool = False, verbose: bool = True,
parameters: Optional[Dict[str, Dict]] = None) -> Dict:
"""Refresh the dataproviders of a document.
Refreshes all dataproviders or a selection, applying per-dataprovider
parameters.
:param doc_id: Document identifier
:type doc_id: str
:param dataprovider_names: Names of the dataproviders to refresh
:type dataprovider_names: Optional[List[str]]
:param refresh_all: Refresh all dataproviders if True
:type refresh_all: bool
:param verbose: Enable detail output
:type verbose: bool
:param parameters: Per-dataprovider parameters as
{"DP_NAME": {"param_id": ["value"]}}
:type parameters: Optional[Dict[str, Dict]]
:return: Dictionary with success, failed and skipped lists
:rtype: Dict
"""
if verbose:
print(f"Refreshing document {doc_id}")
dataproviders = self.client.dataproviders.get_dataproviders(doc_id)
if not dataproviders:
print("No dataprovider to refresh")
return {"success": [], "failed": [], "skipped": []}
if verbose:
print(f"{len(dataproviders)} dataprovider(s) found")
results = {
"success": [],
"failed": [],
"skipped": []
}
for dp in dataproviders:
name = dp.get("name", "")
dp_id = dp.get("id", "")
row_count = dp.get("rowCount", 0)
should_refresh = refresh_all or (dataprovider_names and name in dataprovider_names)
if should_refresh:
try:
if verbose:
print(f" {name} ({row_count:,} rows)...", end=" ")
dp_parameters = None
if parameters and name in parameters:
dp_parameters = parameters[name]
if verbose:
print("[with parameters]...", end=" ")
self.refresh_dataprovider(doc_id, dp_id, dp_parameters)
if verbose:
print("done")
results["success"].append(name)
except Exception as e:
if verbose:
print(f"failed: {e}")
results["failed"].append({"name": name, "error": str(e)})
else:
results["skipped"].append(name)
if verbose:
print("\nDone:")
print(f" - Success: {len(results['success'])}")
print(f" - Failed: {len(results['failed'])}")
print(f" - Skipped: {len(results['skipped'])}")
return results