SPARQL Helper

SPARQL Helper, Centralized SPARQL query execution with automatic fallback.

This module is a SPARQL client that handles: - Automatic GET -> POST fallback for endpoints that require POST - Exponential backoff retry logic for transient failures - Support for SELECT (JSON) and CONSTRUCT (Turtle/N3) queries - HTML error detection in responses - Consistent logging across all SPARQL operations - Support for pagination (limit and offset usage)

Usage:

from rdfsolve.sparql_helper import SparqlHelper

# Create a helper for an endpoint helper = SparqlHelper(”https://sparql.example.org/”)

# Execute SELECT query (returns dict) results = helper.select(“SELECT ?s ?p ?o WHERE { ?s ?p ?o } LIMIT 10”)

# Execute CONSTRUCT query (returns bytes/string) turtle_data = helper.construct(“CONSTRUCT { ?s ?p ?o } WHERE { ?s ?p ?o }”)

# Execute ASK query (returns bool) exists = helper.ask(“ASK { ?s a <http://example.org/Class> }”)

class QueryRecord(query: str, query_type: ~typing.Literal['SELECT', 'CONSTRUCT', 'ASK'], endpoint_url: str, timestamp: str = <factory>, description: str = '', keywords: list[str] = <factory>, success: bool = True)[source]

Bases: object

Record of a SPARQL query execution.

query: str = <dataclasses._MISSING_TYPE object>

query_type: Literal['SELECT', 'CONSTRUCT', 'ASK'] = <dataclasses._MISSING_TYPE object>

endpoint_url: str = <dataclasses._MISSING_TYPE object>

timestamp: str = <dataclasses._MISSING_TYPE object>

description: str = ''

keywords: list[str] = <dataclasses._MISSING_TYPE object>

success: bool = True

query_id() → str[source]: Generate a unique ID for this query based on content hash.

exception SparqlHelperError[source]

Bases: Exception

Base exception for SPARQL helper errors.

exception EndpointError[source]

Bases: SparqlHelperError

Raised when the endpoint returns an error.

exception EndpointTimeoutError[source]

Bases: EndpointError

Raised when the endpoint times out (read / connect).

exception EndpointUnhealthyError[source]

Bases: EndpointError

Raised when the endpoint returns a 200/400 with a non-SPARQL body.

Typical examples: database in recovery mode, backend proxy errors, maintenance pages returned as text/plain or text/html.

exception PaginationTruncatedError(msg: str, offset: int = 0)[source]

Bases: EndpointTimeoutError

Raised by select_chunked when pagination is abandoned mid-stream.

This means some rows were already yielded before the error, so the caller received a partial result set. The offset attribute records where pagination stopped.

Initialize a pagination truncation error.

Parameters:

msg – Error message.
offset – Offset at which pagination stopped.

exception QueryError[source]

Bases: SparqlHelperError

Raised when the query itself is invalid.

class MimeTypes[source]

Bases: object

Standard MIME types for SPARQL protocol.

JSON = 'application/sparql-results+json'

XML = 'application/sparql-results+xml'

TURTLE = 'text/turtle'

N3 = 'text/n3'

NTRIPLES = 'application/n-triples'

RDFXML = 'application/rdf+xml'

JSONLD = 'application/ld+json'

SELECT_ACCEPT = 'application/sparql-results+json, application/sparql-results+xml;q=0.9'

CONSTRUCT_ACCEPT = 'text/turtle, text/n3;q=0.9, application/n-triples;q=0.8, application/rdf+xml;q=0.7'

class SparqlHelper(endpoint_url: str, *, use_post: bool = False, max_retries: int = 10, initial_backoff: float = 1.0, max_backoff: float = 30.0, timeout: float = 10000.0, sparql_engine: str = '', sparql_strategy: str = '', source_name: str = '', inter_request_delay: float = 0.0)[source]

Bases: object

Centralized SPARQL query executor with automatic fallback and retry logic.

This class provides: - Automatic GET/POST method fallback when endpoints return HTML/500 errors - Configurable retry with exponential backoff for transient failures - Consistent error handling and logging - Support for SELECT, CONSTRUCT, and ASK queries

Uses standard requests library.

endpoint_url: The SPARQL endpoint URL

use_post: If True, always use POST method (skip GET attempt)

max_retries: Maximum number of retry attempts

initial_backoff: Initial backoff delay in seconds

max_backoff: Maximum backoff delay in seconds

timeout: Request timeout in seconds

Example

>>> 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"])

Initialize the SPARQL helper.

Parameters:

endpoint_url – SPARQL endpoint URL
use_post – Always use POST (default: False, tries GET first)
max_retries – Maximum retry attempts for transient failures
initial_backoff – Initial delay between retries (seconds)
max_backoff – Maximum delay between retries (seconds)
timeout – Request timeout in seconds (default: 60)
sparql_engine – Engine hint from endpoint probes (e.g. "virtuoso", "blazegraph", "qlever"). Used for engine-specific workarounds.
sparql_strategy – Preferred query strategy from endpoint probes (e.g. "get+json", "post+form+json"). When set, the helper starts with this strategy and falls back to the default GET→POST chain only on failure.
source_name – Source name from sources.yaml. When set, winning strategy discoveries are recorded in _strategy_updates for later flush via flush_strategy_updates().
inter_request_delay – Seconds to sleep before each new logical SPARQL request (not between retry attempts). Use for polite throttling of remote public endpoints. Zero (the default) disables the delay — suitable for local QLever instances where there is no need to be polite.

POST_RETRY_PATTERNS = ('html', '500', 'internal', 'error', 'method not allowed')

HTML_MARKERS = ('<!DOCTYPE', '<html', '<HTML', '<!doctype')

RETRY_STATUS_CODES = (500, 502, 503, 504, 429)

COST_LIMIT_PATTERNS: ClassVar[tuple[str, ...]] = ('estimated execution time', 'exceeds the limit', 'query timed out', 'timeout expired', 'execution time limit', 'statement timeout', 'cost limit exceeded', 'waited for a result from another thread which then failed', 'memory limit exceeded')

classmethod get_strategy_updates() → dict[str, str][source]: Return accumulated strategy updates (source_name → winning strategy).

classmethod flush_strategy_updates(sources_yaml: str | Path | None = None) → int[source]

Write accumulated strategy discoveries back to sources.yaml.

For each source whose winning strategy differs from (or was absent in) the YAML, update sparql_strategy in-place.

Parameters:: sources_yaml – Path to sources.yaml. Defaults to <repo>/data/sources.yaml.
Returns:: Number of entries updated.
Return type:: int

classmethod enable_query_collection() → None[source]: Enable collection of all executed queries.

classmethod disable_query_collection() → None[source]: Disable query collection.

classmethod get_collected_queries() → list[QueryRecord][source]: Get all collected queries.

classmethod clear_collected_queries() → None[source]: Clear all collected queries.

classmethod export_queries_as_ttl(output_file: str | None = None, base_uri: str = 'https://example.org/sparql-queries/', dataset_name: str = 'dataset') → str[source]

Export collected queries as TTL using SHACL SPARQL representation.

Parameters:

output_file – Optional file path to write TTL
base_uri – Base URI for query IRIs
dataset_name – Name of the dataset for namespacing

Returns:

TTL string with all collected queries

classmethod from_source_entry(entry: dict[str, Any], *, timeout: float | None = None, max_retries: int = 10) → SparqlHelper[source]

Create a SparqlHelper from a sources.yaml entry dict.

Reads endpoint, sparql_engine, sparql_strategy, and timeout from entry and passes them through.

Parameters:

entry – A single source dict (or SourceEntry / SourceModel).
timeout – Override timeout (else uses entry’s or default).
max_retries – Override max retries.

Returns:

Configured SparqlHelper instance.

select(query: str, purpose: str = '') → dict[str, Any][source]

Execute a SELECT query and return JSON results.

Parameters:

query – SPARQL SELECT query string.
purpose – Caller context for logs, e.g. "mining/typed-object".

Returns:

Dictionary with SPARQL JSON results format containing "head" and "results" keys.

Raises:

EndpointError – If the endpoint returns an error after all retries.
QueryError – If the query is malformed.

construct(query: str) → str[source]

Execute a CONSTRUCT query and return Turtle RDF data.

Parameters:

query – SPARQL CONSTRUCT query string

Returns:

Turtle-formatted RDF string

Raises:

EndpointError – If the endpoint returns an error after all retries
QueryError – If the query is malformed

construct_graph(query: str) → Graph[source]

Execute a CONSTRUCT query and return an RDFLib Graph.

The CONSTRUCT method internally uses _execute which handles GET->POST fallback automatically when HTML is detected in the response string.

Parameters:

query – SPARQL CONSTRUCT query string

Returns:

RDFLib Graph containing the constructed triples

Raises:

EndpointError – If the endpoint returns an error after all retries
QueryError – If the query is malformed

ask(query: str) → bool[source]

Execute an ASK query and return boolean result.

Parameters:

query – SPARQL ASK query string

Returns:

True if the pattern exists, False otherwise

Raises:

EndpointError – If the endpoint returns an error after all retries
QueryError – If the query is malformed

find_classes_for_uri_pattern(uri_prefix: str) → list[str][source]

Find all rdf:type classes whose instances match uri_prefix.

Uses an IRI-range filter (index-friendly on QLever and most SPARQL engines):

SELECT DISTINCT ?c
WHERE {
  ?s a ?c .
  FILTER(
    ?s >= <uri_prefix> &&
    ?s <  <uri_prefix_next>
  )
}

The upper-bound uri_prefix_next is derived by incrementing the last character of uri_prefix to the next IRI-safe codepoint (e.g. "…/obo/AAO_" → "…/obo/AAOa" because _ + 1 = backtick which is IRI-unsafe, so we skip to a).

Falls back to STRSTARTS only if no safe upper-bound character can be found within 16 codepoints (virtually never happens). STRSTARTS forces a full table scan and can take minutes on large datasets.

Parameters:: uri_prefix – URI prefix string, e.g. "https://identifiers.org/ensembl/".
Returns:: Deduplicated list of class URIs (may be empty).

find_classes_for_iris_by_graph(iris: list[str], values_batch_size: int = 50) → dict[str, dict[str, list[str]]][source]

Find rdf:type classes for specific IRIs, grouped by named graph.

Splits iris into chunks of values_batch_size to avoid exceeding HTTP header / request-line limits (Beast default is ~8 KB). Results from all chunks are merged before returning.

Uses a VALUES-based query to resolve multiple IRIs at once:

SELECT DISTINCT ?s ?g ?c
WHERE {
  VALUES ?s { <iri1> <iri2> ... }
  GRAPH ?g { ?s a ?c }
}

Parameters:

iris – Full entity IRIs to look up.
values_batch_size – Max IRIs per VALUES block (default 50).

Returns:

Nested dict {entity_iri: {graph_uri: [class_uri, ...]}}. Only IRIs found in at least one graph are included.

find_classes_for_iris(iris: list[str], values_batch_size: int = 50) → dict[str, list[str]][source]

Find rdf:type classes for specific IRIs (no named-graph grouping).

Splits iris into chunks of values_batch_size to avoid exceeding HTTP header / request-line limits.

Parameters:

iris – Full entity IRIs to look up.
values_batch_size – Max IRIs per VALUES block (default 50).

Returns:

Dict {entity_iri: [class_uri, ...]}. Only IRIs found in the endpoint are included.

find_all_classes() → dict[str, list[str]][source]

Return all rdf:type classes used in this endpoint, with their instances.

Runs a direct SELECT DISTINCT ?s ?c WHERE { ?s a ?c } — no entity-IRI filter. Used as a fallback when no entity IRIs are available to look up (e.g. instance-mapping files that store only bare CURIEs that could not be resolved).

Returns:: Dict {entity_iri: [class_uri, ...]}.

get_bindings(query: str, purpose: str = '') → list[dict[str, str]][source]

Execute SELECT query and return simplified bindings list.

Convenience method that extracts just the variable values.

Parameters:

query – SPARQL SELECT query string
purpose – Optional tag for log identification

Returns:

List of dicts mapping variable names to their values

Example

>>> bindings = helper.get_bindings("SELECT ?s ?p { ?s ?p ?o }")
>>> for row in bindings:
...     print(row["s"], row["p"])

select_chunked(query_template: str, chunk_size: int = 100, max_total_results: int | None = None, delay_between_chunks: float = 0.5, purpose: str = '') → Any[source]

Execute a SELECT query in chunks using OFFSET/LIMIT pagination.

Uses adaptive pagination: when the endpoint times out, the chunk (LIMIT) is reduced by ~15 % and the same offset is retried after a cooldown pause. The chunk size will never shrink below 60 % of the original value (i.e. a maximum cumulative reduction of ~40 %). Up to 3 consecutive shrinks are attempted per offset before giving up on that page.

After a successful fetch with a reduced chunk size, the smaller size is kept for subsequent pages (the endpoint is consistently slow).

Parameters:

query_template – SPARQL query with {offset} and {limit} placeholders.
chunk_size – Initial number of results per chunk.
max_total_results – Cap on total results (None = all).
delay_between_chunks – Polite pause between pages (seconds).
purpose – Caller context for log messages.

Yields:

List of bindings (dicts) from each chunk.

static prepare_paginated_query(base_query: str) → str[source]

Prepare a SPARQL query for use with select_chunked by escaping braces.

SPARQL queries contain curly braces {} which conflict with Python’s str.format() used for pagination placeholders. This method: 1. Escapes all existing braces ({{ and }}) 2. Appends OFFSET {offset} and LIMIT {limit} placeholders

Parameters:: base_query – SPARQL query WITHOUT OFFSET/LIMIT clauses. Should be a complete query ready to execute.
Returns:: Query template safe for use with str.format(offset=N, limit=M)

Example

>>> query = "SELECT ?s WHERE { ?s a ?class }"
>>> template = SparqlHelper.prepare_paginated_query(query)
>>> # template is now safe for: template.format(offset=0, limit=100)
>>> for bindings in helper.select_chunked(template):
...     process(bindings)

static escape_sparql_for_format(query: str) → str[source]

Escape SPARQL braces so the query can be used with str.format().

This is useful when you need to add your own placeholders to a query that contains SPARQL curly braces.

Parameters:: query – SPARQL query with literal curly braces
Returns:: Query with braces doubled for .format() compatibility

Example

>>> q = "SELECT ?s WHERE { ?s a <{class_uri}> }"  # Won't work!
>>> # Instead:
>>> q = SparqlHelper.escape_sparql_for_format(
...     "SELECT ?s WHERE { ?s a <CLASS_PLACEHOLDER> }"
... )
>>> q = q.replace("CLASS_PLACEHOLDER", "{class_uri}")

close() → None[source]: Close the underlying requests session.

sparql_select(endpoint_url: str, query: str, use_post: bool = False, purpose: str = '') → dict[str, Any][source]

Execute a one-off SELECT query.

Convenience function when you don’t need to reuse the helper.

Parameters:

endpoint_url – SPARQL endpoint URL
query – SPARQL SELECT query
use_post – Force POST method
purpose – Optional tag for log identification

Returns:

SPARQL JSON results

sparql_construct(endpoint_url: str, query: str, use_post: bool = False) → Graph[source]

Execute a one-off CONSTRUCT query.

Convenience function when you don’t need to reuse the helper.

Parameters:

endpoint_url – SPARQL endpoint URL
query – SPARQL CONSTRUCT query
use_post – Force POST method

Returns:

RDFLib Graph with constructed triples