Query

SPARQL query execution - pure-library module (no Flask dependency).

This module provides structured SPARQL query execution built on top of SparqlHelper. It adds:

• Pydantic result models (ResultCell, QueryResult) that give strongly-typed access to SPARQL JSON result bindings.

• A single execute_sparql() convenience function with a clean dict-return signature suitable for the public API.

All HTTP / retry / GET->POST logic is delegated to SparqlHelper.

class ResultCell(*, value: str, type: str, lang: str | None = None, datatype: str | None = None)

Bases: BaseModel

One cell in a SPARQL result row.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

value: str

type: str

lang: str | None

datatype: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class QueryResult(*, query: str, endpoint: str, variables: list[str], rows: list[dict[str, ~rdfsolve.query.ResultCell]], variable_map: dict[str, str] = <factory>, row_count: int, duration_ms: int, error: str | None = None)

Bases: BaseModel

Structured result from a SPARQL query execution.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

query: str

endpoint: str

variables: list[str]

rows: list[dict[str, ResultCell]]

variable_map: dict[str, str]

row_count: int

duration_ms: int

error: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

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 QueryResult.

Parameters:

• query – Full SPARQL query string.

• endpoint – URL of the SPARQL endpoint.

• method – HTTP method ("GET" or "POST"). If "GET" fails the underlying SparqlHelper will automatically retry with POST.

• timeout – Request timeout in seconds.

• variable_map – Optional mapping of SPARQL ?variable names to schema URIs.

Returns:

Pydantic model with query, endpoint, variables, rows, variable_map, row_count, duration_ms, and optionally error.

Return type:

QueryResult