Utilities

TSVGenerator

TSVGenerator

TSVGenerator(config_path: str)

Config-driven TSV generator for Alma data processing.

This utility creates TSV files based on JSON configuration files, allowing for flexible column definitions and data sources.

Initialize TSV Generator with configuration.

Parameters:

Name	Type	Description	Default
config_path	str	Path to JSON configuration file	required

Source code in src/almaapitk/utils/tsv_generator.py

def __init__(self, config_path: str):
    """
    Initialize TSV Generator with configuration.

    Args:
        config_path: Path to JSON configuration file
    """
    self.config_path = config_path
    # Logger goes up first so ``_load_config`` can use ``self.logger``
    # instead of ``print()`` (issue #14).
    self.logger = get_logger("tsv_generator")
    self.config = self._load_config()
    self.alma_client = None
    self.admin_client = None

generate_tsv

generate_tsv(
    set_id_override: str = None,
    environment_override: str = None,
) -> str

Generate TSV file based on configuration.

Parameters:

Name	Type	Description	Default
set_id_override	str	Optional override for the Alma set ID	None
environment_override	str	Optional override for the environment	None

Returns:

Type	Description
str	Path to the created TSV file

Source code in src/almaapitk/utils/tsv_generator.py

def generate_tsv(self, set_id_override: str = None, environment_override: str = None) -> str:
    """
    Generate TSV file based on configuration.

    Args:
        set_id_override: Optional override for the Alma set ID
        environment_override: Optional override for the environment

    Returns:
        Path to the created TSV file
    """
    self.logger.info(f"\n=== TSV Generation Started ===")
    self.logger.info(f"Config file: {self.config_path}")

    try:
        # Step 1: Initialize Alma clients
        self._initialize_alma_clients(environment_override)

        # Step 2: Get MMS IDs from Alma set
        mms_ids = self._get_mms_ids(set_id_override)

        if not mms_ids:
            raise RuntimeError("No MMS IDs retrieved from the set")

        # Step 3: Create output directory
        output_dir = self._create_output_directory()

        # Step 4: Generate and write TSV file
        tsv_path = self._write_tsv_file(mms_ids, output_dir)

        # Step 5: Validate the created file
        self._validate_tsv_file(tsv_path)

        self.logger.info(f"\n=== TSV Generation Completed Successfully ===")
        self.logger.info(f"Output file: {tsv_path}")

        return tsv_path

    except Exception:
        # Issue #154 F-003: logger.exception attaches the traceback
        # via exc_info; do not interpolate ``{e}`` into the message.
        self.logger.exception("TSV Generation Failed")
        raise

preview_config

preview_config() -> None

Print a preview of the current configuration.

Source code in src/almaapitk/utils/tsv_generator.py

def preview_config(self) -> None:
    """Print a preview of the current configuration."""
    self.logger.info(f"\n=== Configuration Preview ===")
    self.logger.info(f"Config file: {self.config_path}")
    self.logger.info(f"Alma Set ID: {self.config['input']['alma_set_id']}")
    self.logger.info(f"Environment: {self.config['input']['environment']}")
    self.logger.info(f"Number of columns: {len(self.config['columns'])}")

    self.logger.info(f"\nColumns:")
    for i, col in enumerate(self.config['columns'], 1):
        name = col.get('name', f'Column_{i}')
        source = col.get('source', 'default_value')
        default = col.get('default_value', '')
        if source == 'alma_set':
            self.logger.debug(f"  {i}. {name}: [MMS IDs from Alma set]")
        else:
            self.logger.debug(f"  {i}. {name}: '{default}'")

    output_settings = self.config['output_settings']
    self.logger.info(f"\nOutput Settings:")
    self.logger.debug(f"  Directory: {output_settings.get('output_directory', './output')}")
    self.logger.debug(f"  File prefix: {output_settings.get('file_prefix', 'alma_output')}")
    self.logger.debug(f"  Include headers: {output_settings.get('include_headers', False)}")

Citation metadata

Helpers used by ResourceSharing.create_lending_request_from_citation to enrich requests from PubMed (PMID) or Crossref (DOI).

enrich_citation_metadata

enrich_citation_metadata(
    pmid: Optional[str] = None,
    doi: Optional[str] = None,
    source_type: Optional[str] = None,
) -> Dict[str, Any]

Fetch citation metadata from PubMed or Crossref.

Convenience function that fetches metadata from specified source or auto-detects.

When source_type is specified, ONLY that source is used (no fallback). This is recommended when you know the identifier type upfront (e.g., from a form).

When source_type is None, uses auto-detect mode with fallback: If both PMID and DOI are provided, tries PubMed first, then Crossref as fallback.

Parameters:

Name	Type	Description	Default
pmid	Optional[str]	PubMed ID (optional)	None
doi	Optional[str]	Digital Object Identifier (optional)	None
source_type	Optional[str]	Explicit source type - 'pmid' or 'doi' (optional). If specified, only that source is tried (no fallback). If None, uses auto-detect with fallback.	None

Returns:

Type	Description
Dict[str, Any]	Metadata dictionary with 'source' field indicating which API was used

Raises:

Type	Description
ValueError	If neither PMID nor DOI provided, or invalid source_type
CitationMetadataError	If metadata fetch fails

Examples:

>>> # Explicit source - recommended for form inputs
>>> metadata = enrich_citation_metadata(
...     pmid="33219451",
...     source_type='pmid'
... )

>>> # Explicit DOI source
>>> metadata = enrich_citation_metadata(
...     doi="10.1038/s41591-020-1124-9",
...     source_type='doi'
... )

>>> # Auto-detect mode (backward compatible)
>>> metadata = enrich_citation_metadata(pmid="33219451")

>>> # Auto-detect with fallback
>>> metadata = enrich_citation_metadata(
...     pmid="33219451",
...     doi="10.1038/s41591-020-1124-9"
... )

Source code in src/almaapitk/utils/citation_metadata.py

def enrich_citation_metadata(
    pmid: Optional[str] = None,
    doi: Optional[str] = None,
    source_type: Optional[str] = None
) -> Dict[str, Any]:
    """
    Fetch citation metadata from PubMed or Crossref.

    Convenience function that fetches metadata from specified source or auto-detects.

    When source_type is specified, ONLY that source is used (no fallback).
    This is recommended when you know the identifier type upfront (e.g., from a form).

    When source_type is None, uses auto-detect mode with fallback:
    If both PMID and DOI are provided, tries PubMed first, then Crossref as fallback.

    Args:
        pmid: PubMed ID (optional)
        doi: Digital Object Identifier (optional)
        source_type: Explicit source type - 'pmid' or 'doi' (optional).
                     If specified, only that source is tried (no fallback).
                     If None, uses auto-detect with fallback.

    Returns:
        Metadata dictionary with 'source' field indicating which API was used

    Raises:
        ValueError: If neither PMID nor DOI provided, or invalid source_type
        CitationMetadataError: If metadata fetch fails

    Examples:
        >>> # Explicit source - recommended for form inputs
        >>> metadata = enrich_citation_metadata(
        ...     pmid="33219451",
        ...     source_type='pmid'
        ... )

        >>> # Explicit DOI source
        >>> metadata = enrich_citation_metadata(
        ...     doi="10.1038/s41591-020-1124-9",
        ...     source_type='doi'
        ... )

        >>> # Auto-detect mode (backward compatible)
        >>> metadata = enrich_citation_metadata(pmid="33219451")

        >>> # Auto-detect with fallback
        >>> metadata = enrich_citation_metadata(
        ...     pmid="33219451",
        ...     doi="10.1038/s41591-020-1124-9"
        ... )
    """
    # Explicit mode - use specified source only (no fallback)
    if source_type == 'pmid':
        if not pmid:
            raise ValueError("pmid parameter required when source_type='pmid'")
        logger.info(f"Fetching metadata from PubMed (explicit mode, PMID: {pmid})")
        metadata = get_pubmed_metadata(pmid)
        metadata['source'] = 'pubmed'
        logger.info("Successfully fetched metadata from PubMed")
        return metadata

    elif source_type == 'doi':
        if not doi:
            raise ValueError("doi parameter required when source_type='doi'")
        logger.info(f"Fetching metadata from Crossref (explicit mode, DOI: {doi})")
        metadata = get_crossref_metadata(doi)
        metadata['source'] = 'crossref'
        logger.info("Successfully fetched metadata from Crossref")
        return metadata

    elif source_type is not None:
        raise ValueError(
            f"Invalid source_type: '{source_type}'. "
            "Valid options: 'pmid', 'doi', or None for auto-detect"
        )

    # Auto-detect mode (backward compatible) - try with fallback
    if not pmid and not doi:
        raise ValueError("Must provide at least one of: pmid, doi")

    errors = []

    # Try PubMed first if PMID provided
    if pmid:
        try:
            logger.info(f"Attempting to fetch metadata from PubMed (PMID: {pmid})")
            metadata = get_pubmed_metadata(pmid)
            metadata['source'] = 'pubmed'
            logger.info("Successfully fetched metadata from PubMed")
            return metadata
        except Exception as e:
            error_msg = f"PubMed fetch failed: {e}"
            logger.warning(error_msg)
            errors.append(error_msg)

    # Try Crossref if DOI provided
    if doi:
        try:
            logger.info(f"Attempting to fetch metadata from Crossref (DOI: {doi})")
            metadata = get_crossref_metadata(doi)
            metadata['source'] = 'crossref'
            logger.info("Successfully fetched metadata from Crossref")
            return metadata
        except Exception as e:
            error_msg = f"Crossref fetch failed: {e}"
            logger.warning(error_msg)
            errors.append(error_msg)

    # All sources failed
    error_summary = "Failed to fetch metadata from all sources:\n" + "\n".join(errors)
    raise CitationMetadataError(error_summary)

CitationMetadataError

Bases: Exception

Base exception for citation metadata errors.