Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.managers.citations

import functools
import logging
from typing import (
    Dict,
    Optional,
    Type,
    Union,
)

from beaker.cache import CacheManager
from beaker.util import parse_cache_config_options

from galaxy.structured_app import BasicSharedApp
from galaxy.tool_util.parser.interface import Citation
from galaxy.util import (
    DEFAULT_SOCKET_TIMEOUT,
    requests,
)

log = logging.getLogger(__name__)

CitationT = Union["BibtexCitation", "DoiCitation"]
OptionalCitationT = Optional[CitationT]


[docs]class CitationsManager:
[docs] def __init__(self, app: BasicSharedApp) -> None: self.app = app self.doi_cache = DoiCache(app.config)
[docs] def citations_for_tool(self, tool): return tool.citations
[docs] def citations_for_tool_ids(self, tool_ids): citation_collection = CitationCollection() for tool_id in tool_ids: tool = self._get_tool(tool_id) for citation in self.citations_for_tool(tool): citation_collection.add(citation) return citation_collection.citations
[docs] def parse_citation(self, citation_model: Citation) -> OptionalCitationT: return parse_citation(citation_model, self)
def _get_tool(self, tool_id): tool = self.app.toolbox.get_tool(tool_id) return tool
[docs]class DoiCache:
[docs] def __init__(self, config): cache_opts = { "cache.type": config.citation_cache_type, "cache.data_dir": config.citation_cache_data_dir, "cache.lock_dir": config.citation_cache_lock_dir, "cache.url": config.citation_cache_url, "cache.table_name": config.citation_cache_table_name, "cache.schema_name": config.citation_cache_schema_name, } self._cache = CacheManager(**parse_cache_config_options(cache_opts)).get_cache("doi")
def _raw_get_bibtex(self, doi): doi_url = f"https://doi.org/{doi}" headers = {"Accept": "application/x-bibtex"} res = requests.get(doi_url, headers=headers, timeout=DEFAULT_SOCKET_TIMEOUT) # To decode the response content, res.text tries to determine the # content encoding from the Content-Type header (res.encoding), and if # that fails, falls back to guessing from the content itself (res.apparent_encoding). # The guessed encoding is sometimes wrong, better to default to utf-8. res.raise_for_status() if res.encoding is None: res.encoding = "utf-8" return res.text
[docs] def get_bibtex(self, doi): createfunc = functools.partial(self._raw_get_bibtex, doi) return self._cache.get(key=doi, createfunc=createfunc)
[docs]def parse_citation(citation_model: Citation, citation_manager) -> OptionalCitationT: """ Parse an abstract citation entry from the specified XML element. """ citation_type = citation_model.type citation_class = CITATION_CLASSES.get(citation_type, None) if not citation_class: log.warning(f"Unknown or unspecified citation type: {citation_type}") return None try: citation = citation_class(citation_model, citation_manager) except Exception as e: raise Exception(f"Invalid citation of type '{citation_type}' with content '{citation_model.content}': {e}") return citation
[docs]class CitationCollection:
[docs] def __init__(self): self.citations = []
def __iter__(self): return self.citations.__iter__() def __len__(self): return len(self.citations)
[docs] def add(self, new_citation): for citation in self.citations: if citation.equals(new_citation): # TODO: We have two equivalent citations, pick the more # informative/complete/correct. return False self.citations.append(new_citation) return True
[docs]class BaseCitation:
[docs] def to_dict(self, citation_format): if citation_format == "bibtex": return dict( format="bibtex", content=self.to_bibtex(), ) else: raise Exception(f"Unknown citation format {citation_format}")
[docs] def equals(self, other_citation): if self.has_doi() and other_citation.has_doi(): return self.doi() == other_citation.doi() else: # TODO: Do a better job figuring out if this is the same citation. return self.to_bibtex() == other_citation.to_bibtex()
[docs] def has_doi(self): return False
[docs]class BibtexCitation(BaseCitation):
[docs] def __init__(self, citation_model: Citation, citation_manager: CitationsManager): self.raw_bibtex = citation_model.content
[docs] def to_bibtex(self) -> str: return self.raw_bibtex
BIBTEX_UNSET = object()
[docs]class DoiCitation(BaseCitation):
[docs] def __init__(self, citation_model: Citation, citation_manager: CitationsManager): self.__doi = citation_model.content self.doi_cache = citation_manager.doi_cache self.raw_bibtex = BIBTEX_UNSET
[docs] def has_doi(self): return True
[docs] def doi(self): return self.__doi
[docs] def to_bibtex(self) -> str: if self.raw_bibtex is BIBTEX_UNSET: try: self.raw_bibtex = self.doi_cache.get_bibtex(self.__doi) except Exception: log.debug("Failed to fetch bibtex for DOI %s", self.__doi) if self.raw_bibtex is BIBTEX_UNSET: return f"""@MISC{{{self.__doi}, DOI = {{{self.__doi}}}, note = {{Failed to fetch BibTeX for DOI.}} }}""" else: return str(self.raw_bibtex)
CITATION_CLASSES: Dict[str, Type[CitationT]] = dict( bibtex=BibtexCitation, doi=DoiCitation, )