biomapper

Python client for the BioMapper2 API — map biological entity names to standardized knowledge-graph identifiers (CHEBI, HMDB, PubChem, RefMet, and more).

from biomapper import map_entity

result = map_entity("L-Histidine")
print(result.primary_curie)     # RM:0129894
print(result.confidence_tier)   # high
print(result.ids_for("CHEBI"))  # ['15971']
print(result.equivalent_ids_for("HMDB"))  # ['HMDB0000177']

Installation

# Core (async HTTP client + Pydantic models)
pip install biomapper

# With the external benchmark suite
pip install 'biomapper[benchmarks]'

1.5.2 is the first release on PyPI that contains the benchmark suite. Earlier published releases, up to and including 1.4.0, ship neither biomapper.benchmarks nor the benchmarks extra, so pip install 'biomapper[benchmarks]' against them warns that the extra is unknown and installs only the core client. The changelog records which versions reached the index; 1.5.0 and 1.5.1 are tagged (not published).

Getting an API key

The BioMapper2 API requires an API key. To request access, email trent.leslie@phenomehealth.org.

Once you have a key, set it in your environment:

export BIOMAPPER_API_KEY=your-key-here

Or add it to a .env file in your project root:

BIOMAPPER_API_KEY=your-key-here

biomapper will pick it up automatically from either location.

Quick start

Single lookup (synchronous)

from biomapper import map_entity

result = map_entity("L-Histidine")

print(result.resolved)          # True
print(result.primary_curie)     # RM:0129894
print(result.chosen_kg_id)      # CHEBI:15971
print(result.chosen_kg_id_review)  # None  (or 'divergent_refmet' when a ChEBI conflict is flagged for review)
print(result.confidence_score)  # 2.489
print(result.confidence_tier)   # high  (≥2.0)
print(result.ids_for("CHEBI"))  # ['15971']
print(result.ids_for("refmet_id"))  # ['RM0129894']

# KG equivalent IDs — all identifiers from the resolved knowledge graph node
print(result.kg_equivalent_ids)            # {'CHEBI': ['15971', '44637'], 'HMDB': ['HMDB0000177'], ...}
print(result.equivalent_ids_for("HMDB"))   # ['HMDB0000177']

Batch mapping (synchronous)

from biomapper import map_entities, summarize

records = [
    {"name": "L-Histidine"},
    {"name": "Glucose", "identifiers": {"HMDB": "HMDB00122"}},
    {"name": "Sphinganine"},
]

results = map_entities(records, progress=True)  # tqdm bar with [notebook]
summary = summarize(results)

print(f"{summary.resolved}/{summary.total_queried} resolved")
print(f"Resolution rate: {summary.resolution_rate:.1%}")
print(summary.vocabulary_coverage)

Inputs are auto-chunked at 1000 entities per request against the native POST /map/batch endpoint, so 10,000 records cost 10 round-trips.

Dataset upload (synchronous)

For larger inputs, hand the server a TSV/CSV file directly and stream results back. The server processes the file row-by-row over the POST /map/dataset/stream endpoint:

from pathlib import Path
from biomapper import map_dataset_file_sync

result = map_dataset_file_sync(
    Path("compounds.tsv"),
    name_column="name",
    provided_id_columns=["hmdb_id"],
    progress=True,         # tqdm bar
    total_hint=1000,       # optional; enables % progress
)
result.raise_for_error()   # opt-in: raise BioMapperError if the stream truncated
print(f"resolved {sum(1 for r in result.results if r.resolved)} of {len(result.results)}")

name_column and provided_id_columns are required — the server uses them to map your file's columns to entity names and identifier hints. For per-result streaming into a UI or custom processing, use the async BioMapperClient.map_dataset_file_iter method (see the tutorial notebook in notebooks/).

Discovering what the API supports

from biomapper import list_annotators, list_vocabularies, list_entity_types

for a in list_annotators():
    print(f"{a.slug:30s} {a.name}")

# 300+ supported vocabularies (CHEBI, HMDB, PubChem, …)
vocabs = list_vocabularies()
print(f"{len(vocabs)} vocabularies supported")

# Biolink entity types with aliases and default vocabulary prefixes
for et in list_entity_types():
    print(f"{et.type}: {', '.join(et.aliases)}")
    if et.default_prefixes:
        print(f"  prefixes: {', '.join(et.default_prefixes)}")

Tuning resolution

The mapping calls accept the API's resolution options as keyword arguments. An option you do not pass is omitted from the request, so the server's own default applies and the payload is unchanged for callers who ignore them.

Coverage is uneven, so check this before reaching for one. Passing an option to a call that does not accept it raises TypeError locally, before any request:

array_delimiters is absent from the dataset routes because the API does not accept it there. The four missing from map_dataset_file_sync are a gap in that wrapper rather than an API limitation: the async map_dataset_file_iter underneath it does accept them, so use that directly if you need them on a file-based run.

from biomapper import map_entity  # map_entity / map_entities accept all six

result = map_entity(
    "PC 34:1",
    vocab="refmet",             # restrict to one vocabulary (or a list)
    prefer_human=False,         # gene/protein: allow a non-human ortholog to win
    prefer_canonical=False,     # non-gene: allow a non-canonical-namespace node to win
    candidate_limit=20,         # candidates each Kestrel search annotator retrieves (1..100)
    array_delimiters=["|"],     # how delimited ID strings are split
    kestrel_top_n=5,            # opt in to raw Kestrel passthrough rows (1..100)
)

kestrel_top_n populates result.kestrel_results with the raw rows Kestrel returned, exactly as returned, for each search endpoint the pipeline used. Those rows are untrusted external data: treat every field as unescaped and unverified.

Resolution certificates and refusal

A mapping result carries the structural certificate the API computed for chosen_kg_id, which is how you tell "we could not check this" apart from "we checked and disagree".

result = map_entity("cortisone")

cert = result.certificate
if cert is not None:
    print(cert.state)                    # corroborated | uncorroborated | contradicted |
                                         # unavailable | not_applicable
    print(cert.independent_source)       # registry consulted, e.g. "pubchem"
    print(cert.independent_of_selection)  # False => corroboration would be circular
    print(cert.refmet_availability)      # voted | no_match | unavailable | not_queried

# Why the resolver declined, when it did. Distinct from `error`:
# an error means the call failed; a refusal means the pipeline ran and declined to assert.
print(result.refusal_reason)

state="contradicted" means a human should look, not that the resolver is wrong: a name lookup at an external registry can itself return a related-but-different compound. state="unavailable" means there was nothing to check against, which is unverifiable rather than wrong.

Async usage

import asyncio
from biomapper import BioMapperClient

async def main() -> None:
    async with BioMapperClient() as client:
        # Verify connectivity
        health = await client.health_check()
        print(health)  # {'status': 'healthy', ...}

        # Single
        result = await client.map_entity(
            "L-Histidine",
            identifiers={"HMDB": "HMDB00177"},
        )

        # Batch — auto-chunked at 1000 entities per request
        results = await client.map_entities(
            [{"name": "L-Histidine"}, {"name": "Glucose"}],
            progress=True,
        )

        # Stream from a file — per-result as they arrive
        from pathlib import Path
        async for r in client.map_dataset_file_iter(
            Path("compounds.tsv"),
            name_column="name",
            provided_id_columns=["hmdb_id"],
        ):
            print(r.query_name, r.primary_curie)

asyncio.run(main())

map_dataset_file_iter is the primitive for UIs and custom processing that want per-result reactivity. Callers needing a blocking, fully-collected result should use map_dataset_file_sync instead (see above).

Jupyter notebooks

Apply nest_asyncio before using sync helpers inside a running event loop:

import nest_asyncio
nest_asyncio.apply()  # required in Jupyter

from biomapper import map_entities
results = map_entities([{"name": "L-Histidine"}], progress=True)

Preprocessing functions

from biomapper.extras.metabolon import clean_compound_name, extract_hmdb_id

# Strip quotes and collision-energy suffixes
clean_compound_name('"1,3-Diphenylguanidine_CE45"')  # '1,3-Diphenylguanidine'
clean_compound_name('"4,6-DIOXOHEPTANOIC ACID"')     # '4,6-DIOXOHEPTANOIC ACID'
clean_compound_name('L-Histidine')                   # 'L-Histidine'  (unchanged)

# Extract HMDB accessions from ms1_compound_name format
extract_hmdb_id('HMDB:HMDB03349-2257 L-Dihydroorotic acid')  # 'HMDB03349'
extract_hmdb_id('HMDB00177')                                  # 'HMDB00177'
extract_hmdb_id(None)                                         # None

Harmonization (cross-dataset equivalence)

biomapper.harmonize links two already-resolved datasets locally. Two entities, one per cohort, are equivalent when they resolve to the same canonical KRAKEN node. It is an identifier-set intersection, never string matching, and it runs entirely on the client: no extra requests, no knowledge-graph access, so it works offline and is fully testable without a network.

from biomapper import map_entities
from biomapper.harmonize import harmonize

ukbb    = map_entities([{"name": "Glucose"},   {"name": "Urea"}])
arivale = map_entities([{"name": "D-glucose"}, {"name": "X-12345"}])

report = harmonize(ukbb, arivale, a_label="ukbb", b_label="arivale")

report.n_links          # 1
report.links[0].shared  # frozenset({'CHEBI:17234', 'KEGG:C00031'}) — what formed the link
report.b_unresolved     # ('X-12345',) — a refusal candidate, never silently dropped
report.summary()

Two rules the linker is built around:

1. Identifier-only. INCHIKEY, INCHI and SMILES are excluded. Linking on a structure hash would make any downstream structural certificate circular and would make precision 100% by construction.
2. Prefix synonyms normalize. KEGG.COMPOUND:C00031 equals KEGG:C00031; genuinely different identifier spaces such as KEGG.GLYCAN stay distinct.

An entity that resolved to nothing is a refusal candidate, not a link. It is named in a_unresolved / b_unresolved, counted in summary(), and left out of the link-rate denominator so non-resolution is never scored as non-equivalence. An entity whose mapping call errored is tracked separately again in a_errors / b_errors.

Import it as from biomapper.harmonize import harmonize. The name is deliberately not bound on the package root, where it would shadow the submodule.

API reference

MappingResult

result.ids_for("CHEBI")        # ['15971']
result.ids_for("refmet_id")    # ['RM0129894']
result.ids_for("PUBCHEM.COMPOUND")  # []

# KG equivalent IDs — all identifiers from the resolved knowledge graph node
result.equivalent_ids_for("HMDB")  # ['HMDB0000177']
result.equivalent_ids_for("LM")   # ['ST01010001', 'ST01010093']

DatasetMappingResult

Return type of map_dataset_file_sync. Captures per-row results plus an opt-in error signal for partial runs.

result.raise_for_error()   # raises BioMapperError if .error is set; else no-op

raise_for_error mirrors httpx.Response.raise_for_status and turns the partial-result contract into an explicit caller opt-in — silent consumption of a truncated run (using .results without checking .error) is the footgun this model is designed to prevent.

Note: confidence_score on dataset-stream results is always None — the /map/dataset/stream endpoint emits a slimmer per-row payload than /map/batch and does not include the annotator assigned_ids block. Use map_entity / map_entities if you need confidence tiers.

Confidence tiers

Error handling

from biomapper import (
    BioMapperError,       # base class
    BioMapperAuthError,   # 401/403 — bad API key
    BioMapperRateLimitError,  # 429 — throttled
    BioMapperServerError,     # 5xx
    BioMapperTimeoutError,    # request timeout
    BioMapperConfigError,     # missing API key / bad config
)

try:
    result = map_entity("Glucose")
except BioMapperRateLimitError as e:
    print(f"Throttled. Retry after: {e.retry_after}s")
except BioMapperAuthError:
    print("Check your BIOMAPPER_API_KEY")

In batch mode (map_entities), per-record errors are caught and returned as MappingResult(error=...) rather than aborting the batch.

Dataset streaming (map_dataset_file_sync) uses a two-tier contract:

• Initial-request errors (401, 422, 500, connect timeout) raise as typed exceptions — these happen before any row is processed, so partial results don't exist to preserve.
• Mid-stream transport failures are captured into DatasetMappingResult.error with the partial results preserved in .results. Call .raise_for_error() to get exception semantics, or inspect .error directly for "accept partial, log the rest" workflows.

Callback exceptions raised from on_result propagate unwrapped and replace the return value — partial results collected up to that point are lost. For UI consumers with failure-prone callbacks, wrap the callback body in your own try/except if you want partial data to survive.

External benchmark suite

An API-only reproduction harness for the benchmarks the BioMapper preprint reports. It runs against a deployment, not a local engine checkout, which matters for two reasons: running in-process against a checkout measures a library rather than the deployed service, and pointing at a deployment pins provenance to the backend that actually served the answers.

pip install 'biomapper[benchmarks]'

python -m biomapper.benchmarks list                 # the 11 arms and the 2 deliberate skips
python -m biomapper.benchmarks all                  # all arms, production endpoint
python -m biomapper.benchmarks arm hajjar           # a single arm
python -m biomapper.benchmarks --endpoint dev all   # dev, for testing

The public KRAKEN endpoint is keyless, and a BioMapper2 deployment with no keys configured is open, so the suite runs unauthenticated by default. Set BIOMAPPER_API_KEY if your deployment requires one; the key is never accepted on the command line, because argv is visible to other processes and lands in shell history.

The arms

Read the Reportable as column before quoting anything. Four of the graph's ingested sources (lipidmaps, refmet, loinc, ncbigene) are gold sources for arms above, so those arms measure coverage, not independent accuracy: the gold identifier and BioMapper's answer come from the same place. Each run labels every arm from the build's own source list rather than from a static note, so the label tracks the graph.

Two further reporting rules the suite enforces rather than documents:

• Gene arms report accuracy per target namespace. The any-namespace roll-up is emitted flagged quotable: false, because the namespaces perform very differently and the roll-up has been quoted as though it described all of them.
• A skipped arm is not a zero and not a pass. It carries a reason into the manifest.

Provenance

Every run records what actually served it, read from Kestrel /health (keyless) rather than hardcoded: kestrel_version, kg_version, biolink_version, build_timestamp, git_commit, the full source_versions map, the package version, each dataset's source SHA, and a run id. The build is read once before the arms start and re-read afterwards; if it moved mid-suite, the manifest says so, because the pins would no longer describe every number.

Results are saved by default to a timestamped directory (override with --out). There is no flag that discards them: the expensive part of a run is live API traffic.

What stayed in the engine repo

The CI regression gates (gate.py, conflation_gate.py, test_regression_gate.py, kg-regression.yml) guard merges and need engine internals. The --resolver-mode {weighted,vote} A/B toggles a resolver constructor argument that is deliberately not on the API surface.

Development

git clone https://github.com/trentleslie/biomapper
cd biomapper
poetry install --with dev --extras all

make check          # format → lint → type-check → test
make test           # tests only
make coverage       # HTML coverage report

docs/solutions/ holds documented solutions to past problems — bugs, best practices, and workflow patterns — organized by category with YAML frontmatter (module, tags, problem_type). Relevant when implementing or debugging in an area something has already been written about.

License

MIT — see LICENSE.

Related

• BioMapper2 API: https://biomapper.expertintheloop.io