Add new funcs

src/pyobo/api/xrefs.py

@@ -84,7 +84,9 @@ def get_xrefs_df(
     prefix: str, *, use_tqdm: bool = False, **kwargs: Unpack[GetOntologyKwargs]
 ) -> pd.DataFrame:
     """Get all xrefs."""
-    warnings.warn("use pyobo.get_sssom_df instead of pyobo.get_xrefs_df", DeprecationWarning, stacklevel=2)
+    warnings.warn(
+        "use pyobo.get_sssom_df instead of pyobo.get_xrefs_df", DeprecationWarning, stacklevel=2
+    )
 
     version = get_version_from_kwargs(prefix, kwargs)
     path = prefix_cache_join(prefix, name="xrefs.tsv", version=version)

src/pyobo/cli/database.py

@@ -1,6 +1,7 @@
 """CLI for PyOBO Database Generation."""
 
 import logging
+import warnings
 from pathlib import Path
 
 import click
@@ -12,6 +13,7 @@
 from .database_utils import (
     _iter_alts,
     _iter_definitions,
+    _iter_mappings,
     _iter_metadata,
     _iter_names,
     _iter_properties,
@@ -108,8 +110,8 @@ def build(ctx: click.Context, **kwargs: Unpack[DatabaseKwargs]) -> None:
         ctx.invoke(alts, **updated_kwargs)
         click.secho("Synonyms", fg="cyan", bold=True)
         ctx.invoke(synonyms, **updated_kwargs)
-        click.secho("Xrefs", fg="cyan", bold=True)
-        ctx.invoke(xrefs, **updated_kwargs)
+        click.secho("Mappings", fg="cyan", bold=True)
+        ctx.invoke(mappings, **updated_kwargs)
         click.secho("Names", fg="cyan", bold=True)
         ctx.invoke(names, **updated_kwargs)
         click.secho("Definitions", fg="cyan", bold=True)
@@ -292,6 +294,7 @@ def properties(zenodo: bool, directory: Path, **kwargs: Unpack[DatabaseKwargs])
 @database_annotate
 def xrefs(zenodo: bool, directory: Path, **kwargs: Unpack[DatabaseKwargs]) -> None:
     """Make the prefix-identifier-xref dump."""
+    warnings.warn("Use pyobo.database.mappings instead", DeprecationWarning, stacklevel=2)
     with logging_redirect_tqdm():
         it = _iter_xrefs(**kwargs)
         paths = db_output_helper(
@@ -306,6 +309,28 @@ def xrefs(zenodo: bool, directory: Path, **kwargs: Unpack[DatabaseKwargs]) -> No
         update_zenodo(JAVERT_RECORD, paths)
 
 
+@database_annotate
+def mappings(zenodo: bool, directory: Path, **kwargs: Unpack[DatabaseKwargs]) -> None:
+    """Make the SSSOM dump."""
+    columns = [
+        "subject_id",
+        "object_id",
+        "predicate_id",
+        "mapping_justification",
+        "mapping_source",
+    ]
+    with logging_redirect_tqdm():
+        it = _iter_mappings(**kwargs)
+        db_output_helper(
+            it,
+            "mappings",
+            columns,
+            directory=directory,
+        )
+    if zenodo:
+        raise NotImplementedError("need to do initial manual upload of SSSOM build")
+
+
 if __name__ == "__main__":
     logging.captureWarnings(True)
     with logging_redirect_tqdm():

src/pyobo/cli/database_utils.py

@@ -5,6 +5,7 @@
 import gzip
 import logging
 from collections.abc import Iterable
+from functools import partial
 from typing import cast
 
 import bioregistry
@@ -20,6 +21,7 @@
     get_metadata,
     get_properties_df,
     get_relations_df,
+    get_sssom_df,
     get_typedef_df,
     get_xrefs_df,
 )
@@ -139,3 +141,18 @@ def _iter_xrefs(
     for df in iter_xref_plugins(skip_below=kwargs.get("skip_below")):
         df.dropna(inplace=True)
         yield from tqdm(df.values, leave=False, total=len(df.index), unit_scale=True)
+
+
+def _iter_mappings(
+    **kwargs: Unpack[IterHelperHelperDict],
+) -> Iterable[tuple[str, str, str, str, str]]:
+    f = partial(get_sssom_df, names=False)
+    # hack in a name to the partial function object since
+    # it's used for the tqdm description in iter_helper_helper
+    f.__name__ = "get_mappings_df" #type:ignore
+    it = iter_helper_helper(f, **kwargs)
+    for prefix, df in it:
+        for row in df.values:
+            # append on the mapping_source
+            # (https://mapping-commons.github.io/sssom/mapping_source/)
+            yield *row, prefix