orchestrata_ultimate.domaingen.stardog

Reference › orchestrata_ultimate › domaingen › stardog

stardog is a minimal HTTP connection to a Stardog database (SPEC-105 FR-006, REQ-105-042). The repositories emitted by generator need only an object with update and query methods, and StardogConnection is a thin real adapter for that. It is built from a base URL, a database name, optional credentials and an optional injectable HTTP client. update_url and query_url form the endpoints <base>/<db>/update and <base>/<db>/query. update POSTs a SPARQL-star update and query POSTs a query and returns the JSON result bindings. A failed request raises StardogConnectionError (ERR-105-005). The docstring says unit tests check only URL and authentication construction, without network access or mocks, while integration tests run on demand against a live server. The package imports it.

At a glance

Class diagram

Dependency diagram

API

Classes

StardogConnectionError

Bases: BaseError · Decorators: traces(spec='105', story='US-105-001', reqs=['REQ-105-042']) · Line 22

Raised when a Stardog HTTP request fails (ERR-105-005).

StardogConnection

Decorators: traces(spec='105', story='US-105-001', reqs=['REQ-105-042']) · Line 27

SPARQL-over-HTTP connection to a Stardog database (insert/query).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/domaingen/stardog.py

"""Minimal Stardog HTTP SPARQL connection (SPEC-105 FR-006 / REQ-105-042).

A thin, real adapter that the generated repositories accept (they only require an
object with ``update``/``query``). Stardog speaks SPARQL 1.1 + RDF-star over HTTP:
updates POST to ``<base>/<db>/update``, queries to ``<base>/<db>/query``. The
on-demand integration tests exercise a live server; unit tests assert URL/auth
construction only (no network, no mocks).
"""

from __future__ import annotations

from typing import Any

from orchestrata_ultimate.core.base.error import BaseError
from orchestrata_ultimate.core.logging import get_logger
from orchestrata_ultimate.core.traceability import traces

logger = get_logger(__name__)


@traces(spec="105", story="US-105-001", reqs=["REQ-105-042"])
class StardogConnectionError(BaseError):
    """Raised when a Stardog HTTP request fails (ERR-105-005)."""


@traces(spec="105", story="US-105-001", reqs=["REQ-105-042"])
class StardogConnection:
    """SPARQL-over-HTTP connection to a Stardog database (insert/query)."""

    def __init__(
        self,
        base_url: str,
        database: str,
        username: str | None = None,
        password: str | None = None,
        http: Any | None = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.database = database
        self._auth = (username, password) if username is not None else None
        self._http = http  # injected httpx.Client-like; created lazily otherwise

    @property
    def update_url(self) -> str:
        return f"{self.base_url}/{self.database}/update"

    @property
    def query_url(self) -> str:
        return f"{self.base_url}/{self.database}/query"

    @traces(spec="105", story="US-105-001", reqs=["REQ-105-042"])
    def _client(self) -> Any:
        if self._http is None:
            import httpx  # noqa: PLC0415 — optional/lazy; only needed for live calls

            self._http = httpx.Client(timeout=30.0)
        return self._http

    @traces(spec="105", story="US-105-001", reqs=["REQ-105-042"])
    def update(self, sparql: str) -> None:
        """POST a SPARQL-star update (insert-only) to Stardog."""
        try:
            resp = self._client().post(
                self.update_url, data={"update": sparql}, auth=self._auth
            )
            resp.raise_for_status()
        except Exception as exc:
            raise StardogConnectionError(f"Stardog update failed: {exc}") from exc

    @traces(spec="105", story="US-105-001", reqs=["REQ-105-042"])
    def query(self, sparql: str) -> dict[str, Any]:
        """POST a SPARQL-star query; return the JSON results bindings."""
        try:
            resp = self._client().post(
                self.query_url,
                data={"query": sparql},
                auth=self._auth,
                headers={"Accept": "application/sparql-results+json"},
            )
            resp.raise_for_status()
            data: dict[str, Any] = resp.json()
            return data
        except Exception as exc:
            raise StardogConnectionError(f"Stardog query failed: {exc}") from exc