"""Main client for the BusinessObjects Web Intelligence API.
Defines the BOAPIClient class, the entry point for API operations with session
handling.
"""
import requests
import getpass
from requests.packages.urllib3.exceptions import InsecureRequestWarning
from .cache import CacheManager
from .reports import ReportsManager
from .dataproviders import DataProvidersManager
from .refresh import RefreshManager
from .export import ExportManager
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
[docs]
class BOAPIClient:
"""Client for the BusinessObjects Web Intelligence API.
Handles connection and authentication, and delegates operations to the
managers. Implements the context manager protocol to close the session on
exit.
:ivar cache: Cache manager
:ivar reports: Reports manager
:ivar dataproviders: Dataproviders manager
:ivar refresh: Refresh manager
:ivar export: Export manager
Example::
with BOAPIClient(base_url, "user") as client:
if client.connect():
df = client.get_dataframe("123456")
"""
def __init__(self, base_url: str, user: str, auth_type: str = "secEnterprise"):
"""Initialize the BusinessObjects client.
:param base_url: API URL (e.g. https://server/biprws)
:type base_url: str
:param user: User name
:type user: str
:param auth_type: Authentication type
:type auth_type: str
"""
self.base_url = base_url
self.user = user
self.auth_type = auth_type
self.token = None
# Reusable HTTP session
self.session = requests.Session()
self.session.verify = False
# Short server name (first label of the host)
hostname = base_url.replace("https://", "").replace("http://", "").split(":")[0]
self.server_name = hostname.split(".")[0]
# Initialize the managers
self.cache = CacheManager()
self.reports = ReportsManager(self)
self.dataproviders = DataProvidersManager(self)
self.refresh = RefreshManager(self)
self.export = ExportManager(self)
def __enter__(self):
"""Enter the context manager."""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""Guarantee a clean disconnection, even on error."""
self.disconnect()
[docs]
def connect(self, password: str = None) -> bool:
"""Open the connection to the BusinessObjects API.
:param password: Password, prompted interactively if None
:type password: str
:return: True if the connection succeeds
:rtype: bool
"""
if password is None:
password = getpass.getpass("Password: ")
login_data = {
"userName": self.user,
"password": password,
"auth": self.auth_type,
}
try:
response = self.session.post(
f"{self.base_url}/logon/long",
json=login_data,
headers={
"Content-Type": "application/json",
"Accept": "application/json"
},
timeout=30
)
if response.status_code == 200:
self.token = response.json()["logonToken"]
print("Connection successful")
return True
else:
print(f"Connection error: {response.status_code}")
try:
error_details = response.json()
print(f"BO error code: {error_details.get('error_code')}")
print(f"Message: {error_details.get('message')}")
except Exception:
print(f"Response: {response.text}")
return False
except requests.exceptions.RequestException as e:
print(f"Connection error: {e}")
return False
[docs]
def disconnect(self):
"""Close the session and release resources."""
if self.token:
try:
self.session.post(
f"{self.base_url}/logoff",
headers={
"X-SAP-LogonToken": f'"{self.token}"',
"Content-Type": "application/json"
}
)
print("Disconnection successful")
except Exception as e:
print(f"Error during disconnection: {e}")
finally:
self.token = None
self.session.close()
self.cache.clear()
# Main methods (delegate to the managers)
[docs]
def get_dataframe(self, doc_id: str, separator: str = None,
use_cache: bool = True, debug: bool = False, **kwargs):
"""Retrieve the data of a document as a DataFrame.
Delegates to the export manager.
:param doc_id: Document identifier
:type doc_id: str
:param separator: CSV separator, auto-detected if None
:type separator: str
:param use_cache: Enable caching
:type use_cache: bool
:param debug: Enable debug mode
:type debug: bool
:param kwargs: Extra arguments (date_format, apply_mapping, etc.)
:return: A pandas DataFrame
:rtype: pd.DataFrame
"""
return self.export.get_dataframe(doc_id, separator, use_cache, debug, **kwargs)
[docs]
def save_to_parquet(self, doc_id: str, output_path: str,
use_cache: bool = True, **kwargs):
"""Save the data to Parquet format.
:param doc_id: Document identifier
:type doc_id: str
:param output_path: Output file path
:type output_path: str
:param use_cache: Enable caching
:type use_cache: bool
:param kwargs: Arguments for to_parquet and export (chunk_size, date_format)
:return: Path of the created file
:rtype: str
"""
return self.export.save_to_parquet(doc_id, output_path, use_cache, **kwargs)
[docs]
def save_to_csv(self, doc_id: str, output_path: str, use_cache: bool = True):
"""Save the data to CSV format.
:param doc_id: Document identifier
:type doc_id: str
:param output_path: Output file path
:type output_path: str
:param use_cache: Enable caching
:type use_cache: bool
:return: Path of the created file
:rtype: str
"""
return self.export.save_to_csv(doc_id, output_path, use_cache)
[docs]
def save_to_json(self, doc_id: str, output_path: str,
use_cache: bool = True, orient: str = 'records', **kwargs):
"""Save the data to JSON format.
:param doc_id: Document identifier
:type doc_id: str
:param output_path: Output file path
:type output_path: str
:param use_cache: Enable caching
:type use_cache: bool
:param orient: JSON format
:type orient: str
:param kwargs: Extra arguments (date_format, etc.)
:return: Path of the created file
:rtype: str
"""
return self.export.save_to_json(doc_id, output_path, use_cache, orient=orient, **kwargs)
[docs]
def list_reports(self, doc_id: str, show_details: bool = True):
"""List the reports of a document.
:param doc_id: Document identifier
:type doc_id: str
:param show_details: Enable detail output
:type show_details: bool
:return: List of reports
:rtype: list
"""
return self.reports.list_reports(doc_id, show_details)
[docs]
def list_dataproviders(self, doc_id: str, show_details: bool = True):
"""List the dataproviders of a document.
:param doc_id: Document identifier
:type doc_id: str
:param show_details: Enable detail output
:type show_details: bool
:return: List of dataproviders
:rtype: list
"""
return self.dataproviders.list_dataproviders(doc_id, show_details)
[docs]
def describe_parameters(self, doc_id: str):
"""Describe the filter parameters of a document, grouped by dataprovider.
Delegates to the dataproviders manager.
: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
"""
return self.dataproviders.describe_parameters(doc_id)
[docs]
def refresh_document(self, doc_id: str, dataprovider_names=None,
refresh_all: bool = False, verbose: bool = True, parameters=None):
"""Refresh the dataproviders of a document.
:param doc_id: Document identifier
:type doc_id: str
:param dataprovider_names: Names of the dataproviders to refresh
:type dataprovider_names: list
:param refresh_all: Refresh all if True
:type refresh_all: bool
:param verbose: Enable detail output
:type verbose: bool
:param parameters: Per-dataprovider parameters
:type parameters: dict
:return: Results with success, failed, skipped
:rtype: dict
"""
return self.refresh.refresh_document(doc_id, dataprovider_names, refresh_all, verbose, parameters)
[docs]
def inspect_document(self, doc_id: str):
"""Print all the information of a document.
:param doc_id: Document identifier
:type doc_id: str
"""
print("=" * 70)
print(f"DOCUMENT INSPECTION {doc_id}")
print("=" * 70)
print("\n" + "-" * 70)
print("REPORTS")
print("-" * 70)
reports = self.list_reports(doc_id, show_details=True)
print("\n" + "-" * 70)
print("DATAPROVIDERS")
print("-" * 70)
dataproviders = self.list_dataproviders(doc_id, show_details=True)
print("\n" + "=" * 70)
print("SUMMARY")
print("=" * 70)
print(f" - Reports: {len(reports)}")
print(f" - Dataproviders: {len(dataproviders)}")
total_rows = sum(dp.get('rowCount', 0) for dp in dataproviders)
if total_rows > 0:
print(f" - Total rows: {total_rows:,}")
print(f"\nUsage: client.get_dataframe('{doc_id}')")