Replace Metapub with Eutils

Metapub is not as well supported as it was in the past and uses a better maintained package,
eutils, under the hood. This change replaces previous functionality that used Metapub to
search for PubMed articles with eutils.

Author: bencap

alembic/versions/22e2d92d602e_add_publication_identifier_metadata_.py

@@ -6,10 +6,13 @@
 
 """
 from typing import Optional
+import os
 
-import metapub
+import eutils
+from eutils._internal.xmlfacades.pubmedarticleset import PubmedArticleSet
 import sqlalchemy as sa
 from eutils import EutilsNCBIError
+from mavedb.lib.exceptions import AmbiguousIdentifierError
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Session
 
@@ -82,11 +85,11 @@ def upgrade():
         if pub_article:
             item.title = pub_article.title
             item.abstract = pub_article.abstract
-            item.publication_doi = pub_article.published_doi
+            item.publication_doi = pub_article.publication_doi
             item.publication_year = pub_article.publication_year
             item.publication_journal = pub_article.publication_journal
 
-            authors = [author["name"].replace("'", "''") for author in pub_article.authors]
+            authors = [str(author["name"]).replace("'", "''") for author in pub_article.authors]
             authors = [{"name": author, "primary": idx == 0} for idx, author in enumerate(authors)]
             item.authors = authors
 
@@ -97,7 +100,7 @@ def upgrade():
             item.preprint_date = bio_article.preprint_date
             item.reference_html = bio_article.reference_html
 
-            authors = [author["name"].replace("'", "''") for author in bio_article.authors]
+            authors = [str(author["name"]).replace("'", "''") for author in bio_article.authors]
             authors = [{"name": author, "primary": idx == 0} for idx, author in enumerate(authors)]
             item.authors = authors
 
@@ -108,7 +111,7 @@ def upgrade():
             item.preprint_date = med_article.preprint_date
             item.reference_html = med_article.reference_html
 
-            authors = [author["name"].replace("'", "''") for author in med_article.authors]
+            authors = [str(author["name"]).replace("'", "''") for author in med_article.authors]
             authors = [{"name": author, "primary": idx == 0} for idx, author in enumerate(authors)]
             item.authors = authors
 
@@ -145,13 +148,18 @@ def fetch_pubmed_article(identifier: str) -> Optional[ExternalPublication]:
     """
     Fetch an existing PubMed article from NCBI
     """
-    fetch = metapub.PubMedFetcher()
+    fetch = eutils.QueryService(api_key=os.getenv("NCBI_API_KEY"))
     try:
-        article = fetch.article_by_pmid(pmid=identifier)
-        if article:
-            article = ExternalPublication(identifier=identifier, db_name="PubMed", external_publication=article)
+        fetched_articles = list(PubmedArticleSet(fetch.efetch({"db": "pubmed", "id": identifier})))
+        assert len(fetched_articles) < 2
+        article = ExternalPublication(identifier=identifier, db_name="PubMed", external_publication=fetched_articles[0])
+
+    except AssertionError as exc:
+        raise AmbiguousIdentifierError(f"Fetched more than 1 PubMed article associated with PMID {identifier}") from exc
     except EutilsNCBIError:
         return None
+    except IndexError:
+        return None
     else:
         return article
 

src/mavedb/lib/identifiers.py

@@ -1,8 +1,11 @@
+import os
 from datetime import date
 from typing import Optional, Union
 
-import metapub
+import eutils
 from eutils import EutilsNCBIError
+from eutils._internal.xmlfacades.pubmedarticle import PubmedArticle
+from eutils._internal.xmlfacades.pubmedarticleset import PubmedArticleSet
 from sqlalchemy.orm import Session
 
 from mavedb.lib.exceptions import AmbiguousIdentifierError, NonexistentIdentifierError
@@ -43,20 +46,23 @@ class ExternalPublication:
     identifier: str
     title: str
     abstract: str
-    authors: list[dict[str, str]]
+    authors: list[dict[str, Union[str, bool]]]
     publication_year: int
-    published_doi: Optional[str]
-    preprint_doi: Optional[str]
+    publication_volume: Optional[str]
+    publication_pages: Optional[str]
+    publication_doi: Optional[str]
     publication_journal: Optional[str]
+    preprint_doi: Optional[str]
     preprint_date: Optional[date]
     db_name: str
-    reference_html: str
+
+    _article_cit_fmt = "{author}. {title}. {journal}. {year}; {volume}:{pages}. {doi}"
 
     def __init__(
         self,
         identifier: str,
         db_name: str,
-        external_publication: Union[RxivContentDetail, metapub.PubMedArticle],
+        external_publication: Union[RxivContentDetail, PubmedArticle],
     ) -> None:
         """
         NOTE: We assume here that the first author in each of these author lists is the primary author
@@ -71,54 +77,46 @@ def __init__(
         self.db_name = db_name
         self.title = str(external_publication.title)
         self.abstract = str(external_publication.abstract)
-        self.authors = self._generate_author_list(external_publication.author_list)
+        self.authors = self._generate_author_list(external_publication.authors)
 
         # Non-shared fields
-        if isinstance(external_publication, metapub.PubMedArticle):
+        if isinstance(external_publication, PubmedArticle):
             self.publication_year = int(external_publication.year)
-            self.publication_journal = str(external_publication.journal)
-            self.published_doi = str(external_publication.doi)
-            self.preprint_doi = None
-            self.preprint_date = None
+            self.publication_journal = external_publication.jrnl
+            self.publication_doi = external_publication.doi
+            self.publication_volume = external_publication.volume
+            self.publication_pages = external_publication.pages
         elif isinstance(external_publication, RxivContentDetail):
             self.preprint_doi = external_publication.doi
             self.preprint_date = external_publication.date
-            self.publication_journal = None
 
-        self.reference_html = str(external_publication.citation_html)
-
-    def _generate_author_list(self, authors: Union[list[str], list[metapub.PubMedAuthor]]) -> list[dict[str, str]]:
+    def _generate_author_list(self, authors: list[str]) -> list[dict[str, Union[str, bool]]]:
         """
-        Generates a tuple of author names associated with this publication.
+        Generates a list of author names and thier authorship level associated with this publication.
         """
-        if not authors:
-            return []
-
-        if isinstance(authors[0], metapub.PubMedAuthor):
-            created_authors = [
-                {"name": ", ".join([str(authors[0].last_name), str(authors[0].fore_name)]), "primary": True}
-            ]
+        return [{"name": author, "primary": idx == 0} for idx, author in enumerate(authors)]
+
+    def _format_authors(self) -> str:
+        """Helper function for returning a well formatted HTML author list"""
+        if self.authors and len(self.authors) > 2:
+            author = str(self.authors[0]["name"]) + ", <i>et al</i>"
+        elif self.authors and len(self.authors) == 2:
+            author = " and ".join([str(author["name"]) for author in self.authors])
+        elif self.authors and len(self.authors) < 2:
+            author = str(self.authors[0]["name"])
         else:
-            created_authors = [{"name": authors[0], "primary": True}]
-
-        for author in authors[1:]:
-            if isinstance(author, metapub.PubMedAuthor):
-                created_authors.append(
-                    {"name": ", ".join([str(author.last_name), str(author.fore_name)]), "primary": False}
-                )
-            else:
-                created_authors.append({"name": author, "primary": False})
+            author = ""
 
-        return created_authors
+        return author
 
     @property
     def first_author(self) -> str:
-        return self.authors[0]["name"]
+        return str(self.authors[0]["name"])
 
     @property
     def secondary_authors(self) -> list[str]:
         if len(self.authors) > 1:
-            return [author["name"] for author in self.authors[1:]]
+            return [str(author["name"]) for author in self.authors[1:]]
         else:
             return []
 
@@ -133,6 +131,35 @@ def url(self) -> str:
         else:
             return ""
 
+    @property
+    def reference_html(self) -> str:
+        """
+        Return a well formatted citation HTML string based on article data.
+        Intends to return an identical citation html string to metapub.PubMedArticle.
+        """
+        author = self._format_authors()
+
+        if self.db_name in ["PubMed"]:
+            doi_str = "" if not self.publication_doi else self.publication_doi
+            title = "(None)" if not self.title else self.title.strip(".")
+            journal = "(None)" if not self.publication_journal else self.publication_journal.strip(".")
+            year = "(Unknown year)" if not self.publication_year else self.publication_year
+            volume = "(Unknown volume)" if not self.publication_volume else self.publication_volume
+            pages = "(Unknown pages)" if not self.publication_pages else self.publication_pages
+        else:
+            doi_str = "" if not self.preprint_doi else self.preprint_doi
+            title = "(None)" if not self.title else self.title.strip(".")
+            journal = "(None)" if not self.publication_journal else self.publication_journal.strip(".")
+            year = "(Unknown year)" if not self.preprint_date else self.preprint_date.year
+
+            # We don't receive these fields from rxiv platforms
+            volume = "(Unknown volume)"
+            pages = "(Unknown pages)"
+
+        return self._article_cit_fmt.format(
+            author=author, volume=volume, pages=pages, year=year, title=title, journal=journal, doi=doi_str
+        )
+
 
 async def find_or_create_doi_identifier(db: Session, identifier: str):
     """
@@ -152,13 +179,18 @@ async def fetch_pubmed_article(identifier: str) -> Optional[ExternalPublication]
     """
     Fetch an existing PubMed article from NCBI
     """
-    fetch = metapub.PubMedFetcher()
+    fetch = eutils.QueryService(api_key=os.getenv("NCBI_API_KEY"))
     try:
-        article = fetch.article_by_pmid(pmid=identifier)
-        if article:
-            article = ExternalPublication(identifier=identifier, db_name="PubMed", external_publication=article)
+        fetched_articles = list(PubmedArticleSet(fetch.efetch({"db": "pubmed", "id": identifier})))
+        assert len(fetched_articles) < 2
+        article = ExternalPublication(identifier=identifier, db_name="PubMed", external_publication=fetched_articles[0])
+
+    except AssertionError as exc:
+        raise AmbiguousIdentifierError(f"Fetched more than 1 PubMed article associated with PMID {identifier}") from exc
     except EutilsNCBIError:
         return None
+    except IndexError:
+        return None
     else:
         return article
 
@@ -280,7 +312,7 @@ def create_generic_article(article: ExternalPublication) -> PublicationIdentifie
             title=article.title,
             abstract=article.abstract,
             authors=article.authors,
-            publication_doi=article.published_doi,
+            publication_doi=article.publication_doi,
             publication_year=article.publication_year,
             publication_journal=article.publication_journal,
             reference_html=article.reference_html,

src/mavedb/lib/rxiv.py

@@ -20,7 +20,7 @@ class RxivContentDetail:
     title: str
     doi: str
     category: str
-    author_list: list[str]
+    authors: list[str]
     author_corresponding: str
     author_corresponding_institution: str
     date: datetime.date
@@ -32,13 +32,11 @@ class RxivContentDetail:
     published: str
     server: Optional[str]  # not guaranteed
 
-    _article_cit_fmt = "{author}. {title}. {journal}. {year}; {volume}:{pages}.{doi}"
-
     def __init__(self, metadata: dict[str, str]) -> None:
         self.title = metadata["title"]
         self.doi = metadata["doi"]
         self.category = metadata["category"]
-        self.author_list = [s.strip() for s in metadata.get("authors", "").split(";")]
+        self.authors = [s.strip() for s in metadata.get("authors", "").split(";")]
         self.author_corresponding = metadata["author_corresponding"]
         self.author_corresponding_institution = metadata["author_corresponding_institution"]
         self.date = datetime.datetime.strptime(metadata["date"], "%Y-%m-%d")
@@ -50,46 +48,13 @@ def __init__(self, metadata: dict[str, str]) -> None:
         self.published = metadata["published"]
         self.server = metadata.get("server")
 
-    def _format_authors(self) -> str:
-        """Helper function for returning a well formatted HTML author list"""
-        if self.author_list and len(self.author_list) > 2:
-            author = self.author_list[0] + ", <i>et al</i>"
-        elif self.author_list and len(self.author_list) == 2:
-            author = " and ".join([author for author in self.author_list])
-        elif self.author_list and len(self.author_list) < 2:
-            author = self.author_list[0]
-        else:
-            author = ""
-
-        return author
-
     @property
     def first_author(self) -> Optional[str]:
-        if len(self.author_list) > 0:
-            return self.author_list[0]
+        if len(self.authors) > 0:
+            return self.authors[0]
         else:
             return None
 
-    @property
-    def citation_html(self):
-        """
-        Return a well formatted citation HTML string based on pre-print article data.
-        Intends to return an identical citation html string to metapub.PubMedArticle.
-        """
-        author = self._format_authors()
-        doi_str = "" if not self.doi else self.doi
-        title = "(None)" if not self.title else self.title.strip(".")
-        journal = "(None)" if not self.server else self.server.strip(".")
-        year = "(Unknown yeaer)" if not self.date.year else self.date.year
-
-        # We don't receive these fields from rxiv platforms
-        volume = "(Unknown volume)"
-        pages = "(Unknown pages)"
-
-        return self._article_cit_fmt.format(
-            author=author, volume=volume, pages=pages, year=year, title=title, journal=journal, doi=doi_str
-        )
-
 
 class RxivPublication:
     """
@@ -167,46 +132,6 @@ def first_author(self) -> Optional[str]:
         else:
             return None
 
-    @property
-    def citation_html(self):
-        """
-        Return a well formatted citation HTML string based on pre-print article data.
-        Intends to return an identical citation html string to metapub.PubMedArticle.
-        """
-        author = self._format_authors()
-        doi_str = "" if not self.preprint_doi else self.preprint_doi
-        title = "(None)" if not self.preprint_title else self.preprint_title.strip(".")
-        journal = "(None)" if not self.preprint_platform else self.preprint_platform.strip(".")
-        year = "(Unknown yeaer)" if not self.preprint_date.year else self.preprint_date.year
-
-        # We don't receive these fields from rxiv platforms
-        volume = "(Unknown volume)"
-        pages = "(Unknown pages)"
-
-        return self._article_cit_fmt.format(
-            author=author, volume=volume, pages=pages, year=year, title=title, journal=journal, doi=doi_str
-        )
-
-    @property
-    def publication_citation_html(self):
-        """
-        Return a well formatted citation HTML string based on publication data.
-        Intends to return an identical citation html string to metapub.PubMedArticle.
-        """
-        author = self._format_authors()
-        doi_str = "" if not self.published_doi else self.published_doi
-        title = "(None)" if not self.preprint_title else self.preprint_title.strip(".")
-        journal = "(None)" if not self.published_journal else self.published_journal.strip(".")
-        year = "(Unknown yeaer)" if not self.published_date.year else self.published_date.year
-
-        # We don't receive these fields from rxiv platforms
-        volume = "(Unknown volume)"
-        pages = "(Unknown pages)"
-
-        return self._article_cit_fmt.format(
-            author=author, volume=volume, pages=pages, year=year, title=title, journal=journal, doi=doi_str
-        )
-
 
 class RxivStatistics:
     interval: str

src/mavedb/server_main.py

@@ -12,8 +12,8 @@
 from slack_sdk.webhook import WebhookClient
 from sqlalchemy.orm import configure_mappers
 from starlette import status
-from starlette.responses import JSONResponse, Response
-from metapub.exceptions import InvalidPMID
+from starlette.responses import JSONResponse
+from eutils._internal.exceptions import EutilsRequestError
 
 from mavedb.models import *
 
@@ -94,8 +94,8 @@ async def nonexistent_identifier_error_exception_handler(request: Request, exc:
     )
 
 
-@app.exception_handler(InvalidPMID)
-async def nonexistent_pmid_error_exception_handler(request: Request, exc: InvalidPMID):
+@app.exception_handler(EutilsRequestError)
+async def nonexistent_pmid_error_exception_handler(request: Request, exc: EutilsRequestError):
     return JSONResponse(
         status_code=404,
         content={"message": str(exc)},