Simple helpers for talking to SPARQL endpoints and working with query results.
This package underpins SPARQL-based integrations used by other parts of ekglib.
SPARQLEndpoint - Interface for executing SPARQL queries against an endpointSPARQLResponse - Wrapper for SPARQL query responsesSPARQLEndpoint.execute_construct(query: str) -> SPARQLResponse | None - Execute a CONSTRUCT querySPARQLEndpoint.execute_select(query: str) -> SPARQLResponse | None - Execute a SELECT querySPARQLEndpoint.endpoint_url() -> str - Get the endpoint URLSPARQLEndpoint.user_id() -> str | None - Get authentication user IDSPARQLEndpoint.password() -> str | None - Get authentication passwordSPARQLEndpoint.handle_error(response) -> bool - Handle HTTP errors from endpointiter_raw(r: requests.Response, chunk_size: int = 1) -> Any - Iterate over raw response chunks for streamingdump(obj: Any) -> None - Debug function to dump object attributesset_cli_params(parser: argparse.ArgumentParser) -> None - Add SPARQL-related CLI argumentsThe CLI parameters include:
--sparql_endpoint-s3_endpoint - The SPARQL endpoint URL--sparql_endpoint-database - The SPARQL database name--sparql_endpoint-userid - Authentication user ID--sparql_endpoint-passwd / --sparql_endpoint-password - Authentication passwordfrom ekglib.sparql import SPARQLEndpoint
# Create endpoint
endpoint = SPARQLEndpoint(
endpoint_url="https://sparql.example.com/query",
user_id="user",
password="pass"
)
# Execute CONSTRUCT query
query = """
PREFIX ex: <http://example.org/>
CONSTRUCT { ?s ?p ?o }
WHERE { ?s ?p ?o }
"""
response = endpoint.execute_construct(query)
if response:
graph = response.convert() # Convert to rdflib.Graph