"""Functions to query a BusinessObjects document and export it to a file.
Credentials are read from ``.env``. The output format is taken from the file
extension (``.parquet``, ``.csv``, ``.json``).
"""
from typing import Any, Dict, List, Optional
from .client import BOAPIClient
from .credentials import load_bo_config
from .config import load_export_config
def _connect_from_env(env: Optional[str] = None) -> BOAPIClient:
"""Build and connect a client from ``.env`` variables.
:param env: Optional environment suffix (e.g. 'dev', 'prod')
:type env: Optional[str]
:return: Connected client
:rtype: BOAPIClient
:raises ValueError: If credentials are incomplete
:raises ConnectionError: If the connection fails
"""
config = load_bo_config(env)
missing = [k for k in ("base_url", "user", "password") if not config.get(k)]
if missing:
raise ValueError(
"Incomplete BO credentials in .env. "
f"Missing variables: {', '.join('BO_' + m.upper() for m in missing)}"
)
client = BOAPIClient(config["base_url"], config["user"], config["auth_type"])
if not client.connect(config["password"]):
client.disconnect()
raise ConnectionError("Failed to connect to BusinessObjects")
return client
def _is_structured_params(params: Dict) -> bool:
"""Return whether ``params`` already uses the nested API format.
The nested format is ``{dataprovider_name: {parameter_id: [values]}}``.
:param params: Parameters mapping
:type params: Dict
:return: True if every value is a dictionary
:rtype: bool
"""
return bool(params) and all(isinstance(v, dict) for v in params.values())
def _build_refresh_parameters(client: BOAPIClient, doc_id: str,
params: Dict[str, Any]) -> Dict[str, Dict[str, List[str]]]:
"""Convert flat parameters to the nested API format.
Input ``{parameter_name: value}`` is converted to
``{dataprovider_name: {parameter_id: [values]}}``. Names are resolved with
:meth:`BOAPIClient.describe_parameters`. A mapping already in the nested
format is returned unchanged.
:param client: Connected client
:type client: BOAPIClient
:param doc_id: Document identifier
:type doc_id: str
:param params: Flat parameters, or the nested format
:type params: Dict[str, Any]
:return: Parameters in the nested format used by refresh_document
:rtype: Dict[str, Dict[str, List[str]]]
:raises ValueError: If a parameter name is not found in the document
"""
if _is_structured_params(params):
return params
description = client.describe_parameters(doc_id)
index: Dict[str, List] = {}
for dp_name, param_list in description.items():
for param in param_list:
index.setdefault(param["name"].lower(), []).append((dp_name, param["id"]))
structured: Dict[str, Dict[str, List[str]]] = {}
for name, value in params.items():
matches = index.get(str(name).lower())
if not matches:
available = sorted({p["name"] for lst in description.values() for p in lst})
raise ValueError(
f"Parameter '{name}' not found in document {doc_id}. "
f"Available parameters: {', '.join(available) or 'none'}"
)
values = value if isinstance(value, (list, tuple)) else [value]
values = [str(v) for v in values]
for dp_name, param_id in matches:
structured.setdefault(dp_name, {})[param_id] = values
return structured
[docs]
def describe_document_parameters(doc_id: str, env: Optional[str] = None) -> Dict[str, List[Dict]]:
"""Describe the filter parameters of a document.
Connects using ``.env`` credentials. Returns the parameters per dataprovider.
:param doc_id: Document identifier
:type doc_id: str
:param env: Optional environment suffix for ``.env``
:type env: Optional[str]
:return: Mapping of dataprovider name to a list of parameter descriptors
(name, id, type, cardinality, mandatory, values)
:rtype: Dict[str, List[Dict]]
"""
with _connect_from_env(env) as client:
return client.describe_parameters(doc_id)
def _document_columns(client: BOAPIClient, doc_id: str) -> List[str]:
"""Return the column names of a document from its dataprovider dictionary.
Reads the dimensions and measures declared by each dataprovider. This does
not require the document to be refreshed and does not fetch any data.
:param client: Connected client
:type client: BOAPIClient
:param doc_id: Document identifier
:type doc_id: str
:return: Ordered list of unique column names
:rtype: List[str]
"""
names = []
for dp in client.dataproviders.get_dataproviders(doc_id):
try:
detail = client.dataproviders.get_dataprovider_detail(doc_id, dp.get('id'))
except Exception:
continue
dictionary = detail.get('dataprovider', {}).get('dictionary', {})
for key in ('dimension', 'measure'):
items = dictionary.get(key, [])
if not isinstance(items, list):
items = [items]
for item in items:
name = item.get('name')
if name and name not in names:
names.append(name)
return names
def _filter_placeholder(param_type: str) -> str:
"""Return a placeholder filter value for a parameter type.
:param param_type: Parameter type reported by the API
:type param_type: str
:return: Placeholder value to fill in the generated config
:rtype: str
"""
if param_type and 'date' in param_type.lower():
return '2026-01-01T00:00:00.000Z'
return 'CHANGE_ME'
[docs]
def generate_export_config(doc_id: str, env: Optional[str] = None,
output: Optional[str] = None,
chunk_size: int = 10000,
date_format: str = '%d/%m/%Y',
with_columns: bool = False) -> str:
"""Build a YAML export config skeleton for a document.
Reads the document parameters and returns YAML text ready to be saved.
Filters are keyed by parameter id, with the parameter name kept as a
comment and a placeholder value. DateTime parameters get a placeholder ISO
value. When ``with_columns`` is set, the column names declared by the
document dataproviders are appended as commented entries so that unlisted
columns stay under pandas inference.
:param doc_id: Document identifier
:type doc_id: str
:param env: Optional environment suffix for ``.env``
:type env: Optional[str]
:param output: Output path written under the ``output`` key, optional
:type output: Optional[str]
:param chunk_size: Value written under the ``chunk_size`` key
:type chunk_size: int
:param date_format: Value written under the ``date_format`` key
:type date_format: str
:param with_columns: List the document column names from the dataprovider
dictionary (no data fetch)
:type with_columns: bool
:return: YAML configuration text
:rtype: str
"""
with _connect_from_env(env) as client:
parameters = client.describe_parameters(doc_id)
columns = []
if with_columns:
columns = _document_columns(client, doc_id)
dp_names = list(parameters.keys())
primary_dp = dp_names[0] if dp_names else None
out_path = output or f"out/{primary_dp or doc_id}.parquet"
lines = [f"# Generated for document {doc_id}"]
if len(dp_names) > 1:
lines.append(f"# Note: several dataproviders found: {', '.join(dp_names)}")
lines.append("# Filters below are keyed by parameter id for the dataprovider set on 'dataprovider'.")
lines.append("")
if env:
lines.append("doc_id:")
lines.append(f" {env}: {doc_id}")
else:
lines.append(f"doc_id: {doc_id}")
lines.append(f"output: {out_path}")
lines.append(f"chunk_size: {chunk_size}")
if primary_dp:
lines.append(f"dataprovider: {primary_dp}")
lines.append(f'date_format: "{date_format}"')
lines.append("")
lines.append("filters:")
has_filter = False
for dp_name, params in parameters.items():
for param in params:
has_filter = True
placeholder = _filter_placeholder(param.get('type'))
name = param.get('name', '').strip()
lines.append(f" '{param.get('id')}': ['{placeholder}'] # {name} ({param.get('type')})")
if not has_filter:
lines.append(" {} # no parameter on this document")
lines.append("")
lines.append("# Columns not listed below are kept as-is (pandas inference).")
lines.append("# Uncomment and type only the ones to convert or rename.")
lines.append("columns:" if columns else "# columns:")
for col in columns:
lines.append(f"# {col}: str")
return "\n".join(lines) + "\n"
[docs]
def export_document(doc_id: str, output: str,
params: Optional[Dict[str, Any]] = None,
env: Optional[str] = None,
refresh: Optional[bool] = None,
**kwargs) -> str:
"""Query a document and export it to a file.
Connects using ``.env`` credentials. Refreshes the document when ``params``
is given. The output format is taken from the ``output`` extension
(``.parquet``, ``.csv``, ``.json``).
:param doc_id: Document identifier
:type doc_id: str
:param output: Output file path; its extension selects the format
:type output: str
:param params: Filter parameters as ``{name: value}``; triggers a refresh
:type params: Optional[Dict[str, Any]]
:param env: Optional environment suffix for ``.env``
:type env: Optional[str]
:param refresh: Force (True) or disable (False) the refresh; when None,
refresh only if ``params`` is provided
:type refresh: Optional[bool]
:param kwargs: Options forwarded to the export (e.g. ``chunk_size``,
``date_format``)
:return: Path of the created file
:rtype: str
"""
do_refresh = refresh if refresh is not None else (params is not None)
with _connect_from_env(env) as client:
if do_refresh:
structured = _build_refresh_parameters(client, doc_id, params) if params else None
client.refresh_document(doc_id=doc_id, refresh_all=True,
parameters=structured, verbose=True)
use_cache = not do_refresh
if output.endswith(".csv"):
return client.save_to_csv(doc_id, output, use_cache=use_cache)
elif output.endswith(".json"):
return client.save_to_json(doc_id, output, use_cache=use_cache, **kwargs)
return client.save_to_parquet(doc_id, output, use_cache=use_cache, **kwargs)
def _build_filter_parameters(client: BOAPIClient, doc_id: str,
filters: Dict[str, Any],
dataprovider: Optional[str]) -> Dict[str, Dict[str, List]]:
"""Wrap parameter-id filters under their dataprovider name.
Filters are keyed by BO parameter id. They are nested as
``{dataprovider_name: {parameter_id: [values]}}``. When ``dataprovider`` is
None, the filters are applied to every dataprovider of the document.
:param client: Connected client
:type client: BOAPIClient
:param doc_id: Document identifier
:type doc_id: str
:param filters: Filters keyed by parameter id
:type filters: Dict[str, Any]
:param dataprovider: Target dataprovider name, or None for all
:type dataprovider: Optional[str]
:return: Parameters in the nested format used by refresh_document
:rtype: Dict[str, Dict[str, List]]
"""
raw = {
str(pid): (value if isinstance(value, (list, tuple)) else [value])
for pid, value in filters.items()
}
if dataprovider:
return {dataprovider: raw}
names = [dp.get("name", "") for dp in client.dataproviders.get_dataproviders(doc_id)]
return {name: raw for name in names if name}
[docs]
def export_from_config(config_path, env: Optional[str] = None,
output: Optional[str] = None, **kwargs) -> str:
"""Export a document from a YAML configuration file.
Reads the doc_id, filters and column typing from the YAML, refreshes the
document when filters are present, then exports it. Column typing applies
to ``.parquet`` and ``.json`` outputs.
:param config_path: Path to the YAML configuration file
:param env: Environment name for ``.env`` and doc_id resolution
:type env: Optional[str]
:param output: Output file path; overrides the YAML ``output``
:type output: Optional[str]
:param kwargs: Extra arguments forwarded to the export
:return: Path of the created file
:rtype: str
:raises ValueError: If no output path is provided
"""
config = load_export_config(config_path, env)
doc_id = config["doc_id"]
out = output or config["output"]
if not out:
raise ValueError("No output path: set 'output' in the YAML or pass output=")
filters = config["filters"]
mapping = config["column_mapping"]
date_format = config["date_format"]
apply_mapping = mapping is not None
with _connect_from_env(env) as client:
if filters:
params = _build_filter_parameters(client, doc_id, filters, config["dataprovider"])
client.refresh_document(doc_id=doc_id, refresh_all=True,
parameters=params, verbose=True)
use_cache = not filters
if out.endswith(".csv"):
if apply_mapping:
print("Note: column typing is not applied to CSV output (raw export).")
return client.save_to_csv(doc_id, out, use_cache=use_cache)
elif out.endswith(".json"):
return client.save_to_json(doc_id, out, use_cache=use_cache,
apply_mapping=apply_mapping, mapping_file=mapping,
date_format=date_format, **kwargs)
return client.save_to_parquet(doc_id, out, use_cache=use_cache,
apply_mapping=apply_mapping, mapping_file=mapping,
chunk_size=config["chunk_size"],
date_format=date_format, **kwargs)