"""BusinessObjects dataprovider management.
A dataprovider is the data source feeding a Web Intelligence document. It holds
a query with its dimensions, measures and parameters.
"""
from typing import List, Dict, TYPE_CHECKING
import json
if TYPE_CHECKING:
from .client import BOAPIClient
def _as_bool(value) -> bool:
"""Convert an API value (str, bool or int) to a boolean.
:param value: Value to convert
:return: Boolean value
:rtype: bool
"""
if isinstance(value, str):
return value.strip().lower() in ("true", "1", "yes", "oui")
return bool(value)
[docs]
class DataProvidersManager:
"""Manager for the dataproviders of a document.
Dataproviders represent the data sources with their queries, dimensions,
measures and filter parameters.
"""
def __init__(self, client: 'BOAPIClient'):
"""Initialize the manager.
:param client: BO client instance
:type client: BOAPIClient
"""
self.client = client
[docs]
def get_dataproviders(self, doc_id: str) -> List[Dict]:
"""Return the list of dataproviders of a document.
:param doc_id: Document identifier
:type doc_id: str
:return: List of dataproviders with id, name, rowCount
:rtype: List[Dict]
:raises Exception: If the API returns an error other than 404
"""
url = f"{self.client.base_url}/raylight/v1/documents/{doc_id}/dataproviders"
response = self.client.session.get(
url,
headers=self.client.get_headers(use_quotes=False)
)
if response.status_code == 200:
data = response.json()
dataproviders = data.get("dataproviders", {}).get("dataprovider", [])
if not isinstance(dataproviders, list):
dataproviders = [dataproviders]
return dataproviders
elif response.status_code == 404:
# The document has no dataproviders (this is normal)
return []
else:
raise Exception(f"Failed to fetch dataproviders: {response.status_code} - {response.text}")
[docs]
def get_dataprovider_detail(self, doc_id: str, dp_id: str) -> Dict:
"""Return the full details of a dataprovider.
:param doc_id: Document identifier
:type doc_id: str
:param dp_id: Dataprovider identifier
:type dp_id: str
:return: Full dataprovider structure
:rtype: Dict
:raises Exception: If the API returns an error
"""
url = f"{self.client.base_url}/raylight/v1/documents/{doc_id}/dataproviders/{dp_id}"
response = self.client.session.get(
url,
headers=self.client.get_headers(use_quotes=False)
)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Failed to fetch dataprovider details: {response.status_code} - {response.text}")
[docs]
def list_dataproviders(self, doc_id: str, show_details: bool = True, show_parameters: bool = False) -> List[Dict]:
"""Print the list of dataproviders of a document.
:param doc_id: Document identifier
:type doc_id: str
:param show_details: Enable detail output
:type show_details: bool
:param show_parameters: Enable parameter output
:type show_parameters: bool
:return: List of dataproviders
:rtype: List[Dict]
"""
dataproviders = self.get_dataproviders(doc_id)
if not dataproviders:
print("No dataprovider found in this document")
return []
print(f"\nAvailable dataproviders ({len(dataproviders)}):")
for dp in dataproviders:
name = dp.get('name', 'Unnamed')
dp_id = dp.get('id', 'N/A')
row_count = dp.get('rowCount', 0)
print(f" - {name}")
if show_details:
print(f" - ID: {dp_id}")
print(f" - Rows: {row_count:,}")
if show_parameters:
try:
parameters = self.get_dataprovider_parameters(doc_id, dp_id)
if parameters:
print(f" - Parameters/Filters ({len(parameters)}):")
for param in parameters:
param_name = param.get('name', 'N/A')
param_id = param.get('id', 'N/A')
print(f" - {param_name} (ID: {param_id})")
except Exception:
pass # Ignore parameter retrieval errors
return dataproviders
[docs]
def get_dataprovider_parameters(self, doc_id: str, dp_id: str) -> List[Dict]:
"""Return the parameters of a dataprovider.
:param doc_id: Document identifier
:type doc_id: str
:param dp_id: Dataprovider identifier
:type dp_id: str
:return: List of parameters with their metadata
:rtype: List[Dict]
"""
url = f"{self.client.base_url}/raylight/v1/documents/{doc_id}/dataproviders/{dp_id}/parameters"
response = self.client.session.get(
url,
headers=self.client.get_headers(use_quotes=False)
)
if response.status_code == 200:
data = response.json()
parameters = data.get("parameters", {}).get("parameter", [])
if not isinstance(parameters, list):
parameters = [parameters] if parameters else []
return parameters
elif response.status_code == 404:
# No parameters (this is normal)
return []
else:
raise Exception(f"Failed to fetch parameters: {response.status_code} - {response.text}")
[docs]
def build_lov_map(self, doc_id: str, dp_id: str) -> Dict[str, Dict[str, str]]:
"""Build a mapping from LOV values to their identifiers.
LOV (List of Values) parameters require a mapping between text values
and their internal identifiers.
:param doc_id: Document identifier
:type doc_id: str
:param dp_id: Dataprovider identifier
:type dp_id: str
:return: Mapping {param_id: {value: lov_id}}
:rtype: Dict[str, Dict[str, str]]
Example::
{"2": {"IDFR": "67", "AURA": "59"}}
"""
parameters = self.get_dataprovider_parameters(doc_id, dp_id)
lov_map = {}
for param in parameters:
param_id = str(param.get('id', ''))
# Check whether the parameter has a LOV
if 'answer' in param and 'info' in param['answer']:
info = param['answer']['info']
if 'lov' in info and 'cvalues' in info['lov']:
value_to_id = {}
cvalues = info['lov']['cvalues'].get('cvalue', [])
if not isinstance(cvalues, list):
cvalues = [cvalues]
for cvalue in cvalues:
lov_id = cvalue.get('@id', '')
# The value is in the first column (id=0)
columns = cvalue.get('column', [])
if not isinstance(columns, list):
columns = [columns]
for col in columns:
if col.get('@id') == 0:
value = col.get('$', '')
if value:
value_to_id[value] = lov_id
break
if value_to_id:
lov_map[param_id] = value_to_id
return lov_map
[docs]
def describe_parameters(self, doc_id: str) -> Dict[str, List[Dict]]:
"""Describe the filter parameters of a document, grouped by dataprovider.
For each dataprovider, returns its parameters with their metadata: name,
internal identifier, type, cardinality, mandatory flag and, for LOV
(List of Values) parameters, the list of allowed values.
:param doc_id: Document identifier
:type doc_id: str
:return: Mapping of dataprovider name to a list of parameter descriptors
with keys name, id, type, cardinality, mandatory, values
:rtype: Dict[str, List[Dict]]
"""
description = {}
for dp in self.get_dataproviders(doc_id):
dp_name = dp.get("name", "")
dp_id = dp.get("id", "")
try:
parameters = self.get_dataprovider_parameters(doc_id, dp_id)
except Exception:
parameters = []
if not parameters:
continue
lov_map = self.build_lov_map(doc_id, dp_id)
params_info = []
for param in parameters:
param_id = str(param.get("id", ""))
values = None
if param_id in lov_map:
values = list(lov_map[param_id].keys())
mandatory = None
if "optional" in param:
mandatory = not _as_bool(param.get("optional"))
elif "mandatory" in param:
mandatory = _as_bool(param.get("mandatory"))
params_info.append({
"name": param.get("name", ""),
"id": param_id,
"type": param.get("type", param.get("answer", {}).get("@type")),
"cardinality": param.get("cardinality"),
"mandatory": mandatory,
"values": values,
})
description[dp_name] = params_info
return description
[docs]
def analyze_dataprovider(self, doc_id: str, dp_id: str, save_json: bool = False) -> Dict:
"""Run a detailed analysis of a dataprovider.
Prints the dimensions, measures and parameters. Optionally saves the
full structure to JSON.
:param doc_id: Document identifier
:type doc_id: str
:param dp_id: Dataprovider identifier
:type dp_id: str
:param save_json: Enable JSON saving
:type save_json: bool
:return: Detailed dataprovider structure
:rtype: Dict
"""
print(f"\nAnalyzing dataprovider {dp_id}")
dp_detail = self.get_dataprovider_detail(doc_id, dp_id)
if save_json:
filename = f"dataprovider_{dp_id}_detail.json"
with open(filename, "w", encoding="utf-8") as f:
json.dump(dp_detail, f, indent=2, ensure_ascii=False)
print(f"Structure saved: {filename}")
dp_info = dp_detail.get("dataprovider", {})
if "dictionary" in dp_info:
dictionary = dp_info["dictionary"]
if "dimension" in dictionary:
dimensions = dictionary["dimension"]
if not isinstance(dimensions, list):
dimensions = [dimensions]
print(f"\n Dimensions ({len(dimensions)}):")
for dim in dimensions:
print(f" - {dim.get('name', 'N/A')}")
if "measure" in dictionary:
measures = dictionary["measure"]
if not isinstance(measures, list):
measures = [measures]
print(f"\n Measures ({len(measures)}):")
for measure in measures:
print(f" - {measure.get('name', 'N/A')}")
try:
parameters = self.get_dataprovider_parameters(doc_id, dp_id)
if parameters:
print(f"\n Parameters/Filters ({len(parameters)}):")
for param in parameters:
param_name = param.get('name', 'N/A')
param_type = param.get('type', 'N/A')
param_id = param.get('id', 'N/A')
cardinality = param.get('cardinality', 'N/A')
print(f" - {param_name}")
print(f" - ID: {param_id}")
print(f" - Type: {param_type}")
print(f" - Cardinality: {cardinality}")
if 'answer' in param:
answer = param['answer']
if 'values' in answer:
values = answer['values']
if isinstance(values, dict) and 'value' in values:
value_list = values['value']
if not isinstance(value_list, list):
value_list = [value_list]
print(f" - Value(s): {', '.join(str(v) for v in value_list)}")
except Exception as e:
print(f"\n Could not retrieve parameters: {e}")
return dp_detail