Analytics

Analytics

Analytics(client: AlmaAPIClient)

Domain class for handling Alma Analytics API operations.

Provides access to Analytics reports through the Alma API, with support for pagination via ResumptionToken.

Initialize the Analytics domain.

Parameters:

Name	Type	Description	Default
client	AlmaAPIClient	The AlmaAPIClient instance for making HTTP requests	required

Source code in src/almaapitk/domains/analytics.py

def __init__(self, client: AlmaAPIClient):
    """
    Initialize the Analytics domain.

    Args:
        client: The AlmaAPIClient instance for making HTTP requests
    """
    self.client = client
    self.logger = client.logger

get_report_headers

get_report_headers(report_path: str) -> List[str]

Get column headers/schema for an Analytics report.

Parameters:

Name	Type	Description	Default
report_path	str	Path to the Analytics report (e.g., '/shared/University/Reports/MyReport')	required

Returns:

Type	Description
List[str]	List of column header names in order (Column0, Column1, etc.)

Raises:

Type	Description
AlmaValidationError	If report_path is empty or None
AlmaAPIError	If the API request fails

Source code in src/almaapitk/domains/analytics.py

def get_report_headers(self, report_path: str) -> List[str]:
    """
    Get column headers/schema for an Analytics report.

    Args:
        report_path: Path to the Analytics report (e.g., '/shared/University/Reports/MyReport')

    Returns:
        List of column header names in order (Column0, Column1, etc.)

    Raises:
        AlmaValidationError: If report_path is empty or None
        AlmaAPIError: If the API request fails
    """
    if not report_path:
        raise AlmaValidationError("report_path is required")

    self.logger.info(f"Fetching headers for report: {report_path}")

    # Build endpoint and params
    # Note: Alma Analytics API requires limit between 25 and 1000
    endpoint = "almaws/v1/analytics/reports"
    params = {
        "path": report_path,
        "limit": "25"  # Minimum allowed by Analytics API
    }

    # Request with JSON Accept header (Analytics API returns XML inside JSON anies field)
    response = self.client.get(
        endpoint,
        params=params,
        custom_headers={"Accept": "application/json"}
    )

    # Extract XML from JSON response
    xml_text = self._extract_xml_from_response(response)
    if not xml_text:
        return []

    return self._parse_headers_from_xml(xml_text)

fetch_report_rows

fetch_report_rows(
    report_path: str,
    limit: int = 1000,
    max_rows: Optional[int] = None,
    progress_callback: Optional[
        Callable[[int], None]
    ] = None,
) -> List[Dict[str, Any]]

Fetch rows from an Analytics report with pagination support.

Parameters:

Name	Type	Description	Default
report_path	str	Path to the Analytics report	required
limit	int	Number of rows per page request (default 1000)	1000
max_rows	Optional[int]	Maximum total rows to return (None for unlimited)	None
progress_callback	Optional[Callable[[int], None]]	Optional callable invoked once per page with the cumulative row count fetched so far. Exceptions raised by the callback are caught and logged; they do not abort the fetch.	None

Returns:

Type	Description
List[Dict[str, Any]]	List of row dictionaries with Column0, Column1, etc. as keys

Raises:

Type	Description
AlmaValidationError	If report_path is empty or None
AlmaAPIError	If the API request fails

Source code in src/almaapitk/domains/analytics.py

def fetch_report_rows(
    self,
    report_path: str,
    limit: int = 1000,
    max_rows: Optional[int] = None,
    progress_callback: Optional[Callable[[int], None]] = None,
) -> List[Dict[str, Any]]:
    """
    Fetch rows from an Analytics report with pagination support.

    Args:
        report_path: Path to the Analytics report
        limit: Number of rows per page request (default 1000)
        max_rows: Maximum total rows to return (None for unlimited)
        progress_callback: Optional callable invoked once per page with
            the cumulative row count fetched so far. Exceptions raised
            by the callback are caught and logged; they do not abort
            the fetch.

    Returns:
        List of row dictionaries with Column0, Column1, etc. as keys

    Raises:
        AlmaValidationError: If report_path is empty or None
        AlmaAPIError: If the API request fails
    """
    if not report_path:
        raise AlmaValidationError("report_path is required")

    # Coerce numeric params before any comparison: consumers commonly supply
    # limit/max_rows from a JSON config or CLI args (strings), and a float
    # would reach Alma as an invalid token like "1000.0". Normalise to int
    # with a clear error instead of a raw TypeError (issue #177).
    try:
        limit = int(limit)
    except (TypeError, ValueError):
        raise AlmaValidationError(f"limit must be an integer, got {limit!r}")

    if max_rows is not None:
        try:
            max_rows = int(max_rows)
        except (TypeError, ValueError):
            raise AlmaValidationError(
                f"max_rows must be an integer or None, got {max_rows!r}"
            )

    if limit < 25:
        raise AlmaValidationError("limit must be at least 25 (Alma Analytics API minimum)")

    if limit > 1000:
        raise AlmaValidationError("limit must be at most 1000 (Alma Analytics API maximum)")

    if max_rows is not None and max_rows < 0:
        raise AlmaValidationError("max_rows cannot be negative")

    if max_rows == 0:
        return []

    self.logger.info(f"Fetching rows from report: {report_path}")

    all_rows = []
    resumption_token = None
    is_finished = False

    while not is_finished:
        # Check if we've reached max_rows
        if max_rows is not None and len(all_rows) >= max_rows:
            all_rows = all_rows[:max_rows]
            break

        # Build request
        endpoint = "almaws/v1/analytics/reports"
        params = {
            "path": report_path,
            "limit": str(limit)
        }

        if resumption_token:
            params["token"] = resumption_token

        # Make request with JSON Accept header (Analytics API returns XML inside JSON anies field)
        response = self.client.get(
            endpoint,
            params=params,
            custom_headers={"Accept": "application/json"}
        )

        # Extract XML from JSON response
        xml_text = self._extract_xml_from_response(response)
        if not xml_text:
            break
        rows, resumption_token, is_finished = self._parse_rows_from_xml(xml_text)

        all_rows.extend(rows)

        self.logger.debug(f"Fetched {len(rows)} rows, total: {len(all_rows)}, finished: {is_finished}")

        if progress_callback is not None:
            try:
                progress_callback(len(all_rows))
            except Exception:
                self.logger.warning("progress_callback raised; continuing fetch", exc_info=True)

        # Break if no more pages
        if not resumption_token or is_finished:
            break

    # Apply max_rows limit
    if max_rows is not None and len(all_rows) > max_rows:
        all_rows = all_rows[:max_rows]

    self.logger.info(f"Total rows fetched: {len(all_rows)}")
    return all_rows