sage.http_server package¶

Submodules¶

sage.http_server.responses module¶

sage.http_server.responses.analyze_term(value: str) → Tuple[str, str, Optional[str], Optional[str]]¶

Analyze a RDF term and extract various information about it.

Argument: The RDF term to analyze.

Returns: A tuple (value, type, extra_label, extra_value) where:

value is the term value.
type is the type of the term (literal or uri).
extra_label is the type of an extra element for this term (datatype or xml:lang).
extra_value is the value of an extra element for this term.

Example:

>>> analyze_term("<http://example.org#Anna>")
("<http://example.org#Anna>", "uri", None, None)
>>> analyze_term('"Anna"')
('"Anna"', "literal", None, None)
>>> analyze_term('"Anna"@en')
('"Anna"', "literal", "xml:lang", "en")
>>> analyze_term('"Anna"^^<http://datatype.org#string>')
('"Anna"', "literal", "datatype", "<http://datatype.org#string>")

sage.http_server.responses.binding_to_json(binding: Dict[str, str]) → dict¶

Format a set of solutions bindings in the W3C SPARQL JSON format.

Argument: A set of solution bindings.

Returns: The input set of solution bindings, encoded in the W3C SPARQL JSON format.

sage.http_server.responses.bindings_to_w3c_xml(bindings: Iterable[Dict[str, str]], skol_url: str) → xml.etree.ElementTree.Element¶

Formats a set of bindings into SPARQL results in the W3C SPARQL XML format.

Args:

bindings: An iterable which yields set of solution bindings.
skol_url: URL used for the skolemization of blank nodes.

Returns: The input set of solution bindings, encoded in the W3C SPARQL XML format.

sage.http_server.responses.ntriples_streaming(triples: Iterable[Tuple[str, str, str]]) → Iterable[str]¶

Serialize RDF triples in N-Triples string format in a iterable-fashion.

Argument: An iterable which yields RDF triples to process.

Yields: RDF triples in a string format, encoded in the N-Triples format.

sage.http_server.responses.raw_json_streaming(bindings: Iterable[Dict[str, str]], next_link: Optional[str], stats: dict, skol_url: str) → Iterable[str]¶

Yield a page of SaGe results in a non-standard JSON format, so it can be sent in an HTTP response.

Args:

bindings: An iterable which yields set of solution bindings.
next_link: Link to a SaGe saved plan. Use None if there is no one, i.e., the query execution has completed during the quantum.
stats: Statistics about query execution.
skol_url: URL used for the skolemization of blank nodes.

Yields:

A page of SaGe results in the W3C SPARQL JSON results format.

sage.http_server.responses.skolemize(bindings: Iterable[Dict[str, str]], url: str) → Iterable[Dict[str, str]]¶

Skolemize blank nodes in a list of solution bindings.

Args:

bindings: An iterable which yields set of solution bindings to process.
url: Prefix URL used for skolemization.

Yields:

Solution bindings, where blank nodes have been skolemized using the input URL.

sage.http_server.responses.skolemize_one(bnode: str, url: str) → str¶

Skolemize a blank node.

If the input value is not a Blank node, then do nothing.

Args:

value: RDF Blank node to skolemize.
url: Prefix URL used for skolemization.

Returns:

The skolemized blank node, or the value itself if it was not a blank node.

sage.http_server.responses.stream_json_list(iterator: Iterable[Dict[str, str]]) → Iterable[str]¶

A generator for streaming a list of JSON results in an HTTP response.

Argument: An iterator which yields solutions bindings.

Yields: Solution bindings as string-encoded JSON.

sage.http_server.responses.w3c_json_streaming(bindings: Iterable[Dict[str, str]], next_link: Optional[str], stats: dict, skol_url: str) → Iterable[str]¶

Yield a page of SaGe results in the W3C SPARQL JSON results format, so it can be sent in an HTTP response.

Args:

bindings: An iterable which yields set of solution bindings.
next_link: Link to a SaGe saved plan. Use None if there is no one, i.e., the query execution has completed during the quantum.
stats: Statistics about query execution.
skol_url: URL used for the skolemization of blank nodes.

Yields:

A page of SaGe results in the W3C SPARQL JSON results format.

sage.http_server.responses.w3c_xml(bindings: Iterable[Dict[str, str]], next_link: Optional[str], stats: dict, skol_url: str) → Iterable[str]¶

Yield a page of SaGe results in the W3C SPARQL XML results format, so it can be sent in an HTTP response.

Args:

bindings: An iterable which yields set of solution bindings.
next_link: Link to a SaGe saved plan. Use None if there is no one, i.e., the query execution has completed during the quantum.
stats: Statistics about query execution.
skol_url: URL used for the skolemization of blank nodes.

Yields:

A page of SaGe results in the W3C SPARQL JSON results format.

sage.http_server.server module¶

class sage.http_server.server.SagePostQuery¶

Bases: pydantic.main.BaseModel

Data model for the body of POST SPARQL queries

defaultGraph: str = None¶

next: str = None¶

query: str = None¶

sage.http_server.server.choose_void_format(mimetypes)¶

sage.http_server.server.create_response(mimetypes: List[str], bindings: List[Dict[str, str]], next_page: Optional[str], stats: dict, skol_url: str) → starlette.responses.Response¶

Create an HTTP response for the results of SPARQL query execution.

Args:

mimetypes: mimetypes from the input HTTP request.
bindings: list of query results.
next_link: Link to a SaGe saved plan. Use None if there is no one, i.e., the query execution has completed during the quantum.
stats: Statistics about query execution.
skol_url: URL used for the skolemization of blank nodes.

Returns:

An HTTP response built from the input mimetypes and the SPARQL query results.

async sage.http_server.server.execute_query(query: str, default_graph_uri: str, next_link: Optional[str], dataset: sage.database.core.dataset.Dataset) → Tuple[List[Dict[str, str]], Optional[str], Dict[str, str]]¶

Execute a query using the SageEngine and returns the appropriate HTTP response.

Any failure will results in a rollback/abort on the current query execution.

Args:

query: SPARQL query to execute.
default_graph_uri: URI of the default RDF graph to use.
next_link: URI to a saved plan. Can be None if query execution should starts from the beginning.
dataset: RDF dataset on which the query is executed.

Returns:

A tuple (bindings, next_page, stats) where: * bindings is a list of query results. * next_page is a link to saved query execution state. Sets to None if query execution completed during the time quantum. * stats are statistics about query execution.

Throws: Any exception that have occured during query execution.

sage.http_server.server.run_app(config_file: str) → fastapi.applications.FastAPI¶

Create the HTTP server, compatible with uvicorn/gunicorn.

Argument: SaGe configuration file, in YAML format.

Returns: The FastAPI HTTP application.

sage.http_server.utils module¶

sage.http_server.utils.decode_saved_plan(input: str) → iterators_pb2.RootTree¶

Decode a Protobuf-based saved plan from a string format.

Argument: A saved plan, encoded as a string of bytes.

Returns: The saved plan, encoded as a Protobuf message.

sage.http_server.utils.encode_saved_plan(savedPlan: iterators_pb2.RootTree) → str¶

Encode a Protobuf-based saved plan into string format.

Argument: A saved plan, encoded as a Protobuf message.

Returns: The saved plan, encoded as a string of bytes.

sage.http_server.utils.secure_url(url: str) → str¶

Secure potentially ill formatted urls.

Argument: URL to secure.

Returns: Secured URL.

sage.http_server package¶

Submodules¶

sage.http_server.responses module¶

sage.http_server.server module¶

sage.http_server.utils module¶

Module contents¶