ontocast.tool.triple_manager.fuseki

Fuseki triple store management for OntoCast.

This module provides a concrete implementation of triple store management using Apache Fuseki as the backend. It supports named graphs for ontologies and facts, with proper authentication and dataset management.

FusekiTripleStoreManager

Bases: TripleStoreManagerWithAuth

Fuseki-based triple store manager.

This class provides a concrete implementation of triple store management using Apache Fuseki. It stores ontologies as named graphs using their URIs as graph names, and supports dataset creation and cleanup.

URI shape: uri must be the Fuseki HTTP server root (e.g. http://localhost:3032), not a dataset path or UI URL. Dataset names are dataset / ontologies_dataset; the client calls {uri}/{dataset_name}/sparql and similar. The UI route /#/dataset/dataset_name is only for the browser; paste the origin (and optional non-dataset path prefix) into FUSEKI_URI, and set FUSEKI_DATASET to dataset_name.

The manager uses Fuseki's REST API for all operations, including: - Dataset creation and management - Named graph operations for ontologies - SPARQL queries for ontology discovery - Graph-level data operations

Attributes:

Name	Type	Description
dataset	str | None	Facts dataset name (first path segment in Fuseki HTTP API).
ontologies_dataset	str	Ontologies dataset name.
shapes_dataset	str	SHACL shapes dataset name. Separate from the ontologies dataset because catalog discovery claims every named graph holding an owl:Ontology subject, and shapes documents declare one.

Source code in ontocast/tool/triple_manager/fuseki.py

class FusekiTripleStoreManager(TripleStoreManagerWithAuth):
    """Fuseki-based triple store manager.

    This class provides a concrete implementation of triple store management
    using Apache Fuseki. It stores ontologies as named graphs using their
    URIs as graph names, and supports dataset creation and cleanup.

    **URI shape:** ``uri`` must be the Fuseki **HTTP server root** (e.g.
    ``http://localhost:3032``), not a dataset path or UI URL. Dataset names are
    ``dataset`` / ``ontologies_dataset``; the client calls
    ``{uri}/{dataset_name}/sparql`` and similar. The UI route
    ``/#/dataset/dataset_name`` is only for the browser; paste the origin (and
    optional non-dataset path prefix) into ``FUSEKI_URI``, and set
    ``FUSEKI_DATASET`` to ``dataset_name``.

    The manager uses Fuseki's REST API for all operations, including:
    - Dataset creation and management
    - Named graph operations for ontologies
    - SPARQL queries for ontology discovery
    - Graph-level data operations

    Attributes:
        dataset: Facts dataset name (first path segment in Fuseki HTTP API).
        ontologies_dataset: Ontologies dataset name.
        shapes_dataset: SHACL shapes dataset name. Separate from the ontologies
            dataset because catalog discovery claims every named graph holding an
            ``owl:Ontology`` subject, and shapes documents declare one.
    """

    dataset: str | None = Field(default=None, description="Fuseki dataset name")
    ontologies_dataset: str = Field(
        default=DEFAULT_ONTOLOGIES_DATASET,
        description="Fuseki dataset name for ontologies",
    )
    shapes_dataset: str = Field(
        default=DEFAULT_SHAPES_DATASET,
        description="Fuseki dataset name for SHACL shapes",
    )

    def __init__(
        self,
        uri=None,
        auth=None,
        dataset=None,
        ontologies_dataset=None,
        shapes_dataset=None,
        **kwargs,
    ):
        """Initialize the Fuseki triple store manager.

        This method sets up the connection to Fuseki and creates the dataset
        if it doesn't exist. The dataset is NOT cleaned on initialization.

        Args:
            uri: Fuseki HTTP service root (e.g. ``http://localhost:3030``), not
                ``.../dataset/name`` and not a ``#/dataset/...`` UI link.
            auth: Authentication tuple (username, password) or string in "user/password" format.
            dataset: Facts dataset name (Fuseki API path segment).
            ontologies_dataset: Ontologies dataset name (separate Fuseki dataset).
            shapes_dataset: SHACL shapes dataset name (separate Fuseki dataset).
            **kwargs: Additional keyword arguments passed to the parent class.

        Example:
            >>> manager = FusekiTripleStoreManager(
            ...     uri="http://localhost:3030",
            ...     dataset="acme--demo--facts",
            ...     ontologies_dataset="acme--demo--ontologies",
            ... )
            >>> await manager.clean()
        """
        super().__init__(
            uri=uri, auth=auth, env_uri="FUSEKI_URI", env_auth="FUSEKI_AUTH", **kwargs
        )
        self.uri = normalize_fuseki_server_uri(self.uri)
        if dataset is None:
            self.dataset = DEFAULT_DATASET
        else:
            self.dataset = dataset
        self.ontologies_dataset = ontologies_dataset or DEFAULT_ONTOLOGIES_DATASET
        self.shapes_dataset = shapes_dataset or DEFAULT_SHAPES_DATASET

        # Initialize httpx client for async operations (recreated per event loop;
        # httpx.AsyncClient is bound to the loop it was created on).
        self._client: httpx.AsyncClient | None = None
        self._client_loop: asyncio.AbstractEventLoop | None = None

        self._full_catalog_fetches = 0
        self._graph_fetches = 0
        self._select_queries = 0
        self._construct_queries = 0
        self._last_catalog_was_partial = False

    async def async_init(self) -> None:
        """Initialize configured Fuseki datasets explicitly.

        Constructors stay side-effect free so callers can resolve tenancy first
        and then create datasets for the final dataset names.
        """
        # Use a temporary client to keep initialization independent from any
        # loop-bound long-lived client state.
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                await self._initialize_datasets()
            finally:
                # Restore original client
                self._client = original_client

    async def _initialize_datasets(self) -> None:
        """Create the configured facts/ontologies/shapes datasets when missing."""
        await self.init_dataset(self.dataset)
        seen = {self.dataset}
        for name in (self.ontologies_dataset, self.shapes_dataset):
            if name not in seen:
                seen.add(name)
                await self.init_dataset(name)

    def _prepare_auth(self) -> httpx.BasicAuth | None:
        """Prepare httpx BasicAuth from self.auth.

        Accepts ``user/password`` and ``user:password``. Both forms appear in
        the wild -- the colon form is what Fuseki's own docs and most HTTP
        tooling use -- and previously only the slash form parsed, so
        ``FUSEKI_AUTH=admin:secret`` silently produced *no* auth header and
        surfaced as an opaque 401. The separator that appears first wins, so a
        password containing the other character still round-trips.

        Returns:
            httpx.BasicAuth instance, or None when no auth is configured.
        """
        if not self.auth:
            return None
        if isinstance(self.auth, tuple):
            return httpx.BasicAuth(*self.auth)
        if isinstance(self.auth, str):
            positions = [
                (self.auth.index(sep), sep) for sep in ("/", ":") if sep in self.auth
            ]
            if positions:
                index, _ = min(positions)
                username, password = self.auth[:index], self.auth[index + 1 :]
                if username:
                    return httpx.BasicAuth(username, password)
            logger.warning(
                "FUSEKI_AUTH is set but is not in 'user/password' or 'user:password' "
                "form; proceeding without authentication."
            )
        return None

    async def _get_client(self) -> httpx.AsyncClient:
        """Get or create the httpx async client for the current running event loop."""
        loop = asyncio.get_running_loop()
        if self._client is not None and self._client_loop is loop:
            return self._client
        # Client from a prior asyncio.run() is bound to a closed loop; do not await
        # aclose() on it (that schedules callbacks on the dead loop).
        self._client = None
        self._client_loop = None
        auth = self._prepare_auth()
        self._client = httpx.AsyncClient(auth=auth, timeout=30.0)
        self._client_loop = loop
        return self._client

    async def close(self):
        """Close the httpx client."""
        if self._client is not None:
            await self._client.aclose()
            self._client = None
        self._client_loop = None

    def last_catalog_was_complete(self) -> bool:
        """False when the last full catalog fetch could not materialize every graph."""
        return not self._last_catalog_was_partial

    def supports_tenancy_partition(self) -> bool:
        return True

    async def update_tenancy(
        self,
        tenant: str,
        project: str,
        *,
        sep: str = TENANCY_SEP,
    ) -> None:
        """Switch the facts/ontologies/shapes Fuseki datasets for ``tenant`` / ``project``."""
        self.dataset = tenant_project_facts_name(tenant, project, sep=sep)
        self.ontologies_dataset = tenant_project_ontologies_name(
            tenant, project, sep=sep
        )
        self.shapes_dataset = tenant_project_shapes_name(tenant, project, sep=sep)
        await self._initialize_datasets()
        logger.info(
            "Fuseki tenancy set to tenant=%r project=%r "
            "(facts=%s ontologies=%s shapes=%s)",
            tenant,
            project,
            self.dataset,
            self.ontologies_dataset,
            self.shapes_dataset,
        )

    async def clean(self, *, include_shapes: bool = False) -> None:
        """Clear the configured facts and ontologies datasets (when distinct).

        The shapes dataset is retained unless ``include_shapes`` is set: dropping
        it disarms the SHACL gate silently.
        """
        assert self.dataset is not None, "Dataset should never be None"
        names = [self.dataset, self.ontologies_dataset]
        if include_shapes:
            names.append(self.shapes_dataset)
        cleaned: set[str] = set()
        for name in names:
            if name in cleaned:
                continue
            cleaned.add(name)
            await self._clean_dataset_by_name(name)
            logger.info("Fuseki dataset '%s' cleaned (all data deleted)", name)

    async def clean_tenancy(
        self,
        tenant: str,
        project: str,
        *,
        sep: str = TENANCY_SEP,
        include_shapes: bool = False,
    ) -> None:
        """Flush the datasets derived from ``tenant`` / ``project``.

        Shapes are retained unless ``include_shapes`` is set -- see :meth:`clean`.
        """
        facts = tenant_project_facts_name(tenant, project, sep=sep)
        ontos = tenant_project_ontologies_name(tenant, project, sep=sep)
        shapes = tenant_project_shapes_name(tenant, project, sep=sep)
        names = [facts, ontos] + ([shapes] if include_shapes else [])
        cleaned: set[str] = set()
        for name in names:
            if name in cleaned:
                continue
            cleaned.add(name)
            await self._clean_dataset_by_name(name)
        logger.info(
            "Fuseki tenancy flush tenant=%r project=%r "
            "(facts=%s ontologies=%s shapes=%s)",
            tenant,
            project,
            facts,
            ontos,
            shapes if include_shapes else "retained",
        )

    async def _clean_dataset_by_name(self, dataset_name: str) -> None:
        """Clean a specific dataset by name.

        This is a helper method that performs the actual cleaning of a single dataset.
        It deletes all named graphs and clears the default graph.

        Uses a temporary client to avoid event loop cleanup issues when called
        from different async contexts.

        Args:
            dataset_name: Name of the dataset to clean.

        Raises:
            Exception: If the cleanup operation fails.
        """
        # Use a temporary client to avoid event loop cleanup issues
        async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
            try:
                dataset_url = f"{self.uri}/{dataset_name}"
                sparql_update_url = f"{dataset_url}/update"
                sparql_url = f"{dataset_url}/sparql"

                # Delete all named graphs
                query = """
                SELECT DISTINCT ?g WHERE {
                  GRAPH ?g { ?s ?p ?o }
                }
                """
                response = await client.post(
                    sparql_url,
                    data={"query": query, "format": "application/sparql-results+json"},
                )

                if response.status_code == 200:
                    results = response.json()
                    tasks = []
                    for binding in results.get("results", {}).get("bindings", []):
                        graph_uri = binding["g"]["value"]
                        # Delete the named graph using SPARQL UPDATE
                        drop_query = f"DROP GRAPH <{graph_uri}>"
                        tasks.append(
                            client.post(
                                sparql_update_url,
                                data={"update": drop_query},
                            )
                        )

                    # Execute all deletions in parallel
                    delete_responses = await asyncio.gather(
                        *tasks, return_exceptions=True
                    )
                    for i, delete_response in enumerate(delete_responses):
                        graph_uri = results["results"]["bindings"][i]["g"]["value"]
                        if isinstance(delete_response, Exception):
                            logger.warning(
                                f"Failed to delete graph {graph_uri}: {delete_response}"
                            )
                        elif isinstance(delete_response, httpx.Response):
                            if delete_response.status_code in (200, 204):
                                logger.debug(f"Deleted named graph: {graph_uri}")
                            else:
                                logger.warning(
                                    f"Failed to delete graph {graph_uri}: {delete_response.status_code}"
                                )

                # Clear the default graph using SPARQL UPDATE
                clear_query = "CLEAR DEFAULT"
                clear_response = await client.post(
                    sparql_update_url,
                    data={"update": clear_query},
                )
                if clear_response.status_code in (200, 204):
                    logger.debug(f"Cleared default graph in dataset '{dataset_name}'")
                else:
                    logger.warning(
                        f"Failed to clear default graph in dataset '{dataset_name}': {clear_response.status_code}"
                    )
            except Exception as e:
                logger.error(f"Failed to clean dataset '{dataset_name}': {e}")
                raise

    async def init_dataset(self, dataset_name):
        """Initialize a Fuseki dataset.

        This method creates a new dataset in Fuseki if it doesn't already exist.
        It uses Fuseki's admin API to create the dataset with TDB2 storage.

        Uses a temporary client to avoid event loop cleanup issues when called
        from different async contexts.

        Args:
            dataset_name: Name of the dataset to create.

        Note:
            This method will not fail if the dataset already exists.
        """
        # Use a temporary client to avoid event loop cleanup issues
        async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
            fuseki_admin_url = f"{self.uri}/$/datasets"

            payload = {"dbName": dataset_name, "dbType": "tdb2"}

            headers = {"Content-Type": "application/x-www-form-urlencoded"}

            response = await client.post(
                fuseki_admin_url, data=payload, headers=headers
            )

            if response.status_code == 200 or response.status_code == 201:
                logger.info(f"Fuseki dataset '{dataset_name}' created successfully.")
            elif response.status_code == 409:
                logger.info(
                    f"Fuseki status code: {response.status_code}; {response.text.strip()}"
                )
            else:
                logger.error(
                    f"Failed to create dataset {dataset_name}. Status code: {response.status_code}"
                )
                logger.error(f"Response: {response.text.strip()}")

    def _get_dataset_url(self):
        """Get the full URL for the dataset.

        Returns:
            str: The complete URL for the dataset endpoint.
        """
        return f"{self.uri}/{self.dataset}"

    def _get_ontologies_dataset_url(self):
        """Get the full URL for the ontologies dataset.

        Returns:
            str: The complete URL for the ontologies dataset endpoint.
        """
        return f"{self.uri}/{self.ontologies_dataset}"

    def _get_shapes_dataset_url(self):
        """Get the full URL for the SHACL shapes dataset.

        Returns:
            str: The complete URL for the shapes dataset endpoint.
        """
        return f"{self.uri}/{self.shapes_dataset}"

    def _dataset_url_for(self, store: StoreKind) -> str:
        """Resolve a :data:`StoreKind` to its Fuseki dataset URL."""
        if store == "ontologies":
            return self._get_ontologies_dataset_url()
        if store == "shapes":
            return self._get_shapes_dataset_url()
        return self._get_dataset_url()

    async def drop_named_graph(
        self, graph_uri: str, *, store: StoreKind = "ontologies"
    ) -> None:
        """Drop a single named graph in the ontologies or main dataset."""
        dataset_url = self._dataset_url_for(store)
        update_url = f"{dataset_url}/update"
        drop_query = f"DROP GRAPH <{graph_uri}>"
        async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
            response = await client.post(update_url, data={"update": drop_query})
            if response.status_code not in (200, 204):
                logger.warning(
                    "Fuseki DROP GRAPH failed for %s: %s %s",
                    graph_uri,
                    response.status_code,
                    response.text,
                )

    async def drop_all_ontology_graphs_for_iri(
        self, ontology_iri: str, *, store: StoreKind = "ontologies"
    ) -> None:
        """Remove named graphs for ``ontology_iri`` (base and ``iri#...`` versioned)."""
        prefix = f"{ontology_iri}#"
        dataset_url = self._dataset_url_for(store)
        async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
            sparql_url = f"{dataset_url}/sparql"
            list_query = """
            SELECT DISTINCT ?g WHERE {
              GRAPH ?g { ?s ?p ?o }
            }
            """
            response = await client.post(
                sparql_url,
                data={"query": list_query, "format": "application/sparql-results+json"},
            )
            if response.status_code != 200:
                logger.error(
                    "Failed to list graphs from Fuseki %s dataset: %s",
                    store,
                    response.text,
                )
                return
            to_drop: list[str] = []
            for binding in response.json().get("results", {}).get("bindings", []):
                g = binding["g"]["value"]
                if g == ontology_iri or g.startswith(prefix):
                    to_drop.append(g)
            update_url = f"{dataset_url}/update"
            for graph_uri in to_drop:
                drop_query = f"DROP GRAPH <{graph_uri}>"
                dr = await client.post(update_url, data={"update": drop_query})
                if dr.status_code not in (200, 204):
                    logger.warning(
                        "Failed to drop graph %s: %s %s",
                        graph_uri,
                        dr.status_code,
                        dr.text,
                    )

    def fetch_ontologies(self) -> list[Ontology]:
        """Synchronous wrapper for fetch_ontologies.

        For async usage, use afetch_ontologies() instead.

        Raises:
            RuntimeError: If called from inside a running event loop; await
                :meth:`afetch_ontologies` there.
        """
        require_no_running_loop(
            "FusekiTripleStoreManager.fetch_ontologies",
            "FusekiTripleStoreManager.afetch_ontologies",
        )
        # Use a temporary client for this operation to avoid event loop cleanup issues
        return asyncio.run(self._fetch_ontologies_with_cleanup())

    async def afetch_ontologies(self) -> list[Ontology]:
        """Async version of fetch_ontologies.

        This is the preferred method when running in an async context.
        """
        return await self._fetch_ontologies_async()

    async def _fetch_ontologies_with_cleanup(self) -> list[Ontology]:
        """Wrapper that ensures proper cleanup when using asyncio.run().

        This method creates a temporary client and ensures it's properly closed
        before returning, preventing "Event loop is closed" errors.
        """
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                return await self._fetch_ontologies_async()
            finally:
                # Restore original client
                self._client = original_client

    async def _sparql_select_rows(
        self, client: httpx.AsyncClient, sparql_url: str, query: str
    ) -> list[dict[str, str]]:
        """POST a SPARQL SELECT and flatten its JSON bindings to lexical values."""
        self._select_queries += 1
        response = await client.post(
            sparql_url,
            data={"query": query, "format": "application/sparql-results+json"},
        )
        response.raise_for_status()
        return [
            {var: binding[var]["value"] for var in binding}
            for binding in response.json().get("results", {}).get("bindings", [])
        ]

    def _sparql_endpoint(self, *, store: StoreKind) -> str:
        """Resolve the SPARQL query endpoint for the active tenancy partition."""
        return f"{self._dataset_url_for(store)}/sparql"

    def supports_sparql_select(self) -> bool:
        return True

    def supports_sparql_construct(self) -> bool:
        return True

    async def aconstruct(
        self, query: str, *, store: StoreKind = "ontologies"
    ) -> RDFGraph:
        """Run a SPARQL CONSTRUCT against the active dataset, parsing Turtle back.

        Tenancy is implicit, as for :meth:`aselect`.
        """
        client = await self._get_client()
        self._construct_queries += 1
        response = await client.post(
            self._sparql_endpoint(store=store),
            data={"query": query},
            headers={"Accept": "text/turtle"},
        )
        response.raise_for_status()
        result = RDFGraph()
        text = response.text
        if text.strip():
            result.parse(data=text, format="turtle")
        return result

    async def aselect(
        self, query: str, *, store: StoreKind = "ontologies"
    ) -> list[dict[str, str]]:
        """Run a SPARQL SELECT against the active dataset.

        Tenancy is implicit: :meth:`update_tenancy` rewrites the dataset names this
        resolves through.
        """
        client = await self._get_client()
        return await self._sparql_select_rows(
            client,
            self._sparql_endpoint(store=store),
            query,
        )

    async def afetch_ontology_catalog(self) -> list[OntologyHeader]:
        """Read one header per stored ontology version via a single SELECT."""
        rows = await self.aselect(ONTOLOGY_HEADER_QUERY)
        return headers_from_select_rows(rows)

    async def afetch_ontologies_by_iri(self, iris: Sequence[str]) -> list[Ontology]:
        """Fetch only the named graphs backing ``iris``, skipping the rest."""
        if not iris:
            return await self.afetch_ontologies()
        wanted = set(iris)
        headers = dedupe_terminal_ontologies(await self.afetch_ontology_catalog())
        graph_uris = [header.graph_uri for header in headers if header.iri in wanted]
        if not graph_uris:
            return []
        client = await self._get_client()
        return await self._fetch_ontology_graphs(client, graph_uris)

    def catalog_io_stats(self) -> dict[str, int]:
        """Counters for catalog I/O, for tests and diagnostics."""
        return {
            "full_catalog_fetches": self._full_catalog_fetches,
            "graph_fetches": self._graph_fetches,
            "select_queries": self._select_queries,
            "construct_queries": self._construct_queries,
        }

    async def _list_ontology_graph_uris(self, client: httpx.AsyncClient) -> list[str]:
        """List every named graph in the ontologies dataset.

        Raises:
            TripleStoreUnavailableError: the listing could not be performed.
                This deliberately does **not** degrade to an empty list: an
                empty catalog is indistinguishable from "no ontologies stored",
                and ``ToolBox.initialize`` treats an empty catalog as grounds to
                prune every indexed ontology IRI from the vector store. A
                transient network error must not be able to wipe the index.
                Mirrors the contract stated for ``aselect``/``aconstruct`` on
                :class:`~ontocast.tool.triple_manager.core.TripleStoreManager`.
        """
        sparql_url = f"{self._get_ontologies_dataset_url()}/sparql"
        try:
            rows = await self._sparql_select_rows(
                client, sparql_url, LIST_NAMED_GRAPHS_QUERY
            )
        except httpx.HTTPError as exc:
            logger.error("Failed to list graphs from Fuseki: %s", exc)
            raise TripleStoreUnavailableError(
                f"Could not list named graphs in {sparql_url}: {exc}"
            ) from exc
        graph_uris = [row["g"] for row in rows if "g" in row]
        logger.debug("Found %d named graphs: %s", len(graph_uris), graph_uris)
        return graph_uris

    async def _fetch_ontology_graphs(
        self, client: httpx.AsyncClient, graph_uris: Sequence[str]
    ) -> list[Ontology]:
        """Materialize the named graphs in ``graph_uris`` in parallel."""

        async def fetch_single_ontology(graph_uri: str) -> Ontology | None:
            """Fetch a single ontology from a graph URI."""
            try:
                self._graph_fetches += 1
                graph = RDFGraph()
                # URL encode the graph URI to handle special characters like #
                encoded_graph_uri = quote(str(graph_uri), safe="/:")
                export_url = f"{self._get_ontologies_dataset_url()}/get?graph={encoded_graph_uri}"
                export_resp = await client.get(
                    export_url, headers={"Accept": "text/turtle"}
                )

                if export_resp.status_code == 200:
                    graph.parse(data=export_resp.text, format="turtle")
                    return ontology_from_named_graph(graph_uri, graph)
                else:
                    logger.warning(
                        f"Failed to fetch graph {graph_uri}: {export_resp.status_code}"
                    )
            except Exception as e:
                logger.warning(f"Error fetching ontology from {graph_uri}: {e}")
            return None

        results = await asyncio.gather(
            *[fetch_single_ontology(uri) for uri in graph_uris], return_exceptions=True
        )

        ontologies: list[Ontology] = []
        for result in results:
            if isinstance(result, Exception):
                logger.warning(f"Exception fetching ontology: {result}")
            elif isinstance(result, Ontology):
                ontologies.append(result)

        missing = len(graph_uris) - len(ontologies)
        if missing:
            # A partial catalog is as dangerous as an empty one: the ontologies
            # that failed to materialize look like orphans to the vector-store
            # prune. Record it so callers can refuse to treat this catalog as
            # authoritative.
            self._last_catalog_was_partial = True
            logger.error(
                "Fetched %d of %d ontology graphs; %d failed. The catalog is "
                "incomplete and must not be treated as authoritative.",
                len(ontologies),
                len(graph_uris),
                missing,
            )
        else:
            self._last_catalog_was_partial = False
        return ontologies

    async def _fetch_ontologies_async(self) -> list[Ontology]:
        """Fetch all ontologies from their corresponding named graphs.

        This method discovers all ontologies in the Fuseki ontologies dataset and
        fetches each one from its corresponding named graph. For versioned ontologies,
        it returns only the latest version for each unique ontology IRI.

        1. Discovery: List all named graphs (which may be versioned URIs)
        2. Fetching: Retrieve each ontology from its named graph (in parallel)
        3. Deduplication: For versioned ontologies, keep only the latest version

        Returns:
            list[Ontology]: List of the latest version of each ontology found.

        Example:
            >>> ontologies = await manager.fetch_ontologies()
            >>> for onto in ontologies:
            ...     print(f"Found ontology: {onto.iri} v{onto.version}")
        """
        self._full_catalog_fetches += 1
        client = await self._get_client()
        graph_uris = await self._list_ontology_graph_uris(client)
        if not graph_uris:
            return []
        all_ontologies = await self._fetch_ontology_graphs(client, graph_uris)
        ontologies = dedupe_terminal_ontologies(all_ontologies)
        logger.info(
            "Successfully loaded %d unique ontologies from Fuseki", len(ontologies)
        )
        return ontologies

    def serialize_graph(self, graph: Graph, **kwargs) -> bool:
        """Synchronous wrapper for serialize_graph.

        For async usage, use aserialize_graph() instead.

        Raises:
            RuntimeError: If called from inside a running event loop; await
                :meth:`aserialize_graph` there.
        """
        require_no_running_loop(
            "FusekiTripleStoreManager.serialize_graph",
            "FusekiTripleStoreManager.aserialize_graph",
        )
        return asyncio.run(self._serialize_graph_with_cleanup(graph, **kwargs))

    async def aserialize_graph(self, graph: Graph, **kwargs) -> bool:
        """Async version of serialize_graph.

        This is the preferred method when running in an async context.
        """
        return await self._serialize_graph_async(graph, **kwargs)

    async def _serialize_graph_with_cleanup(self, graph: Graph, **kwargs) -> bool:
        """Wrapper that ensures proper cleanup when using asyncio.run().

        This method creates a temporary client and ensures it's properly closed
        before returning, preventing "Event loop is closed" errors.
        """
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                return await self._serialize_graph_async(graph, **kwargs)
            finally:
                # Restore original client
                self._client = original_client

    async def _serialize_graph_async(self, graph: Graph, **kwargs) -> bool:
        """Store an RDF graph as a named graph in a specific Fuseki dataset.

        This is a private helper method that handles the common logic for storing
        graphs in Fuseki datasets.

        Args:
            graph: The RDF graph to store.
            **kwargs: ``graph_uri``, ``store`` (a :data:`StoreKind`, default
                ``"facts"``), ``default_graph_uri`` and ``log_prefix``.

        Returns:
            bool: True if the graph was successfully stored, False otherwise.
        """
        client = await self._get_client()
        graph_uri = kwargs.get("graph_uri")
        dataset_url = self._dataset_url_for(kwargs.get("store", "facts"))
        default_graph_uri = kwargs.get("default_graph_uri")
        log_prefix = kwargs.get("log_prefix")

        if isinstance(graph, RDFGraph):
            turtle_data = graph.serialize_canonical_turtle()
        else:
            rdf_graph = RDFGraph()
            for triple in graph:
                rdf_graph.add(triple)
            for prefix, namespace in graph.namespaces():
                rdf_graph.bind(prefix, namespace)
            turtle_data = rdf_graph.serialize_canonical_turtle()
        if graph_uri is None:
            graph_uri = default_graph_uri

        # URL encode the graph URI to handle special characters like #
        encoded_graph_uri = quote(str(graph_uri), safe="/:")
        url = f"{dataset_url}/data?graph={encoded_graph_uri}"
        headers = {"Content-Type": "text/turtle;charset=utf-8"}
        response = await client.put(url, headers=headers, content=turtle_data)
        if response.status_code in (200, 201, 204):
            logger.info(
                f"{log_prefix} graph {graph_uri} uploaded to Fuseki as named graph."
            )
            return True
        else:
            logger.error(
                f"Failed to upload {log_prefix.lower() if log_prefix else 'unknown'} graph {graph_uri}. Status code: {response.status_code}"
            )
            logger.error(f"Response: {response.text}")
            return False

    def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool:
        """Synchronous wrapper for serialize.

        For async usage, use aserialize() instead.

        Raises:
            RuntimeError: If called from inside a running event loop; await
                :meth:`aserialize` there.
        """
        require_no_running_loop(
            "FusekiTripleStoreManager.serialize",
            "FusekiTripleStoreManager.aserialize",
        )
        return asyncio.run(self._serialize_with_cleanup(o, **kwargs))

    async def aserialize(self, o: Ontology | RDFGraph, **kwargs) -> bool:
        """Async version of serialize.

        This is the preferred method when running in an async context.
        """
        return await self._serialize_async(o, **kwargs)

    async def _serialize_with_cleanup(self, o: Ontology | RDFGraph, **kwargs) -> bool:
        """Wrapper that ensures proper cleanup when using asyncio.run().

        This method creates a temporary client and ensures it's properly closed
        before returning, preventing "Event loop is closed" errors.
        """
        async with httpx.AsyncClient(
            auth=self._prepare_auth(), timeout=30.0
        ) as temp_client:
            # Temporarily replace the client
            original_client = self._client
            self._client = temp_client
            try:
                return await self._serialize_async(o, **kwargs)
            finally:
                # Restore original client
                self._client = original_client

    async def _serialize_async(self, o: Ontology | RDFGraph, **kwargs) -> bool:
        """Store an RDF graph as a named graph in Fuseki.

        This method stores the given RDF graph as a named graph in Fuseki.
        The graph name is taken from the graph_uri parameter or defaults to
        "urn:data:default".

        Args:
            o: RDF graph or Ontology object.
            **kwargs: ``graph_uri`` and ``store``. ``store`` overrides the
                partition the payload type would otherwise imply -- an
                ``Ontology`` carrying SHACL shapes goes to ``"shapes"``.

        Returns:
            bool: True if the graph was successfully stored, False otherwise.

        Example:
            >>> graph = RDFGraph()
            >>> success = await manager.serialize(graph)

            >>> success = await manager.serialize(graph, graph_uri="http://example.org/chunk1")
        """
        graph_uri = kwargs.get("graph_uri")
        requested: StoreKind | None = kwargs.get("store")

        if isinstance(o, Ontology):
            if o.iri and not o.is_null():
                # Persist author @prefix names as triples before they die at
                # the store boundary (idempotent, excluded from content hash).
                o.graph.materialize_prefix_declarations(URIRef(o.iri))
            graph = o.graph
            # Use versioned IRI for storage to enable multiple versions to coexist
            graph_uri = o.versioned_iri
            default_graph_uri = "urn:ontology:default"
            log_prefix = "Ontology"
            store: StoreKind = requested or "ontologies"
        elif isinstance(o, RDFGraph):
            graph = o
            default_graph_uri = "urn:data:default"
            log_prefix = "Graph"
            store = requested or "facts"
        else:
            raise TypeError(f"unsupported obj of type {type(o)} received")

        return await self._serialize_graph_async(
            graph=graph,
            graph_uri=graph_uri,
            store=store,
            default_graph_uri=default_graph_uri,
            log_prefix=log_prefix,
        )

__init__(uri=None, auth=None, dataset=None, ontologies_dataset=None, shapes_dataset=None, **kwargs)

Initialize the Fuseki triple store manager.

This method sets up the connection to Fuseki and creates the dataset if it doesn't exist. The dataset is NOT cleaned on initialization.

Parameters:

Name	Type	Description	Default
uri		Fuseki HTTP service root (e.g. http://localhost:3030), not .../dataset/name and not a #/dataset/... UI link.	None
auth		Authentication tuple (username, password) or string in "user/password" format.	None
dataset		Facts dataset name (Fuseki API path segment).	None
ontologies_dataset		Ontologies dataset name (separate Fuseki dataset).	None
shapes_dataset		SHACL shapes dataset name (separate Fuseki dataset).	None
**kwargs		Additional keyword arguments passed to the parent class.	{}

Example

manager = FusekiTripleStoreManager( ... uri="http://localhost:3030", ... dataset="acme--demo--facts", ... ontologies_dataset="acme--demo--ontologies", ... ) await manager.clean()

Source code in ontocast/tool/triple_manager/fuseki.py

def __init__(
    self,
    uri=None,
    auth=None,
    dataset=None,
    ontologies_dataset=None,
    shapes_dataset=None,
    **kwargs,
):
    """Initialize the Fuseki triple store manager.

    This method sets up the connection to Fuseki and creates the dataset
    if it doesn't exist. The dataset is NOT cleaned on initialization.

    Args:
        uri: Fuseki HTTP service root (e.g. ``http://localhost:3030``), not
            ``.../dataset/name`` and not a ``#/dataset/...`` UI link.
        auth: Authentication tuple (username, password) or string in "user/password" format.
        dataset: Facts dataset name (Fuseki API path segment).
        ontologies_dataset: Ontologies dataset name (separate Fuseki dataset).
        shapes_dataset: SHACL shapes dataset name (separate Fuseki dataset).
        **kwargs: Additional keyword arguments passed to the parent class.

    Example:
        >>> manager = FusekiTripleStoreManager(
        ...     uri="http://localhost:3030",
        ...     dataset="acme--demo--facts",
        ...     ontologies_dataset="acme--demo--ontologies",
        ... )
        >>> await manager.clean()
    """
    super().__init__(
        uri=uri, auth=auth, env_uri="FUSEKI_URI", env_auth="FUSEKI_AUTH", **kwargs
    )
    self.uri = normalize_fuseki_server_uri(self.uri)
    if dataset is None:
        self.dataset = DEFAULT_DATASET
    else:
        self.dataset = dataset
    self.ontologies_dataset = ontologies_dataset or DEFAULT_ONTOLOGIES_DATASET
    self.shapes_dataset = shapes_dataset or DEFAULT_SHAPES_DATASET

    # Initialize httpx client for async operations (recreated per event loop;
    # httpx.AsyncClient is bound to the loop it was created on).
    self._client: httpx.AsyncClient | None = None
    self._client_loop: asyncio.AbstractEventLoop | None = None

    self._full_catalog_fetches = 0
    self._graph_fetches = 0
    self._select_queries = 0
    self._construct_queries = 0
    self._last_catalog_was_partial = False

aconstruct(query, *, store='ontologies') async

Run a SPARQL CONSTRUCT against the active dataset, parsing Turtle back.

Tenancy is implicit, as for :meth:aselect.

Source code in ontocast/tool/triple_manager/fuseki.py

async def aconstruct(
    self, query: str, *, store: StoreKind = "ontologies"
) -> RDFGraph:
    """Run a SPARQL CONSTRUCT against the active dataset, parsing Turtle back.

    Tenancy is implicit, as for :meth:`aselect`.
    """
    client = await self._get_client()
    self._construct_queries += 1
    response = await client.post(
        self._sparql_endpoint(store=store),
        data={"query": query},
        headers={"Accept": "text/turtle"},
    )
    response.raise_for_status()
    result = RDFGraph()
    text = response.text
    if text.strip():
        result.parse(data=text, format="turtle")
    return result

afetch_ontologies() async

Async version of fetch_ontologies.

This is the preferred method when running in an async context.

Source code in ontocast/tool/triple_manager/fuseki.py

async def afetch_ontologies(self) -> list[Ontology]:
    """Async version of fetch_ontologies.

    This is the preferred method when running in an async context.
    """
    return await self._fetch_ontologies_async()

afetch_ontologies_by_iri(iris) async

Fetch only the named graphs backing iris, skipping the rest.

Source code in ontocast/tool/triple_manager/fuseki.py

async def afetch_ontologies_by_iri(self, iris: Sequence[str]) -> list[Ontology]:
    """Fetch only the named graphs backing ``iris``, skipping the rest."""
    if not iris:
        return await self.afetch_ontologies()
    wanted = set(iris)
    headers = dedupe_terminal_ontologies(await self.afetch_ontology_catalog())
    graph_uris = [header.graph_uri for header in headers if header.iri in wanted]
    if not graph_uris:
        return []
    client = await self._get_client()
    return await self._fetch_ontology_graphs(client, graph_uris)

afetch_ontology_catalog() async

Read one header per stored ontology version via a single SELECT.

Source code in ontocast/tool/triple_manager/fuseki.py

async def afetch_ontology_catalog(self) -> list[OntologyHeader]:
    """Read one header per stored ontology version via a single SELECT."""
    rows = await self.aselect(ONTOLOGY_HEADER_QUERY)
    return headers_from_select_rows(rows)

aselect(query, *, store='ontologies') async

Run a SPARQL SELECT against the active dataset.

Tenancy is implicit: :meth:update_tenancy rewrites the dataset names this resolves through.

Source code in ontocast/tool/triple_manager/fuseki.py

async def aselect(
    self, query: str, *, store: StoreKind = "ontologies"
) -> list[dict[str, str]]:
    """Run a SPARQL SELECT against the active dataset.

    Tenancy is implicit: :meth:`update_tenancy` rewrites the dataset names this
    resolves through.
    """
    client = await self._get_client()
    return await self._sparql_select_rows(
        client,
        self._sparql_endpoint(store=store),
        query,
    )

aserialize(o, **kwargs) async

Async version of serialize.

This is the preferred method when running in an async context.

Source code in ontocast/tool/triple_manager/fuseki.py

async def aserialize(self, o: Ontology | RDFGraph, **kwargs) -> bool:
    """Async version of serialize.

    This is the preferred method when running in an async context.
    """
    return await self._serialize_async(o, **kwargs)

aserialize_graph(graph, **kwargs) async

Async version of serialize_graph.

This is the preferred method when running in an async context.

Source code in ontocast/tool/triple_manager/fuseki.py

async def aserialize_graph(self, graph: Graph, **kwargs) -> bool:
    """Async version of serialize_graph.

    This is the preferred method when running in an async context.
    """
    return await self._serialize_graph_async(graph, **kwargs)

async_init() async

Initialize configured Fuseki datasets explicitly.

Constructors stay side-effect free so callers can resolve tenancy first and then create datasets for the final dataset names.

Source code in ontocast/tool/triple_manager/fuseki.py

async def async_init(self) -> None:
    """Initialize configured Fuseki datasets explicitly.

    Constructors stay side-effect free so callers can resolve tenancy first
    and then create datasets for the final dataset names.
    """
    # Use a temporary client to keep initialization independent from any
    # loop-bound long-lived client state.
    async with httpx.AsyncClient(
        auth=self._prepare_auth(), timeout=30.0
    ) as temp_client:
        # Temporarily replace the client
        original_client = self._client
        self._client = temp_client
        try:
            await self._initialize_datasets()
        finally:
            # Restore original client
            self._client = original_client

catalog_io_stats()

Counters for catalog I/O, for tests and diagnostics.

Source code in ontocast/tool/triple_manager/fuseki.py

def catalog_io_stats(self) -> dict[str, int]:
    """Counters for catalog I/O, for tests and diagnostics."""
    return {
        "full_catalog_fetches": self._full_catalog_fetches,
        "graph_fetches": self._graph_fetches,
        "select_queries": self._select_queries,
        "construct_queries": self._construct_queries,
    }

clean(*, include_shapes=False) async

Clear the configured facts and ontologies datasets (when distinct).

The shapes dataset is retained unless include_shapes is set: dropping it disarms the SHACL gate silently.

Source code in ontocast/tool/triple_manager/fuseki.py

async def clean(self, *, include_shapes: bool = False) -> None:
    """Clear the configured facts and ontologies datasets (when distinct).

    The shapes dataset is retained unless ``include_shapes`` is set: dropping
    it disarms the SHACL gate silently.
    """
    assert self.dataset is not None, "Dataset should never be None"
    names = [self.dataset, self.ontologies_dataset]
    if include_shapes:
        names.append(self.shapes_dataset)
    cleaned: set[str] = set()
    for name in names:
        if name in cleaned:
            continue
        cleaned.add(name)
        await self._clean_dataset_by_name(name)
        logger.info("Fuseki dataset '%s' cleaned (all data deleted)", name)

clean_tenancy(tenant, project, *, sep=TENANCY_SEP, include_shapes=False) async

Flush the datasets derived from tenant / project.

Shapes are retained unless include_shapes is set -- see :meth:clean.

Source code in ontocast/tool/triple_manager/fuseki.py

async def clean_tenancy(
    self,
    tenant: str,
    project: str,
    *,
    sep: str = TENANCY_SEP,
    include_shapes: bool = False,
) -> None:
    """Flush the datasets derived from ``tenant`` / ``project``.

    Shapes are retained unless ``include_shapes`` is set -- see :meth:`clean`.
    """
    facts = tenant_project_facts_name(tenant, project, sep=sep)
    ontos = tenant_project_ontologies_name(tenant, project, sep=sep)
    shapes = tenant_project_shapes_name(tenant, project, sep=sep)
    names = [facts, ontos] + ([shapes] if include_shapes else [])
    cleaned: set[str] = set()
    for name in names:
        if name in cleaned:
            continue
        cleaned.add(name)
        await self._clean_dataset_by_name(name)
    logger.info(
        "Fuseki tenancy flush tenant=%r project=%r "
        "(facts=%s ontologies=%s shapes=%s)",
        tenant,
        project,
        facts,
        ontos,
        shapes if include_shapes else "retained",
    )

close() async

Close the httpx client.

Source code in ontocast/tool/triple_manager/fuseki.py

async def close(self):
    """Close the httpx client."""
    if self._client is not None:
        await self._client.aclose()
        self._client = None
    self._client_loop = None

drop_all_ontology_graphs_for_iri(ontology_iri, *, store='ontologies') async

Remove named graphs for ontology_iri (base and iri#... versioned).

Source code in ontocast/tool/triple_manager/fuseki.py

async def drop_all_ontology_graphs_for_iri(
    self, ontology_iri: str, *, store: StoreKind = "ontologies"
) -> None:
    """Remove named graphs for ``ontology_iri`` (base and ``iri#...`` versioned)."""
    prefix = f"{ontology_iri}#"
    dataset_url = self._dataset_url_for(store)
    async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
        sparql_url = f"{dataset_url}/sparql"
        list_query = """
        SELECT DISTINCT ?g WHERE {
          GRAPH ?g { ?s ?p ?o }
        }
        """
        response = await client.post(
            sparql_url,
            data={"query": list_query, "format": "application/sparql-results+json"},
        )
        if response.status_code != 200:
            logger.error(
                "Failed to list graphs from Fuseki %s dataset: %s",
                store,
                response.text,
            )
            return
        to_drop: list[str] = []
        for binding in response.json().get("results", {}).get("bindings", []):
            g = binding["g"]["value"]
            if g == ontology_iri or g.startswith(prefix):
                to_drop.append(g)
        update_url = f"{dataset_url}/update"
        for graph_uri in to_drop:
            drop_query = f"DROP GRAPH <{graph_uri}>"
            dr = await client.post(update_url, data={"update": drop_query})
            if dr.status_code not in (200, 204):
                logger.warning(
                    "Failed to drop graph %s: %s %s",
                    graph_uri,
                    dr.status_code,
                    dr.text,
                )

drop_named_graph(graph_uri, *, store='ontologies') async

Drop a single named graph in the ontologies or main dataset.

Source code in ontocast/tool/triple_manager/fuseki.py

async def drop_named_graph(
    self, graph_uri: str, *, store: StoreKind = "ontologies"
) -> None:
    """Drop a single named graph in the ontologies or main dataset."""
    dataset_url = self._dataset_url_for(store)
    update_url = f"{dataset_url}/update"
    drop_query = f"DROP GRAPH <{graph_uri}>"
    async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
        response = await client.post(update_url, data={"update": drop_query})
        if response.status_code not in (200, 204):
            logger.warning(
                "Fuseki DROP GRAPH failed for %s: %s %s",
                graph_uri,
                response.status_code,
                response.text,
            )

fetch_ontologies()

Synchronous wrapper for fetch_ontologies.

For async usage, use afetch_ontologies() instead.

Raises:

Type	Description
RuntimeError	If called from inside a running event loop; await :meth:afetch_ontologies there.

Source code in ontocast/tool/triple_manager/fuseki.py

def fetch_ontologies(self) -> list[Ontology]:
    """Synchronous wrapper for fetch_ontologies.

    For async usage, use afetch_ontologies() instead.

    Raises:
        RuntimeError: If called from inside a running event loop; await
            :meth:`afetch_ontologies` there.
    """
    require_no_running_loop(
        "FusekiTripleStoreManager.fetch_ontologies",
        "FusekiTripleStoreManager.afetch_ontologies",
    )
    # Use a temporary client for this operation to avoid event loop cleanup issues
    return asyncio.run(self._fetch_ontologies_with_cleanup())

init_dataset(dataset_name) async

Initialize a Fuseki dataset.

This method creates a new dataset in Fuseki if it doesn't already exist. It uses Fuseki's admin API to create the dataset with TDB2 storage.

Uses a temporary client to avoid event loop cleanup issues when called from different async contexts.

Parameters:

Name	Type	Description	Default
dataset_name		Name of the dataset to create.	required

Note

This method will not fail if the dataset already exists.

Source code in ontocast/tool/triple_manager/fuseki.py

async def init_dataset(self, dataset_name):
    """Initialize a Fuseki dataset.

    This method creates a new dataset in Fuseki if it doesn't already exist.
    It uses Fuseki's admin API to create the dataset with TDB2 storage.

    Uses a temporary client to avoid event loop cleanup issues when called
    from different async contexts.

    Args:
        dataset_name: Name of the dataset to create.

    Note:
        This method will not fail if the dataset already exists.
    """
    # Use a temporary client to avoid event loop cleanup issues
    async with httpx.AsyncClient(auth=self._prepare_auth(), timeout=30.0) as client:
        fuseki_admin_url = f"{self.uri}/$/datasets"

        payload = {"dbName": dataset_name, "dbType": "tdb2"}

        headers = {"Content-Type": "application/x-www-form-urlencoded"}

        response = await client.post(
            fuseki_admin_url, data=payload, headers=headers
        )

        if response.status_code == 200 or response.status_code == 201:
            logger.info(f"Fuseki dataset '{dataset_name}' created successfully.")
        elif response.status_code == 409:
            logger.info(
                f"Fuseki status code: {response.status_code}; {response.text.strip()}"
            )
        else:
            logger.error(
                f"Failed to create dataset {dataset_name}. Status code: {response.status_code}"
            )
            logger.error(f"Response: {response.text.strip()}")

last_catalog_was_complete()

False when the last full catalog fetch could not materialize every graph.

Source code in ontocast/tool/triple_manager/fuseki.py

def last_catalog_was_complete(self) -> bool:
    """False when the last full catalog fetch could not materialize every graph."""
    return not self._last_catalog_was_partial

serialize(o, **kwargs)

Synchronous wrapper for serialize.

For async usage, use aserialize() instead.

Raises:

Type	Description
RuntimeError	If called from inside a running event loop; await :meth:aserialize there.

Source code in ontocast/tool/triple_manager/fuseki.py

def serialize(self, o: Ontology | RDFGraph, **kwargs) -> bool:
    """Synchronous wrapper for serialize.

    For async usage, use aserialize() instead.

    Raises:
        RuntimeError: If called from inside a running event loop; await
            :meth:`aserialize` there.
    """
    require_no_running_loop(
        "FusekiTripleStoreManager.serialize",
        "FusekiTripleStoreManager.aserialize",
    )
    return asyncio.run(self._serialize_with_cleanup(o, **kwargs))

serialize_graph(graph, **kwargs)

Synchronous wrapper for serialize_graph.

For async usage, use aserialize_graph() instead.

Raises:

Type	Description
RuntimeError	If called from inside a running event loop; await :meth:aserialize_graph there.

Source code in ontocast/tool/triple_manager/fuseki.py

def serialize_graph(self, graph: Graph, **kwargs) -> bool:
    """Synchronous wrapper for serialize_graph.

    For async usage, use aserialize_graph() instead.

    Raises:
        RuntimeError: If called from inside a running event loop; await
            :meth:`aserialize_graph` there.
    """
    require_no_running_loop(
        "FusekiTripleStoreManager.serialize_graph",
        "FusekiTripleStoreManager.aserialize_graph",
    )
    return asyncio.run(self._serialize_graph_with_cleanup(graph, **kwargs))

update_tenancy(tenant, project, *, sep=TENANCY_SEP) async

Switch the facts/ontologies/shapes Fuseki datasets for tenant / project.

Source code in ontocast/tool/triple_manager/fuseki.py

async def update_tenancy(
    self,
    tenant: str,
    project: str,
    *,
    sep: str = TENANCY_SEP,
) -> None:
    """Switch the facts/ontologies/shapes Fuseki datasets for ``tenant`` / ``project``."""
    self.dataset = tenant_project_facts_name(tenant, project, sep=sep)
    self.ontologies_dataset = tenant_project_ontologies_name(
        tenant, project, sep=sep
    )
    self.shapes_dataset = tenant_project_shapes_name(tenant, project, sep=sep)
    await self._initialize_datasets()
    logger.info(
        "Fuseki tenancy set to tenant=%r project=%r "
        "(facts=%s ontologies=%s shapes=%s)",
        tenant,
        project,
        self.dataset,
        self.ontologies_dataset,
        self.shapes_dataset,
    )

normalize_fuseki_server_uri(raw)

Normalize FUSEKI_URI to the Fuseki HTTP service root.

SPARQL and Graph Store HTTP endpoints are {base}/{dataset}/sparql, {base}/{dataset}/update, and so on. The Fuseki web UI links look like http://host:port/#/dataset/dataset_name; the fragment is client-side only and must not be sent with API requests. Trailing slashes on the base URL are removed so {base} and {dataset} concatenate to correct paths.

Parameters:

Name	Type	Description	Default
raw	str | None	Connection URI (e.g. from FUSEKI_URI).	required

Returns:

Type	Description
str | None	Normalized base URL, or None if raw is None. Malformed values
str | None	without scheme/netloc are returned unchanged (after strip).

Source code in ontocast/tool/triple_manager/fuseki.py

def normalize_fuseki_server_uri(raw: str | None) -> str | None:
    """Normalize ``FUSEKI_URI`` to the Fuseki HTTP service root.

    SPARQL and Graph Store HTTP endpoints are ``{base}/{dataset}/sparql``,
    ``{base}/{dataset}/update``, and so on. The Fuseki web UI links look like
    ``http://host:port/#/dataset/dataset_name``; the fragment is client-side only
    and must not be sent with API requests. Trailing slashes on the base URL are
    removed so ``{base}`` and ``{dataset}`` concatenate to correct paths.

    Args:
        raw: Connection URI (e.g. from ``FUSEKI_URI``).

    Returns:
        Normalized base URL, or ``None`` if ``raw`` is ``None``. Malformed values
        without scheme/netloc are returned unchanged (after ``strip``).
    """
    if raw is None:
        return None
    text = raw.strip()
    parsed = urlparse(text)
    if not parsed.scheme or not parsed.netloc:
        return text
    path = (parsed.path or "").rstrip("/")
    return urlunparse(
        (parsed.scheme, parsed.netloc, path, parsed.params, parsed.query, "")
    )