Adding a source
To add an Example source you need a config YAML and a parser module.
pysec2pri.api and pysec2pri.cli will pick up the new source
automatically once those are in place.
The goal of the parser is to extract all primary IDs, all primary labels, and
all available secondary mappings for them and generate an IdMappingSets
and/or a LabelMappingSets:
docs/source/example/ contains a working example to copy from.
type |
maps |
example |
|---|---|---|
|
old ID → current ID |
|
|
old symbol → current symbol |
|
1. config/example.yaml
Copy docs/source/example/example.yaml to
src/pysec2pri/config/<source>.yaml and edit it.
# An example datasource config. Copy this file to
# src/pysec2pri/config/<source>.yaml and edit it.
#
# Example publishes two files per release:
# withdrawn.tsv example_id, replacement_id, status
# complete.tsv example_id, symbol, prev_symbols, alias_symbols
config_id: "example"
datasource_id: "example"
name: "Example"
parser_class: "ExampleParser" # the class in parsers/example.py
# The arguments that split one release into separate datasets. Each one adds a
# slug: a segment of the IRIs that says which dataset a mapping set is, where
# the version alone is not enough. Example publishes one dataset per release,
# so it declares none, and its IRIs stop at the version. If it also published a
# "3star" and a "complete" subset, `products: ["subset"]` would give:
# mapping_set_id https://sec2pri.github.io/example/2024-01-01/3star
# record_id sec2pri:example/2024-01-01/3star/<hash>
# products: ["subset"]
prefix: "EX"
curie_base_url: "https://example.org/entry/"
entity_types: ["genes"]
download_urls:
withdrawn: "https://example.org/current/withdrawn.tsv"
complete: "https://example.org/current/complete.tsv"
# Which mapping sets can we get from this source.
mapping_sets:
ids:
method: parse
inputs: { withdrawn: input_path, complete: complete_set_path }
formats: [sssom, sec2pri, pri_ids, secondary, rdf, json, owl, all]
labels:
method: parse_labels
inputs: { complete: complete_set_path }
formats: [sssom, label_sec2pri, name2synonym, pri_labels, rdf, json, all]
primary_ids:
method: parse_primary_ids
inputs: { complete: complete_set_path }
formats: [pri_ids]
# SSSOM metadata for the mapping set as a whole.
mappingset:
curie_map: # Add needed
EX: "https://example.org/entry/"
IAO: "http://purl.obolibrary.org/obo/IAO_"
example: "https://example.org/"
oboInOwl: "http://www.geneontology.org/formats/oboInOwl#"
semapv: "https://w3id.org/semapv/vocab/"
sssom: "https://w3id.org/sssom/"
orcid: "https://orcid.org/"
mappings: []
mapping_set_id: "https://sec2pri.github.io/example"
mapping_set_title: "Example Secondary to Primary Mapping"
mapping_set_description: "Mappings from retired to current Example IDs."
creator_id: ["orcid:0000-0000-0000-0000"]
creator_label: ["Your Name"]
license: "https://creativecommons.org/publicdomain/zero/1.0/"
subject_source: "example:"
object_source: "example:"
mapping_provider: "https://github.com/sec2pri/pysec2pri"
mapping_tool: "https://pypi.org/project/pysec2pri"
comment: "Generated by pysec2pri"
see_also: "https://github.com/sec2pri/pysec2pri"
issue_tracker: "https://github.com/sec2pri/pysec2pri/issues"
subject_preprocessing: "https://w3id.org/semapv/vocab/BackgroundKnowledgeBasedMatching"
object_preprocessing: "https://w3id.org/semapv/vocab/BackgroundKnowledgeBasedMatching"
# SSSOM defaults for every row.
mapping:
record_id: "sec2pri:example/"
predicate_id: "IAO:0100001"
predicate_label: "term replaced by"
mapping_justification: "semapv:BackgroundKnowledgeBasedMatching"
license: "https://creativecommons.org/publicdomain/zero/1.0/"
subject_source: "example:"
object_source: "example:"
mapping_source: "example:"
mapping_tool: "pysec2pri"
confidence: 1.0
id_pattern: "^EX:\\d+$"
default_output_filename: "example_sec2pri.sssom.tsv"
source: "Example database"
homepage: "https://example.org/"
data_license: "https://creativecommons.org/publicdomain/zero/1.0/"
inputs (under mapping_sets) specifies which files to download and
which argument each one becomes in generate_ids() and
generate_labels().
Datasets within a release
Most sources publish one dataset per release. HGNC is one, so the source and the version name it completely:
https://sec2pri.github.io/hgnc/2024-01-01
Some publish several. ChEBI release 245 is both a 3star subset and a
complete one. Ensembl release 115 is a different set of mappings for every
species. Those are separate mapping sets with separate IRIs.
products names the parser arguments that also become slugs in the generated
mapping document IRI and the set of record_ids it contains.
products: ["subset"]
Means that each subset of release 245 gets its own IRI pattern:
|
|
|
|---|---|---|
|
|
|
|
|
|
Each name in products also becomes a CLI option (e.g. --subset).
2. parsers/example.py
Each source publishes its deprecation and synonym release files differently. In this simplified example there are two files:
the withdrawn file: the secondary mappings.
the complete set: every current entry ID and primary label.
But in your case the whole information could be extracted from one or several files.
Copy docs/source/example/example.py to
src/pysec2pri/parsers/<source>.py. Name the class to match the
parser_class value in the config YAML.
parse() must return the
IdMappingSet and
parse_labels() the
LabelMappingSet.
"""A simple copy-paste parser example. Copy to ``src/pysec2pri/parsers/<source>.py``.
This example uses two files per release:
- ``withdrawn.tsv``: ``example_id, replacement_id, status``.
- ``complete.tsv``: ``example_id, symbol, prev_symbols, alias_symbols``.
The class name must match ``parser_class`` in the config, and every method the
config's ``mapping_sets`` names must exist here. The specific published files
for a datasource will differ, adapt as needed.
"""
from __future__ import annotations
from pathlib import Path
from pysec2pri.parsers.base import (
WITHDRAWN_ENTRY,
WITHDRAWN_ENTRY_LABEL,
BaseMappingSet,
BaseParser,
)
# Column names in Example's files. _find_column matches case-insensitively.
EXAMPLE_ID = "example_id"
REPLACEMENT_ID = "replacement_id"
STATUS = "status"
SYMBOL = "symbol"
PREV_SYMBOLS = "prev_symbols"
ALIAS_SYMBOLS = "alias_symbols"
class ExampleParser(BaseParser):
"""Parser for Example's withdrawn and complete files."""
datasource_name = "example"
def parse(
self,
input_path: Path | str | None,
complete_set_path: Path | str | None = None,
) -> BaseMappingSet:
"""Parse the withdrawn file into retired -> current ID mappings.
Args:
input_path: Path to ``withdrawn.tsv``.
complete_set_path: Path to ``complete.tsv``. Optional: without it
the mappings are the same, but the full current-ID list and
ambiguity checks are incomplete.
Returns:
An ``IdMappingSet``.
Raises:
ValueError: If *input_path* is missing, or lacks its ID column.
"""
if input_path is None:
raise ValueError("input_path must not be None")
input_path = Path(input_path)
self._resolve_version(input_path)
df = self._read_tsv(input_path)
id_col = self._find_column(df.columns, EXAMPLE_ID)
if id_col is None:
raise ValueError(f"No {EXAMPLE_ID} column in {input_path}")
new_col = self._find_column(df.columns, REPLACEMENT_ID)
status_col = self._find_column(df.columns, STATUS)
# Fields shared by every row (license, sources, justification ...).
fixed = self._fixed_mapping_fields()
namespace = self._record_namespace()
m_meta = self.get_mapping_metadata()
rows: list[dict[str, str | None]] = []
for row in df.iter_rows(named=True):
retired = row.get(id_col)
if not retired:
continue
replacement = row.get(new_col) if new_col else None
status = str(row.get(status_col) or "") if status_col else ""
if replacement:
rows.append(
{
"subject_id": retired,
"object_id": replacement,
"predicate_id": m_meta["predicate_id"],
"predicate_label": m_meta.get("predicate_label"),
"record_id": self._record_id(namespace, replacement, retired),
}
)
elif "withdrawn" in status.lower():
# Retired with nothing to point at.
rows.append(
{
"subject_id": retired,
"object_id": WITHDRAWN_ENTRY,
"object_label": WITHDRAWN_ENTRY_LABEL,
"predicate_id": "oboInOwl:consider",
"comment": "Withdrawn entry with no replacement.",
"record_id": self._record_id(namespace, WITHDRAWN_ENTRY, retired),
}
)
mappings = self._build_mappings(rows, fixed, desc="Processing withdrawn")
primary_ids = (
self._extract_primary_ids(Path(complete_set_path)) if complete_set_path else None
)
return self.create_mapping_set(mappings, mapping_type="id", primary_ids=primary_ids)
def parse_labels(self, complete_set_path: Path | str | None) -> BaseMappingSet:
"""Parse the complete file into old -> current symbol mappings.
Each current entry lists the symbols it used to have (``prev_symbols``)
and the ones it also goes by (``alias_symbols``). ``_label_type`` picks
the predicate: ``previous`` renames, ``alias`` synonyms.
Args:
complete_set_path: Path to ``complete.tsv``.
Returns:
A ``LabelMappingSet``.
Raises:
ValueError: If *complete_set_path* is missing, or lacks its columns.
"""
if complete_set_path is None:
raise ValueError("complete_set_path must not be None")
complete_set_path = Path(complete_set_path)
self._resolve_version(complete_set_path)
df = self._read_tsv(complete_set_path)
id_col = self._find_column(df.columns, EXAMPLE_ID)
symbol_col = self._find_column(df.columns, SYMBOL)
if id_col is None or symbol_col is None:
raise ValueError(f"No {EXAMPLE_ID}/{SYMBOL} column in {complete_set_path}")
prev_col = self._find_column(df.columns, PREV_SYMBOLS)
alias_col = self._find_column(df.columns, ALIAS_SYMBOLS)
fixed = self._fixed_mapping_fields()
namespace = self._record_namespace()
rows: list[dict[str, str | None]] = []
for row in df.iter_rows(named=True):
current_id, current = row.get(id_col), row.get(symbol_col)
if not current_id or not current:
continue
for col, label_type in ((prev_col, "previous"), (alias_col, "alias")):
raw = row.get(col) if col else None
for old in self._split_labels(labels_str=str(raw)) if raw else []:
rows.append(
{
"subject_label": old,
"object_id": current_id,
"object_label": current,
"_label_type": label_type,
"record_id": self._record_id(namespace, current_id, old),
}
)
mappings = self._build_mappings(rows, fixed, desc="Processing symbols")
return self.create_mapping_set(
mappings,
mapping_type="label",
primary_labels=self._extract_primary_labels(complete_set_path),
)
def parse_primary_ids(self, complete_set_path: Path | str | None) -> BaseMappingSet:
"""Return a mapping set holding only the full current-ID list.
Args:
complete_set_path: Path to ``complete.tsv``.
Returns:
A mapping set with no mappings and ``_primary_ids`` populated.
Raises:
ValueError: If *complete_set_path* is missing.
"""
if complete_set_path is None:
raise ValueError("complete_set_path must not be None")
complete_set_path = Path(complete_set_path)
self._resolve_version(complete_set_path)
return self.create_mapping_set(
[], mapping_type="id", primary_ids=self._extract_primary_ids(complete_set_path)
)
def _extract_primary_ids(self, file_path: Path) -> set[str]:
"""Return every current ID in the complete file.
Raises:
ValueError: If the file lacks its ID column.
"""
df = self._read_tsv(file_path)
id_col = self._find_column(df.columns, EXAMPLE_ID)
if id_col is None:
raise ValueError(f"No {EXAMPLE_ID} column in {file_path}")
return {str(v) for v in df[id_col].drop_nulls().to_list()}
def _extract_primary_labels(self, file_path: Path) -> dict[str, set[str]]:
"""Return ``{current symbol: {ID, ...}}`` for the complete file.
Raises:
ValueError: If the file lacks its ID or symbol column.
"""
df = self._read_tsv(file_path)
id_col = self._find_column(df.columns, EXAMPLE_ID)
symbol_col = self._find_column(df.columns, SYMBOL)
if id_col is None or symbol_col is None:
raise ValueError(f"No {EXAMPLE_ID}/{SYMBOL} column in {file_path}")
result: dict[str, set[str]] = {}
for id_, label in df.select([id_col, symbol_col]).drop_nulls().rows():
result.setdefault(str(label), set()).add(str(id_))
return result
__all__ = ["ExampleParser"]
1. parsers/__init__.py
The parser package uses lazy imports (PEP 562): the heavy dependencies
(polars, sssom, rdflib) are only loaded the first time a parser
class is actually used. For this to work, every parser class must be registered
in three places in src/pysec2pri/parsers/__init__.py.
_LAZY_EXPORTS
This dict maps each public name to the submodule that defines it. The
module-level __getattr__ hook reads it to import and cache the class on
first access, so pysec2pri.parsers.ExampleParser resolves correctly without
eagerly importing the module at startup.
_LAZY_EXPORTS: dict[str, str] = {
...
"ExampleParser": "pysec2pri.parsers.example", # ← add this
}
TYPE_CHECKING block
Static analysis tools (mypy, Pyright) and IDEs never execute __getattr__,
so they cannot resolve lazily-exported names. The if TYPE_CHECKING: block
provides the explicit import they need. It is never executed at runtime, so it
costs nothing.
if TYPE_CHECKING:
...
from pysec2pri.parsers.example import ExampleParser # ← add this
__all__
__all__ declares the public API of the package. It is used by
from pysec2pri.parsers import *, dir(pysec2pri.parsers), and
documentation generators. A name absent from __all__ is considered
private even if it can be imported.
__all__ = [
...
"ExampleParser", # ← add this
]
4. downloads/example.py: only if URLs are not fixed
If Example publishes the same static URLs every release, skip this step.
The download_urls in the config are sufficient: each URL is fetched,
decompressed if needed, and handed to the parser.
Write a downloader only when:
the URL contains a version string or species slug that must be substituted,
you need to find or list available releases programmatically, or
you need the real upstream release date for the SSSOM
mapping_datefield.
"""Copy to ``src/pysec2pri/downloads/<source>.py``.
Only needed when the URL needs a version to have to find the (latest) release.
Otherwise the config's ``download_urls`` provide the whole download logic and
this file is not needed.
Example keeps every release at ``https://example.org/{version}/``, and lists
them at ``https://example.org/archive/``.
"""
from __future__ import annotations
import re
from datetime import datetime
from typing import TYPE_CHECKING, Any
import httpx
from mapkgsutils.download import ReleaseInfo
from pysec2pri.logging import logger
from pysec2pri.parsers.base import BaseDownloader
if TYPE_CHECKING:
from pathlib import Path
from pysec2pri.parsers.base import DatasourceConfig
__all__ = ["ExampleDownloader", "check_example_release", "urls_and_date"]
_ARCHIVE_URL = "https://example.org/archive/"
def _urls_for_version(version: str) -> dict[str, str]:
"""Return each file's download_urls key -> URL, for one release."""
return {
"withdrawn": f"https://example.org/{version}/withdrawn.tsv",
"complete": f"https://example.org/{version}/complete.tsv",
}
class ExampleDownloader(BaseDownloader):
"""Downloader for Example's per-release files."""
datasource_name = "example"
def get_download_urls(self, version: str | None = None, **kwargs: Any) -> dict[str, str]:
"""Return each file's download_urls key -> URL.
Args:
version: Release to build URLs for. Latest when ``None``.
**kwargs: Unused; accepted for interface uniformity.
"""
return _urls_for_version(version or check_example_release().version or "")
def download(
self,
output_dir: Path,
version: str | None = None,
decompress: bool = True,
**kwargs: Any,
) -> dict[str, Path]:
"""Download the files into *output_dir*, returning key -> local path."""
return self._download_urls(self.get_download_urls(version), output_dir, decompress)
def list_versions(self) -> list[str]:
"""Return every release Example keeps, oldest first.
Only needed for a source with an archive: this is what
``pysec2pri list-versions`` and ``--consolidate`` walk.
"""
with httpx.Client(follow_redirects=True, timeout=30.0) as client:
response = client.get(_ARCHIVE_URL)
response.raise_for_status()
return sorted(set(re.findall(r'href="(\d{4}-\d{2}-\d{2})/"', response.text)))
def check_example_release() -> ReleaseInfo:
"""Return the newest release: its version, its date, and its files."""
with httpx.Client(follow_redirects=True, timeout=30.0) as client:
response = client.get(_ARCHIVE_URL)
response.raise_for_status()
versions = re.findall(r'href="(\d{4}-\d{2}-\d{2})/"', response.text)
if not versions:
raise ValueError(f"No Example releases found at {_ARCHIVE_URL}")
latest = max(versions)
logger.info("Latest Example release: %s", latest)
return ReleaseInfo(
datasource="example",
version=latest,
release_date=datetime.strptime(latest, "%Y-%m-%d"),
is_new=True,
files=_urls_for_version(latest),
)
def urls_and_date(
version: str | None, config: DatasourceConfig, **kwargs: Any
) -> tuple[dict[str, str], datetime | None]:
"""Return the files to download and the date that release came out.
The date becomes each mapping's ``mapping_date``.
Args:
version: Release to download, or ``None`` for the latest.
config: Example's config. Unused here: the URLs come from the version.
**kwargs: Unused in example, use if needed.
"""
if version is None:
release = check_example_release()
return release.files, release.release_date
return _urls_for_version(version), datetime.strptime(version, "%Y-%m-%d")
downloads/__init__.py registries
After writing the downloader, register it in the three dicts in
src/pysec2pri/downloads/__init__.py.
``CHECK_RELEASE`` needed by
mapkgsutils.download.get_latest_release_info(). Maps the datasource name
to the function that queries the upstream source and returns the latest
available release identifier.
CHECK_RELEASE = {
...
"example": example.check_example_release, # ← add this
}
``DOWNLOADERS`` needed by mapkgsutils.download.list_versions().
Maps the datasource name to the BaseDownloader subclass.
DOWNLOADERS: dict[str, type[BaseDownloader]] = {
...
"example": example.ExampleDownloader, # ← add this (if versioned)
}
``URLS_AND_DATE`` needed by
mapkgsutils.download.resolve_datasource_urls(). Maps the datasource name
to the function that returns the concrete download URLs and the release date for
a given version.
URLS_AND_DATE = {
...
"example": example.urls_and_date, # ← add this
}
Also add the import of the submodule and the downloader class at the top of
downloads/__init__.py, and include the class in __all__.
1. Tests
Put a small sample of the real upstream files in tests/data/ and write a
test that correctly parses them into mapping sets. The config tests in
tests/test_config.py already verify automatically that parser_class,
every method, and every inputs argument named in the config are real attributes on the parser class, and that each inputs key corresponds to a downloadable file.
Checklist
src/pysec2pri/config/example.yaml—parser_class,download_urls,mapping_setssrc/pysec2pri/parsers/example.py—ExampleParserwith every method the config namessrc/pysec2pri/parsers/__init__.py— parser class in_LAZY_EXPORTS, theTYPE_CHECKINGblock, and__all__src/pysec2pri/downloads/example.py— only if URLs are not fixedsrc/pysec2pri/downloads/__init__.py— downloader inCHECK_RELEASE,DOWNLOADERS,URLS_AND_DATE, and__all__(only if a downloader was written)tests/data/+tests/test_example.py— sample file and a parsing test