RDFSolve

https://rdfsolve.readthedocs.io/en/latest/
Admin
RDFSolve is a Python toolkit for RDF schema mining, SPARQL endpoint exploration, Linked Open Data...
Tokens:120,730
Snippets:774
Trust Score:9.6
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Install rdfsolve from PyPI

Source: https://rdfsolve.readthedocs.io/en/latest/_sources/installation.rst.txt

Install the most recent release from PyPI using uv.

```shell
uv pip install rdfsolve
```

--------------------------------

### Install rdfsolve from GitHub

Source: https://rdfsolve.readthedocs.io/en/latest/_sources/installation.rst.txt

Install the most recent code and data directly from the GitHub repository using uv.

```shell
uv pip install git+https://github.com/jmillanacosta/rdfsolve.git
```

--------------------------------

### Install rdfsolve in development mode

Source: https://rdfsolve.readthedocs.io/en/latest/_sources/installation.rst.txt

Clone the repository and perform an editable install for development.

```shell
git clone git+https://github.com/jmillanacosta/rdfsolve.git
cd rdfsolve
uv pip install -e .
```

--------------------------------

### make_valid_linkml_name conversion examples

Source: https://rdfsolve.readthedocs.io/en/latest/schema_models.html

Shows example conversions from URI or CURIE to valid LinkML identifiers. LinkML identifiers must start with a letter and contain only letters, digits, and underscores.

```text
"aopo:KeyEvent"           -> "aopo_KeyEvent"
"edam.data1025"           -> "edam_data_1025"
"http://example.org/Cls"  -> prefix_Cls  (via bioregistry)
"40:Something"            -> "item_40_Something"
```

--------------------------------

### GET and POST query wrappers (_get_query, _post_query)

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/sparql_helper.html

Convenience wrappers that call _request with the HTTP method 'GET' or 'POST'. _get_query sends a bounded GET request, _post_query sends a bounded form POST request.

```python
def _get_query(self, query: str, accept: str) -> str:
    """Send a bounded GET request."""
    return self._request("GET", query, accept)
```

```python
def _post_query(self, query: str, accept: str) -> str:
    """Send a bounded form POST request."""
    return self._request("POST", query, accept)
```

--------------------------------

### load_shacl

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/sparql_helper.html

Loads local query examples and paths. Does not execute or fetch imports.

```APIDOC
## load_shacl

### Description
Loads local query examples and paths. Does not execute or fetch imports.

### Method
load_shacl

### Signature
```python
def load_shacl(self, source: str | Path | Graph) -> list[str]
```

### Parameters
- **source** (str | Path | Graph) - Required - The source from which to load query examples. Can be a string, path, or graph.

### Returns
- **list[str]** - A list of loaded query strings.

```

--------------------------------

### SparqlHelper usage example

Source: https://rdfsolve.readthedocs.io/en/latest/sparql_helper.html

Shows how to instantiate SparqlHelper with a SPARQL endpoint URL and execute a SELECT query, iterating over the bindings in the results.

```python
>>> helper = SparqlHelper("https://sparql.swisslipids.org/")
>>> results = helper.select("SELECT ?g { GRAPH ?g { ?s ?p ?o } }")
>>> for binding in results["results"]["bindings"]:
...     print(binding["g"]["value"])
```

--------------------------------

### Imports and logger setup

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/void_discover.html

Imports required modules and sets up a logger with a NullHandler to avoid output unless the user configures logging.

```python
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from rdfsolve.schema_models.metadata import MetadataDocument

import pandas as pd
from linkml_runtime.linkml_model import SchemaDefinition
from rdflib import Graph, URIRef

from rdfsolve.schema_models.core import MinedSchema

# Create logger with NullHandler by default , no output unless user configures
logger = logging.getLogger(__name__)
if not logger.handlers:
    logger.addHandler(logging.NullHandler())
```

--------------------------------

### get_local_name() example

Source: https://rdfsolve.readthedocs.io/en/latest/uri.html

Shows the expected output for two example URIs: one using a '#' separator and one using a '/' separator. Both return the local name 'Bar'.

```python
>>> get_local_name("http://example.org/foo#Bar")
'Bar'
>>> get_local_name("http://example.org/foo/Bar")
'Bar'
```

--------------------------------

### Convert VoID to RDF-config with rdfsolve.to_rdfconfig_from_file

Source: https://rdfsolve.readthedocs.io/en/latest/_sources/usage.rst.txt

Convert a VoID file to RDF-config using to_rdfconfig_from_file. The example specifies an endpoint URL for the configuration.

```python
files = rdfsolve.to_rdfconfig_from_file(
    "dataset_void.ttl",
    endpoint_url="https://sparql.example.org",
)
```

--------------------------------

### Get Bioregistry metadata for a prefix

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/sources.html

Returns a structured metadata dict for a Bioregistry prefix. Requires the bioregistry package; raises ImportError if not installed and ValueError if the prefix is unknown. Missing optional fields are omitted from the result.

```python
def get_bioregistry_metadata(br_prefix: str) -> dict[str, Any]:
    """Return a structured metadata dict for a Bioregistry prefix.

    Parameters
    ----------
    br_prefix:
        A valid Bioregistry prefix string.

    Returns
    -------
    dict
        All available metadata; missing optional fields are omitted.

    Raises
    ------
    ValueError
        If *br_prefix* is not known to Bioregistry.
    """
    try:
        import bioregistry
    except ImportError as exc:
        raise ImportError("bioregistry must be installed for metadata lookup") from exc

    resource = bioregistry.get_resource(br_prefix)
    if resource is None:
        raise ValueError(f"Unknown Bioregistry prefix: {br_prefix!r}")

    meta: dict[str, Any] = {"prefix": br_prefix}
    _extract_scalar_metadata(resource, meta)
    _extract_collection_metadata(resource, meta)
    return meta
```

--------------------------------

### Manage mining session with graph store download

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/miner.html

Context manager that initializes a report, optionally downloads graphs from a store, and ensures cleanup. On exception, sets abort reason, finishes unfinished phases, finalises the report, and re-raises. Restores the original helper in finally.

```python
@contextmanager
def _session(self, dataset_name: str | None) -> Iterator[None]:
    """Start one report before any phase and retain failures."""
    self._ontology_classes = None
    self._declared_classes = set()
    self._init_report(
        dataset_name, self._build_strategy_string(), datetime.now(timezone.utc).isoformat()
    )
    self.last_report = self._report.report
    remote_helper = self._helper
    try:
        if self.get_graphs_from_store:
            from dataclasses import asdict

            from rdfsolve.graph_store import download_graphs, load_downloads
            from rdfsolve.mining.local_graph import LocalGraphHelper

            downloads = download_graphs(
                self.graph_store_url or "",
                self.graph_uris or [],
                self.graph_store_dir,
                max_bytes=self.graph_store_max_bytes,
                timeout=self.timeout,
            )
            self._helper = LocalGraphHelper(
                self.endpoint_url,
                load_downloads(downloads, endpoint_url=self.endpoint_url, timeout=self.timeout),
            )
            self._report.report.config["graph_store"] = {
                "url": self.graph_store_url,
                "engine": "rdflib",
                "graphs": [asdict(item) for item in downloads],
            }
        yield
    except BaseException as exc:
        reason = f"{type(exc).__name__}: {exc}"
        self._report.set_abort_reason(reason)
        report = self._report.report
        for phase in report.phases:
            if phase.finished_at is None:
                self._report.finish_phase(phase, error=reason)
        self._report.finalise(
            pattern_count=report.pattern_count,
            class_count=report.class_count,
            property_count=report.property_count,
            uris_labelled=report.unique_uris_labelled,
        )
        raise
    finally:
        if self._helper is not remote_helper:
            self._helper.close()
            self._helper = remote_helper
```

--------------------------------

### Convert schema to SHACL and print first 100 characters

Source: https://rdfsolve.readthedocs.io/en/latest/schema_models.html

Shows how to load a schema from JSON-LD, convert it to SHACL Turtle, and print the first 100 characters. The output begins with the SHACL prefix declaration.

```python
>>> schema = MinedSchema.from_jsonld("schema.jsonld")
>>> shacl_ttl = schema.to_shacl()
>>> print(shacl_ttl[:100])
@prefix sh: <http://www.w3.org/ns/shacl#> .
```

--------------------------------

### to_rdfconfig

Source: https://rdfsolve.readthedocs.io/en/latest/schema_models.html

Exports RDF-config from canonical patterns and source examples.

```APIDOC
## to_rdfconfig

### Description
Export RDF-config from canonical patterns and source examples.

### Method
Function

### Signature
`to_rdfconfig(*, endpoint_url: str | None = None, endpoint_name: str | None = None, graph_uri: str | None = None, trim_descriptions: int | None = None) -> dict[str, str]`

### Parameters
#### Keyword Arguments
- **endpoint_url** (str | None) - Optional - URL of the SPARQL endpoint.
- **endpoint_name** (str | None) - Optional - Name of the endpoint.
- **graph_uri** (str | None) - Optional - URI of the graph.
- **trim_descriptions** (int | None) - Optional - If provided, trims descriptions to the given length.

### Returns
- **dict[str, str]** - RDF-config dictionary.
```

--------------------------------

### SchemaMiner.query_enrichment()

Source: https://rdfsolve.readthedocs.io/en/latest/miner.html

Queries definitions and observed examples with the miner's settings.

```APIDOC
## SchemaMiner.query_enrichment()

### Description
Queries definitions and observed examples with this miner's settings. Assign the return value to schema.enrichment when called after mining. No ontology or external vocabulary download is attempted.

### Method
query_enrichment(schema: MinedSchema, *, annotation_iris: list[str] | None = None) -> SchemaEnrichment

### Parameters
- **schema** (MinedSchema) - Required - The mined schema to enrich.
- **annotation_iris** (list[str] | None) - Optional - List of annotation IRIs to use for enrichment.

### Returns
SchemaEnrichment - The enrichment result.
```

--------------------------------

### Export mapping to networkx MultiDiGraph

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/mapping_models/core.html

Exports the mapping as an ``nx.MultiDiGraph``. Requires the ``networkx`` package; if missing, raises an ImportError with instructions to install it via ``pip install networkx``. Adds nodes for source and target classes with their datasets, and edges with predicate, datasets, mapping type, and confidence attributes.

```python
def to_networkx(self) -> Any:
    """Export the mapping as an ``nx.MultiDiGraph``."""
    try:
        import networkx as _nx
    except ImportError as exc:
        raise ImportError(
            "networkx is required for to_networkx(); install it with: pip install networkx",
        ) from exc

    graph: Any = _nx.MultiDiGraph()
    for edge in self.edges:
        for uri, ds in (
            (edge.source_class, edge.source_dataset),
            (edge.target_class, edge.target_dataset),
        ):
            if uri not in graph:
                graph.add_node(uri, dataset=ds)
        graph.add_edge(
            edge.source_class,
            edge.target_class,
            predicate=edge.predicate,
            source_dataset=edge.source_dataset,
            target_dataset=edge.target_dataset,
            strategy=self.mapping_type,
            confidence=edge.confidence,
        )
    return graph
```

--------------------------------

### Module imports and __all__ for rdfsolve.instance_matcher

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/instance_matcher.html

Module imports and metadata for rdfsolve.instance_matcher, including the __all__ list that exposes probe_endpoint and probe_resource.

```python
"""Probe SPARQL endpoints to discover entity classes across datasets using URI pattern matching."""

from __future__ import annotations

import logging
from typing import Any

from rdfsolve.mapping_models.core import (
    SKOS_NARROW_MATCH,
    InstanceMatchResult,
    MappingEdge,
)
from rdfsolve.mapping_models.instance import InstanceMapping
from rdfsolve.schema_models.about import AboutMetadata
from rdfsolve.sparql_helper import SparqlHelper

_log = logging.getLogger(__name__)

__all__ = ["probe_endpoint", "probe_resource"]
```

--------------------------------

### SparqlHelper.from_source_entry()

Source: https://rdfsolve.readthedocs.io/en/latest/sparql_helper.html

Creates a SparqlHelper instance from a source entry.

```APIDOC
## SparqlHelper.from_source_entry()

### Description
Creates a SparqlHelper instance from a source entry.

### Method
from_source_entry

### Signature
from_source_entry(entry: dict) -> SparqlHelper

### Parameters
- **entry** (dict) - Required - A dictionary representing a source entry.

### Returns
- A new SparqlHelper instance configured from the source entry.
```

--------------------------------

### get_local_name function

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/_uri.html

Extracts the local name from a URI by splitting on '#' or the last '/'. Includes doctest examples.

```python
def get_local_name(uri: str) -> str:
    """Extract the local name from a URI.

    Examples::

        >>> get_local_name("http://example.org/foo#Bar")
        'Bar'
        >>> get_local_name("http://example.org/foo/Bar")
        'Bar'
    """
    if "#" in uri:
        return uri.split("#")[-1]
    return uri.rstrip("/").rsplit("/", 1)[-1] if "/" in uri else uri
```

--------------------------------

### Execute SPARQL query with execute_sparql and Pydantic models

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/query.html

Defines the Pydantic models ResultCell and QueryResult, and the execute_sparql function that runs a SPARQL SELECT query against an endpoint. The function returns a QueryResult with rows parsed into ResultCell objects, and on exception returns a QueryResult with an error string. The method parameter defaults to GET, and if GET fails the SparqlHelper automatically retries with POST.

```python
"""Structured SPARQL query execution with Pydantic result models."""

from __future__ import annotations

import time
from typing import Any

from pydantic import BaseModel, Field

from rdfsolve.sparql_helper import SparqlHelper

# Result models




[docs]
class ResultCell(BaseModel):
    """One cell in a SPARQL result row."""

    value: str
    type: str  # "uri" | "literal" | "bnode"
    lang: str | None = None
    datatype: str | None = None






[docs]
class QueryResult(BaseModel):
    """Structured result from a SPARQL query execution."""

    query: str
    endpoint: str
    variables: list[str]
    rows: list[dict[str, ResultCell]]
    variable_map: dict[str, str] = Field(default_factory=dict)
    row_count: int
    duration_ms: int
    error: str | None = None




# Public helper




[docs]
def execute_sparql(
    query: str,
    endpoint: str,
    *,
    method: str = "GET",
    timeout: int = 30,
    variable_map: dict[str, str] | None = None,
) -> QueryResult:
    """Execute a SPARQL SELECT query and return a :class:`QueryResult`.

    Parameters
    ----------
    query:
        Full SPARQL query string.
    endpoint:
        URL of the SPARQL endpoint.
    method:
        HTTP method (``"GET"`` or ``"POST"``).  If ``"GET"`` fails the
        underlying :class:`SparqlHelper` will automatically retry with
        POST.
    timeout:
        Request timeout in seconds.
    variable_map:
        Optional mapping of SPARQL ``?variable`` names to schema URIs.

    Returns
    -------
    QueryResult
        Pydantic model with ``query``, ``endpoint``, ``variables``,
        ``rows``, ``variable_map``, ``row_count``, ``duration_ms``, and
        optionally ``error``.
    """
    t0 = time.monotonic()

    try:
        helper = SparqlHelper(
            endpoint,
            use_post=(method.upper() == "POST"),
            timeout=float(timeout),
        )
        json_result = helper.select(query, purpose="user-query")
    except Exception as exc:
        return QueryResult(
            query=query,
            endpoint=endpoint,
            variables=[],
            rows=[],
            variable_map=variable_map or {},
            row_count=0,
            duration_ms=int((time.monotonic() - t0) * 1000),
            error=str(exc),
        )

    variables: list[str] = json_result.get("head", {}).get("vars", [])
    bindings: list[dict[str, Any]] = json_result.get("results", {}).get("bindings", [])

    rows: list[dict[str, ResultCell]] = []
    for binding in bindings:
        row: dict[str, ResultCell] = {}
        for var in variables:
            cell_data = binding.get(var)
            if cell_data:
                cell_type = cell_data.get("type", "literal")
                if cell_type == "uri":
                    rtype = "uri"
                elif cell_type == "bnode":
                    rtype = "bnode"
                else:
                    rtype = "literal"
                row[var] = ResultCell(
                    value=cell_data["value"],
                    type=rtype,
                    lang=cell_data.get("xml:lang"),
                    datatype=cell_data.get("datatype"),
                )
        rows.append(row)

    duration_ms = int((time.monotonic() - t0) * 1000)

    return QueryResult(
        query=query,
        endpoint=endpoint,
        variables=variables,
        rows=rows,
        variable_map=variable_map or {},
        row_count=len(rows),
        duration_ms=duration_ms,
    )

```

--------------------------------

### SparqlHelper.construct()

Source: https://rdfsolve.readthedocs.io/en/latest/sparql_helper.html

Executes a SPARQL CONSTRUCT query and returns the resulting RDF graph as a string.

```APIDOC
## SparqlHelper.construct()

### Description
Executes a SPARQL CONSTRUCT query and returns the resulting RDF graph as a string.

### Method
construct

### Signature
construct(query: str, **kwargs) -> str

### Parameters
- **query** (str) - Required - The SPARQL CONSTRUCT query string.
- **kwargs** - Optional - Additional arguments passed to run_query.

### Returns
- A string representation of the constructed RDF graph.
```

--------------------------------

### Get metadata

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/void_discover.html

Returns a MetadataDocument containing the retained RDF graph and graph URIs without querying an endpoint.

```python
def get_metadata(self) -> MetadataDocument:
    """Return retained RDF metadata without querying an endpoint."""
    from rdfsolve.schema_models.metadata import MetadataDocument

    return MetadataDocument(
        graph=self.graph + Graph(), graph_uris=self.graph_uris, scope="retained VoID RDF"
    )
```

--------------------------------

### to_linkml_from_file function

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/api.html

Converts a VoID file to a LinkML YAML schema string. Accepts optional schema name, description, and base URI.

```python
[docs]
def to_linkml_from_file(
    void_file_path: str,
    schema_name: str | None = None,
    schema_description: str | None = None,
    schema_base_uri: str | None = None,
) -> str:
    """Convert a VoID file to LinkML YAML schema.

    Args:
        void_file_path: Path to VoID file
        schema_name: Name for the schema
        schema_description: Description for the schema
        schema_base_uri: Base URI for the schema

    Returns:
        LinkML YAML schema string
    """
    parser = load_parser_from_file(void_file_path)
    return parser.to_linkml_yaml(
        schema_name=schema_name,
        schema_description=schema_description,
        schema_base_uri=schema_base_uri,
    )
```

--------------------------------

### Convert VoID to LinkML YAML with rdfsolve.to_linkml_from_file

Source: https://rdfsolve.readthedocs.io/en/latest/_sources/usage.rst.txt

Convert a VoID file to LinkML YAML using to_linkml_from_file.

```python
linkml_yaml = rdfsolve.to_linkml_from_file("dataset_void.ttl")
```

--------------------------------

### SPARQL HTTP fallback: GET to POST to raw POST

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/sparql_helper.html

Fallback logic for SPARQL endpoint requests: if a GET returns HTML or an HTTP error (400, 405, 414), retry with form-encoded POST; if that returns HTML or an error (400, 405, 415), retry with raw POST body. Raises EndpointError if HTML persists even with POST. Logs fallback usage and sets self._requires_post and record.fallback_used.

```python
                # Check if we got HTML instead of expected format
                if self._is_html_response(result):
                    if not use_post and not _use_raw_post:
                        logger.info("Fallback: GET returned HTML; use POST")
                        fallback_used = True
                        if record is not None:
                            record.fallback_used = True
                        self._requires_post = True
                        use_post = True
                        continue
                    elif use_post and not _tried_raw_post:
                        logger.info("Fallback: form POST returned HTML; use raw POST")
                        fallback_used = True
                        if record is not None:
                            record.fallback_used = True
                        _use_raw_post = True
                        _tried_raw_post = True
                        continue
                    else:
                        raise EndpointError(
                            f"{purpose} | Endpoint returned HTML error even with POST"
                        )

                parsed = json.loads(result) if parse_json else result

                logger.info(
                    "%s completed (%s)",
                    query_type,
                    "HTTP fallback used" if fallback_used else "no HTTP fallback",
                )
                return parsed
```

```python
            except requests.exceptions.HTTPError as e:
                status_code = e.response.status_code if e.response is not None else 0

                # Check if this looks like a POST-required error
                # 400 = Bad Request (QLever rejects GET), 405 = Method Not Allowed,
                # 414 = URI Too Long
                if not use_post and not _use_raw_post and status_code in (400, 405, 414):
                    logger.info(
                        "Fallback: GET returned %d; use POST",
                        status_code,
                    )
                    fallback_used = True
                    if record is not None:
                        record.fallback_used = True
                    self._requires_post = True
                    use_post = True
                    continue

                # Form-encoded POST rejected -> try raw POST body
                if use_post and not _tried_raw_post and status_code in (400, 405, 415):
                    logger.info(
                        "Fallback: form POST returned %d; use raw POST",
                        status_code,
                    )
                    fallback_used = True
                    if record is not None:
                        record.fallback_used = True
                    _use_raw_post = True
                    _tried_raw_post = True
                    continue
```

--------------------------------

### Load graph from file

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/void_discover.html

Loads the VoID file into the RDF graph, detecting the format from the file extension and defaulting to turtle.

```python
def _load_graph(self) -> None:
    """Load the VoID file into an RDF graph.

    Automatically detects format from file extension:
    - .ttl -> turtle
    - .jsonld, .json -> json-ld
    - .xml, .rdf -> xml
    - .nt -> ntriples
    """
    from pathlib import Path

    if not self.void_file_path:
        return

    # Detect format from extension
    path = Path(self.void_file_path)
    suffix = path.suffix.lower()

    format_map = {
        ".ttl": "turtle",
        ".jsonld": "json-ld",
        ".json": "json-ld",
        ".xml": "xml",
        ".rdf": "xml",
        ".nt": "ntriples",
        ".nq": "nquads",
    }

    fmt = format_map.get(suffix, "turtle")  # default to turtle
    self.graph.parse(self.void_file_path, format=fmt)
```

--------------------------------

### load_sources(sources_file: str | Path) -> list[SourceEntry]

Source: https://rdfsolve.readthedocs.io/en/latest/api.html

Load sources from a YAML file.

```APIDOC
## load_sources

### Description
Load sources from a YAML file.

### Signature
```python
load_sources(sources_file: str | Path) -> list[SourceEntry]
```

### Parameters
- **sources_file** (str | Path) - Required - Path to the sources YAML file.

### Returns
- **list[SourceEntry]** - List of source entries.
```

--------------------------------

### InstanceMapping.to_jsonld method

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/mapping_models/instance.html

Extends the base JSON-LD output with instance-matcher provenance, adding resource, uri_formats_queried, and strategy fields to the @about object.

```python
[docs]
    def to_jsonld(self) -> dict[str, Any]:
        """Extend base JSON-LD with instance-matcher provenance."""
        doc = super().to_jsonld()
        about = doc.get("@about", {})
        about["resource"] = self.resource_prefix
        about["uri_formats_queried"] = self.uri_formats
        about["strategy"] = "instance_matcher"
        doc["@about"] = about
        return doc
```

--------------------------------

### Parse VoID Turtle into MinedSchema

Source: https://rdfsolve.readthedocs.io/en/latest/schema_models.html

Example usage of void_to_minedschema: reads a VoID Turtle file and prints the number of patterns in the resulting MinedSchema.

```python
>>> void_ttl = Path("dataset_void.ttl").read_text()
>>> schema = void_to_minedschema(void_ttl)
>>> print(len(schema.patterns))
435
```

--------------------------------

### discover_void_source(endpoint, name, output_dir=None, *, tag='discovered_remote', fmt='all', timeout=30.0, max_retries=1, graph_uris=None, batch_size=100, graph_batch_size=8, max_pages=1000, get_graphs_from_store=False, graph_store_url=None, graph_store_dir='graph-store', graph_store_max_bytes=67108864) -> VoidSchema

Source: https://rdfsolve.readthedocs.io/en/latest/api.html

Return published VoID as an object. Export only when output_dir is set. Use .to_mined_schema() for canonical patterns, .datasets for typed VoID descriptions, and .graph for all retrieved RDF.

```APIDOC
## discover_void_source

### Description
Return published VoID as an object. Export only when output_dir is set. Use .to_mined_schema() for canonical patterns, .datasets for typed VoID descriptions, and .graph for all retrieved RDF. Empty patterns do not imply an empty description. Graph Store mode requires explicit scope.

### Signature
```python
discover_void_source(endpoint: str, name: str, output_dir: str | Path | None = None, *, tag: str = 'discovered_remote', fmt: str = 'all', timeout: float = 30.0, max_retries: int = 1, graph_uris: str | list[str] | None = None, batch_size: int = 100, graph_batch_size: int = 8, max_pages: int = 1000, get_graphs_from_store: bool = False, graph_store_url: str | None = None, graph_store_dir: str | Path = 'graph-store', graph_store_max_bytes: int = 67108864) -> VoidSchema
```

### Parameters
- **endpoint** (str) - Required - The SPARQL endpoint URL.
- **name** (str) - Required - The name of the source.
- **output_dir** (str | Path | None) - Optional - Directory to export files; if None, no export.
- **tag** (str) - Optional - Tag for the export (default 'discovered_remote').
- **fmt** (str) - Optional - Format for export (default 'all').
- **timeout** (float) - Optional - Timeout in seconds (default 30.0).
- **max_retries** (int) - Optional - Maximum number of retries (default 1).
- **graph_uris** (str | list[str] | None) - Optional - Graph URIs to filter extraction.
- **batch_size** (int) - Optional - Batch size (default 100).
- **graph_batch_size** (int) - Optional - Graph batch size (default 8).
- **max_pages** (int) - Optional - Maximum number of pages (default 1000).
- **get_graphs_from_store** (bool) - Optional - Whether to get graphs from a graph store (default False).
- **graph_store_url** (str | None) - Optional - URL of the graph store.
- **graph_store_dir** (str | Path) - Optional - Directory for the graph store (default 'graph-store').
- **graph_store_max_bytes** (int) - Optional - Maximum bytes for the graph store (default 67108864).

### Returns
- **VoidSchema** - The published VoID as an object.
```

--------------------------------

### Parse SHACL Turtle into MinedSchema

Source: https://rdfsolve.readthedocs.io/en/latest/schema_models.html

Reads a SHACL Turtle file and reconstructs a MinedSchema instance. The example prints the number of patterns, which is 10 for the given file.

```python
>>> shacl_ttl = Path("shapes.ttl").read_text()
>>> schema = MinedSchema.from_shacl(shacl_ttl)
>>> print(len(schema.patterns))
10
```

--------------------------------

### Parse VoID Turtle into MinedSchema

Source: https://rdfsolve.readthedocs.io/en/latest/schema_models.html

Reads a VoID Turtle file and reconstructs a MinedSchema instance. The example prints the number of patterns, which is 435 for the given file.

```python
>>> void_ttl = Path("dataset_void.ttl").read_text()
>>> schema = MinedSchema.from_void(void_ttl)
>>> print(len(schema.patterns))
435
```

--------------------------------

### SparqlHelper.select()

Source: https://rdfsolve.readthedocs.io/en/latest/sparql_helper.html

Executes a SPARQL SELECT query and returns the results as a list of bindings.

```APIDOC
## SparqlHelper.select()

### Description
Executes a SPARQL SELECT query and returns the results as a list of bindings.

### Method
select

### Signature
select(query: str, **kwargs) -> list

### Parameters
- **query** (str) - Required - The SPARQL SELECT query string.
- **kwargs** - Optional - Additional arguments passed to run_query.

### Returns
- A list of binding dictionaries, where each dictionary maps variable names to values.
```

--------------------------------

### load_parser_from_file(file_path: str | Path, fmt: str | None = None) -> Parser

Source: https://rdfsolve.readthedocs.io/en/latest/api.html

Load a parser from a file.

```APIDOC
## load_parser_from_file

### Description
Load a parser from a file.

### Signature
```python
load_parser_from_file(file_path: str | Path, fmt: str | None = None) -> Parser
```

### Parameters
- **file_path** (str | Path) - Required - Path to the file.
- **fmt** (str | None) - Optional - Format of the file.

### Returns
- **Parser** - The loaded parser.
```

--------------------------------

### rdfsolve.sources module source code

Source: https://rdfsolve.readthedocs.io/en/latest/_modules/rdfsolve/sources.html

Source code for the rdfsolve.sources module, defining the SourceEntry TypedDict, default file paths, and Bioregistry enrichment helpers. The snippet is truncated at the start of _resolve_bioregistry_prefix.

```python
"""Load and enrich SPARQL data source definitions from YAML with Bioregistry metadata."""

from __future__ import annotations

import csv
import json
import logging
from pathlib import Path
from typing import Any, TypedDict, cast

import pandas as pd
import yaml

logger = logging.getLogger(__name__)




[docs]
class SourceEntry(TypedDict, total=False):
    """Typed dictionary for a single data-source definition."""

    name: str
    endpoint: str
    dataset_metadata: dict[str, Any] | None
    metadata_graph_uris: list[str] | None
    enrichment: dict[str, Any]
    endpoint_status: str
    last_checked: str
    last_success: str
    last_error: str
    failure_count: int
    avg_response_time: float | None
    has_void: bool
    has_void_partitions: bool
    has_void_patterns: bool
    void_default_graph: bool
    void_iri: str  # DEPRECATED: use void_graphs instead
    void_uri_base: str
    void_graphs: list[str]  # Discovered VoID metadata graph URIs
    void_schema: list[str]  # VoID graphs with mineable partitions
    graph_uris: list[str]
    use_graph: bool
    chunk_size: int
    class_batch_size: int
    class_chunk_size: int | None
    timeout: float
    delay: float
    counts: bool
    unsafe_paging: bool
    notes: str
    # Endpoint metadata (populated by probe/discovery scripts)
    sparql_engine: str
    sparql_strategy: str
    supports_graph: bool
    endpoint_down: bool
    # Bioregistry-derived metadata (populated by enrich_source_with_bioregistry)
    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]]




# default path

_REPO_ROOT = Path(__file__).resolve().parent.parent.parent
DEFAULT_SOURCES_YAML = _REPO_ROOT / "data" / "sources.yaml"
DEFAULT_SOURCES_JSONLD = _REPO_ROOT / "data" / "sources.jsonld"
DEFAULT_SOURCES_CSV = _REPO_ROOT / "data" / "sources.csv"


def _default_sources_path() -> Path:
    """Return the default sources file, preferring YAML.

    Raises
    ------
    FileNotFoundError
        If neither ``data/sources.yaml`` nor ``data/sources.jsonld`` exists
        relative to the repository root.
    """
    if DEFAULT_SOURCES_YAML.exists():
        return DEFAULT_SOURCES_YAML
    if DEFAULT_SOURCES_JSONLD.exists():
        return DEFAULT_SOURCES_JSONLD
    raise FileNotFoundError(
        f"Default sources file not found. Looked for:\n"
        f"  {DEFAULT_SOURCES_YAML}\n"
        f"  {DEFAULT_SOURCES_JSONLD}\n"
        "Pass an explicit 'path' argument to load_sources()."
    )


# Bioregistry enrichment


# Lazily populated reverse index: extra-provider code -> list of BR prefixes
# whose resource lists that code in get_extra_providers().
_EXTRA_PROVIDER_INDEX: dict[str, list[str]] | None = None


def _build_extra_provider_index() -> dict[str, list[str]]:
    """Build a mapping from provider code -> [bioregistry prefix, ...].

    This allows resolving names like ``"bio2rdf.uniprot"`` to ``"uniprot"``
    by looking up which resource has ``bio2rdf`` as an extra provider.
    The index is computed once and cached in :data:`_EXTRA_PROVIDER_INDEX`.
    """
    try:
        import bioregistry

        index: dict[str, list[str]] = {}
        for prefix, resource in bioregistry.manager.registry.items():
            for ep in resource.get_extra_providers():
                index.setdefault(ep.code, []).append(prefix)
        return index
    except Exception:
        logger.debug("bioregistry extra-provider index unavailable", exc_info=True)
        return {}


def _get_extra_provider_index() -> dict[str, list[str]]:
    global _EXTRA_PROVIDER_INDEX
    if _EXTRA_PROVIDER_INDEX is None:
        _EXTRA_PROVIDER_INDEX = _build_extra_provider_index()
    return _EXTRA_PROVIDER_INDEX


def _resolve_bioregistry_prefix(entry: SourceEntry) -> str | None:
    """Resolve the canonical Bioregistry prefix for a source entry.

    Resolution is attempted in the following order, stopping at the first hit:

```
Tokens:
Raw