Sources

Load data-source definitions from data/sources.yaml.

The canonical source registry is a YAML file containing a flat list of mappings, one per SPARQL data source. Each mapping carries:

name - unique human-readable identifier.
endpoint - SPARQL endpoint URL.
graph_uris - named graphs to query.
use_graph - whether to wrap queries in a GRAPH clause.
two_phase - use two-phase mining (default True).
Optional tuning knobs: chunk_size, class_batch_size, class_chunk_size, timeout, delay, counts, unsafe_paging.

Each entry can be enriched with Bioregistry metadata via enrich_source_with_bioregistry(), which resolves the canonical Bioregistry prefix for the underlying dataset (regardless of how rdfsolve serialises or partitions it) and populates bioregistry_* fields.

The resolution strategy handles four cases:

Exact match — source name is itself a valid Bioregistry prefix (e.g. "chebi", "hgnc").
Root-prefix match — the first dot-separated segment of name resolves (e.g. "drugbank.drugs" → "drugbank").
local_provider field — the entry declares local_provider which is itself a Bioregistry prefix (e.g. local_provider: pubchem).
Extra-provider reverse lookup — source name follows the pattern "{provider}.{dataset}" (e.g. "bio2rdf.uniprot") and the dataset resource lists that provider code in its extra providers.

The full metadata dict returned by get_bioregistry_metadata() mirrors the fields shown on the Bioregistry resource page (name, description, homepage, license, domain, keywords, publications, uri_prefix, synonyms, mappings, extra_providers) and can be exported to JSON-LD with sources_to_jsonld().

Usage:

from rdfsolve.sources import load_sources, enrich_source_with_bioregistry

for src in load_sources("data/sources.yaml"):
    enrich_source_with_bioregistry(src)
    print(src["name"], src.get("bioregistry_name"))

class SourceEntry[source]

Bases: TypedDict

Typed dictionary for a single data-source definition.

name: str

endpoint: str

void_iri: str

graph_uris: list[str]

use_graph: bool

two_phase: bool

chunk_size: int

class_batch_size: int

class_chunk_size: int | None

timeout: float

delay: float

counts: bool

unsafe_paging: bool

notes: str

sparql_engine: str

sparql_strategy: str

supports_graph: bool

endpoint_down: bool

bioregistry_prefix: str

bioregistry_name: str

bioregistry_description: str

bioregistry_homepage: str

bioregistry_license: str

bioregistry_domain: str

keywords: list[str]

bioregistry_publications: list[dict[str, str | None]]

bioregistry_uri_prefix: str

bioregistry_uri_prefixes: list[str]

bioregistry_synonyms: list[str]

bioregistry_mappings: dict[str, str]

bioregistry_logo: str

bioregistry_extra_providers: list[dict[str, str | None]]

get_bioregistry_metadata(br_prefix: str) → dict[str, Any][source]

Return a structured metadata dict for a Bioregistry prefix.

The returned dictionary includes all fields visible on the Bioregistry resource page and suitable for embedding in JSON-LD or YAML:

{
    "prefix": "drugbank",
    "name": "DrugBank",
    "description": "...",
    "homepage": "http://www.drugbank.ca",
    "license": None,
    "domain": "chemical",
    "keywords": ["drug", "chemical structure", ...],
    "publications": [{"pubmed": "...", "doi": "...", "title": "..."}, ...],
    "uri_prefix": "https://go.drugbank.com/drugs/",
    "uri_prefixes": ["https://go.drugbank.com/drugs/", ...],
    "synonyms": ["DrugBank", "DRUGBANK_ID"],
    "mappings": {"wikidata": "P715", ...},
    "logo": "https://...",
    "extra_providers": [
        {"code": "bio2rdf", "name": "Bio2RDF",
         "uri_format": "http://bio2rdf.org/drugbank:$1"},
        ...
    ],
}

Parameters:: br_prefix – A valid Bioregistry prefix string.
Returns:: All available metadata; missing optional fields are omitted.
Return type:: dict
Raises:: ValueError – If br_prefix is not known to Bioregistry.

enrich_source_with_bioregistry(entry: SourceEntry) → str | None[source]

Populate bioregistry_* fields on entry in-place.

Resolves the canonical Bioregistry prefix for the source’s underlying dataset and writes all available metadata into the entry dict.

Parameters:: entry – A SourceEntry dict, modified in-place.
Returns:: The resolved Bioregistry prefix, or None if no match was found.
Return type:: str or None

Example

src = load_sources()[0]  # e.g. name="drugbank.drugs"
prefix = enrich_source_with_bioregistry(src)
print(prefix)  # "drugbank"
print(src["bioregistry_name"])  # "DrugBank"

sources_to_jsonld(entries: list[SourceEntry], *, enrich: bool = False) → dict[str, Any][source]

Serialise a list of source entries to a JSON-LD document.

Each entry becomes a node in the @graph array. Bioregistry-derived fields (bioregistry_*) are mapped to standard vocabulary predicates using a compact JSON-LD context.

Parameters:

entries – Source entries, typically returned by load_sources().
enrich – When True, call enrich_source_with_bioregistry() on each entry before serialisation (entries are not modified in place when enrich=True; a shallow copy is used per entry).

Returns:

A JSON-LD document with @context and @graph keys, ready for json.dump().

Return type:

dict

Example

import json
from rdfsolve.sources import load_sources, sources_to_jsonld

entries = load_sources()
doc = sources_to_jsonld(entries, enrich=True)
with open("sources.jsonld", "w") as f:
    json.dump(doc, f, indent=2)

load_sources(path: str | Path | None = None) → list[SourceEntry][source]

Load data-source definitions from a YAML, JSON-LD, or CSV file.

Parameters:: path – Path to the sources file. When None the default data/sources.yaml (or .jsonld / .csv fallback) is used.
Returns:: One dict per data source, keys normalised to snake_case. Sources without an endpoint are included (callers may skip them).
Return type:: list[SourceEntry]

load_sources_dataframe(path: str | Path | None = None, *, ports_json: str | Path | None = None) → DataFrame[source]

Load sources and return a DataFrame.

The DataFrame has columns compatible with probe_resource(): dataset_name, endpoint_url, graph_uri, use_graph, void_iri.

Parameters:

path – Path to the sources file. None = auto-detect default.
ports_json – Optional path to a QLever ports.json file mapping {dataset_name: port}. When supplied, endpoint_url is replaced with http://localhost:{port} for every dataset present in the file, and datasets not in the file are dropped. This ensures all queries go to local QLever instances instead of remote SPARQL endpoints.

classify_source_mode(entry: SourceEntry) → str[source]

Classify a source as 'local', 'remote', 'both', or 'unknown'.

Classification rules (in order):

'local' — at least one download_* field points to an RDF dump file (.ttl, .nq, .nt, .owl, etc.).
'remote' — endpoint is set, endpoint_down is not True, and no download links are present.
'both' — download links and a live endpoint are both present.
'unknown'— neither condition holds (no endpoint, no downloads).

Parameters:: entry – A SourceEntry dict.
Returns:: One of 'local', 'remote', 'both', 'unknown'.
Return type:: str