src/pybel/manager/citation_utils.py
def _get_citations_by_identifiers(
    manager: Manager,
    identifiers: Iterable[Union[str, int]],
    *,
    group_size: Optional[int] = None,
    offline: bool = False,
    prefix: Optional[str] = None,
) -> Tuple[Dict[str, Dict], Set[str]]:
    """Get citation information for the given list of PubMed identifiers using the NCBI's eUtils service.

    :type manager: pybel.Manager
    :param identifiers: an iterable of PubMed identifiers
    :param group_size: The number of PubMed identifiers to query at a time. Defaults to 200 identifiers.
    :return: A dictionary of {identifier: data dictionary} or a pair of this dictionary and a set ot erroneous
             identifiers.
    """
    if prefix is None:
        prefix = 'pubmed'

    helper = _HELPERS.get(prefix)
    if helper is None:
        raise ValueError(f'can not work on prefix: {prefix}')

    group_size = group_size if group_size is not None else 200

    identifiers = clean_pubmed_identifiers(identifiers)
    logger.info('ensuring %d %s identifiers', len(identifiers), prefix)

    enriched_models = {}
    unenriched_models = {}

    id_to_model = {
        citation_model.db_id: citation_model
        for citation_model in _get_citation_models(identifiers, prefix=prefix, manager=manager)
    }
    logger.info('%d of %d %s identifiers are already cached', len(id_to_model), len(identifiers), prefix)
    for identifier in tqdm(identifiers, desc=f'creating {prefix} models'):
        model = id_to_model.get(identifier)
        if model is None:
            model = id_to_model[identifier] = manager.get_or_create_citation(identifier=identifier, namespace=prefix)
        if model.is_enriched:
            enriched_models[identifier] = model.to_json()
        else:
            unenriched_models[identifier] = model

    logger.info('%d of %d %s are identifiers already enriched', len(enriched_models), len(identifiers), prefix)
    manager.session.commit()

    errors = set()
    if not unenriched_models or offline:
        return enriched_models, errors

    it = tqdm(unenriched_models, desc=f'getting {prefix} data in chunks of {group_size}')
    for identifier_chunk in chunked(it, n=group_size):
        helper(
            identifier_chunk,
            manager=manager,
            enriched_models=enriched_models,
            unenriched_models=unenriched_models,
            errors=errors,
        )

    return enriched_models, errors
Get citation information for the given list of PubMed identifiers using the NCBI's eUtils service.

    :type manager: pybel.Manager
    :param identifiers: an iterable of PubMed identifiers
    :param group_size: The number of PubMed identifiers to query at a time. Defaults to 200 identifiers.
    :return: A dictionary of {identifier: data dictionary} or a pair of this dictionary and a set ot erroneous
             identifiers.
    
Get citation information for the given list of PubMed identifiers using the NCBI's eUtils service.
Get citation information for the given list of PubMed identifiers using the NCBI ' s eUtils service .
def _get_citations_by_identifiers(
    manager: Manager,
    identifiers: Iterable[Union[str, int]],
    *,
    group_size: Optional[int] = None,
    offline: bool = False,
    prefix: Optional[str] = None,
) -> Tuple[Dict[str, Dict], Set[str]]:
    if prefix is None:
        prefix = 'pubmed'
    helper = _HELPERS.get(prefix)
    if helper is None:
        raise ValueError(f'can not work on prefix: {prefix}')
    group_size = group_size if group_size is not None else 200
    identifiers = clean_pubmed_identifiers(identifiers)
    logger.info('ensuring %d %s identifiers', len(identifiers), prefix)
    enriched_models = {}
    unenriched_models = {}
    id_to_model = {
        citation_model.db_id: citation_model
        for citation_model in _get_citation_models(identifiers, prefix=prefix, manager=manager)
    }
    logger.info('%d of %d %s identifiers are already cached', len(id_to_model), len(identifiers), prefix)
    for identifier in tqdm(identifiers, desc=f'creating {prefix} models'):
        model = id_to_model.get(identifier)
        if model is None:
            model = id_to_model[identifier] = manager.get_or_create_citation(identifier=identifier, namespace=prefix)
        if model.is_enriched:
            enriched_models[identifier] = model.to_json()
        else:
            unenriched_models[identifier] = model
    logger.info('%d of %d %s are identifiers already enriched', len(enriched_models), len(identifiers), prefix)
    manager.session.commit()
    errors = set()
    if not unenriched_models or offline:
        return enriched_models, errors
    it = tqdm(unenriched_models, desc=f'getting {prefix} data in chunks of {group_size}')
    for identifier_chunk in chunked(it, n=group_size):
        helper(
            identifier_chunk,
            manager=manager,
            enriched_models=enriched_models,
            unenriched_models=unenriched_models,
            errors=errors,
        )
    return enriched_models, errors
def _get_citations_by_identifiers ( manager : Manager , identifiers : Iterable [ Union [ str , int ] ] , * , group_size : Optional [ int ] = None , offline : bool = False , prefix : Optional [ str ] = None , ) -> Tuple [ Dict [ str , Dict ] , Set [ str ] ] : if prefix is None : prefix = 'pubmed' helper = _HELPERS . get ( prefix ) if helper is None : raise ValueError ( f'can not work on prefix: {prefix}' ) group_size = group_size if group_size is not None else 200 identifiers = clean_pubmed_identifiers ( identifiers ) logger . info ( 'ensuring %d %s identifiers' , len ( identifiers ) , prefix ) enriched_models = { } unenriched_models = { } id_to_model = { citation_model . db_id : citation_model for citation_model in _get_citation_models ( identifiers , prefix = prefix , manager = manager ) } logger . info ( '%d of %d %s identifiers are already cached' , len ( id_to_model ) , len ( identifiers ) , prefix ) for identifier in tqdm ( identifiers , desc = f'creating {prefix} models' ) : model = id_to_model . get ( identifier ) if model is None : model = id_to_model [ identifier ] = manager . get_or_create_citation ( identifier = identifier , namespace = prefix ) if model . is_enriched : enriched_models [ identifier ] = model . to_json ( ) else : unenriched_models [ identifier ] = model logger . info ( '%d of %d %s are identifiers already enriched' , len ( enriched_models ) , len ( identifiers ) , prefix ) manager . session . commit ( ) errors = set ( ) if not unenriched_models or offline : return enriched_models , errors it = tqdm ( unenriched_models , desc = f'getting {prefix} data in chunks of {group_size}' ) for identifier_chunk in chunked ( it , n = group_size ) : helper ( identifier_chunk , manager = manager , enriched_models = enriched_models , unenriched_models = unenriched_models , errors = errors , ) return enriched_models , errors
Get citation information for the given list of PubMed identifiers using the NCBI's eUtils service.
Get citation information for the given list of PubMed identifiers using the NCBI ' s eUtils service .
"""Get citation information for the given list of PubMed identifiers using the NCBI's eUtils service.

    :type manager: pybel.Manager
    :param identifiers: an iterable of PubMed identifiers
    :param group_size: The number of PubMed identifiers to query at a time. Defaults to 200 identifiers.
    :return: A dictionary of {identifier: data dictionary} or a pair of this dictionary and a set ot erroneous
             identifiers.
    """
A dictionary of {identifier: data dictionary} or a pair of this dictionary and a set ot erroneous
identifiers.
A dictionary of { identifier : data dictionary } or a pair of this dictionary and a set ot erroneous identifiers .
manager
Manager
identifiers
Iterable[Union[str, int]]
an iterable of PubMed identifiers
an iterable of PubMed identifiers
group_size
Optional[int]
The number of PubMed identifiers to query at a time. Defaults to 200 identifiers.
The number of PubMed identifiers to query at a time . Defaults to 200 identifiers .
offline
bool
prefix
Optional[str]