Scenario Bundles — architecture & developer guide

Living document

This page is a developer-oriented map of how the Scenario Bundles feature is wired together. It is expected to grow and be corrected over time — if you touch this area and something below is stale, please update it in the same PR. Line numbers are pointers, not contracts; treat them as "look near here".

The user- and API-facing overview lives on the Scenario Bundles feature page (what a bundle is, and the create/get/update/delete JSON API). This page is about the code: which app owns what, how the layers talk, and where to make a change.

The three layers

A scenario bundle is authored in a React UI, processed by a Django app (factsheet), and stored as RDF triples in the OEKG (a Jena Fuseki triple store), with terms drawn from the OEO ontology.

flowchart LR
    subgraph Browser
        UI["React frontend<br/>factsheet/frontend/src"]
    end
    subgraph Django
        V["factsheet app<br/>views.py / urls.py / helper.py"]
        C["factsheet/oekg/connection.py<br/>(SPARQL + OWL graph setup)"]
        Q["oekg app<br/>oekg/sparqlQuery.py"]
    end
    subgraph Stores
        F[("OEKG<br/>Jena Fuseki<br/>triple store")]
        O[("OEO ontology<br/>oeo-full.owl")]
    end

    UI -- "axios JSON<br/>/scenario-bundles/*" --> V
    V --> C
    Q --> C
    V -- "read term lists" --> O
    C -- "SPARQL query/update" --> F
    Q -- "SPARQL query/update" --> F

• Frontend — factsheet/frontend/src/ (React, built with Vite). Authoring, overview, comparison and history views.
• Backend — the Django factsheet app (URL prefix scenario-bundles/). Turns JSON payloads into RDF and back.
• Graph store (OEKG) — Apache Jena Fuseki, reached over SPARQL.
• Ontology (OEO) — the Open Energy Ontology, the source of the controlled vocabulary (sectors, technologies, study descriptors, …) offered in the form.

Where the code lives

Concern	Location
Django app (views, URLs, helpers)	factsheet/
URL routing (scenario-bundles/ prefix)	oeplatform/urls.py, factsheet/urls.py
React source	factsheet/frontend/src/
Frontend build config	vite.config.mjs (entry factsheet: ./factsheet/frontend/src/index.jsx)
OEKG/OEO connection + query wrappers	factsheet/oekg/connection.py, factsheet/oekg/filters.py, factsheet/oekg/namespaces.py
Reusable SPARQL query module	oekg/sparqlQuery.py (top-level oekg app)
SPARQL UI (YASGUI)	oekg app (oekg:main)

Two things are called oekg

• factsheet/oekg/ — an internal package of the factsheet app. Holds the connection setup: the SPARQLWrapper clients (sparql, update_endpoint) and the in-memory OEO graph (oeo, oeo_owl).
• oekg/ (repo root) — a separate Django app with the reusable sparqlQuery.py query functions, models, views, and the YASGUI SPARQL explorer. See its own oekg/README.md.

New OEKG-interaction functionality should live in / extend the top-level oekg app; the factsheet app imports from it.

Frontend

Django serves a single template for every scenario-bundles route (factsheets_index_view, factsheet/views.py); the React app reads window.location.pathname and routes client-side in factsheet/frontend/src/App.jsx:

Route	Component	Purpose
scenario-bundles/main	home.jsx → components/customTable.jsx	All-bundles overview / listing + filter
scenario-bundles/id/<uuid> or /new	components/scenarioBundle.tsx	Create / edit / view one bundle (tabbed form)
scenario-bundles/compare/…	comparisonBoardMain	Compare bundles
scenario-bundles/oekg_history, …/oekg_modifications	history / diff views	Change history

The bundle authoring form (scenarioBundle.tsx) is organised into tabs; the "Sectors and technology" tab and the study descriptors section are the parts fed by the OEO (see below). The frontend talks to Django with axios, posting/getting JSON to scenario-bundles/<name>/ endpoints; a CSRF token is included on write requests.

Backend & data flow

The factsheet views translate between the frontend JSON (see the example payloads) and RDF triples in the OEKG.

• Bundle lifecycle — add/, get/, update/, delete/ map to view functions in factsheet/views.py. On write, the JSON is parsed and emitted as triples (e.g. sectors/divisions via predicate OEO_00390079); on read, triples are re-assembled into the JSON the form expects.
• Populating the form's option lists — populate_factsheets_elements_view (factsheet/views.py, URL name populate-factsheets-elements) returns the controlled-vocabulary lists the form renders: sector_divisions, sectors, scenario_descriptors, technologies. The React form fetches this once when it loads (scenarioBundle.tsx).
• Listing & filtering — the overview (customTable.jsx) fetches scenario-bundles/all/; filter queries run through oekg/sparqlQuery.py (e.g. scenario_bundle_filter_oekg, list_factsheets_oekg).

Two ways to reach the OEO

Both patterns are established in the codebase; pick per use case:

1. In-memory OWL graph — oeo (an rdflib.Graph parsed from the on-disk oeo-full.owl) and oeo_owl (owlready2), set up in factsheet/oekg/connection.py. Used for ontology-structure reads: class hierarchies, labels, definitions (IAO_0000115 / SKOS / rdfs:comment). See build_sector_dropdowns_from_oeo and get_all_sub_classes in factsheet/helper.py.
2. Live SPARQL endpoint — the sparql / update_endpoint clients (SPARQLWrapper) in factsheet/oekg/connection.py, driven from oekg/sparqlQuery.py. Used for querying/updating the OEKG instance data (the bundles themselves). More efficient for data than parsing a graph into Python objects.

The OEO-driven form fields

Several form fields offer terms from the OEO. All of them are now sourced from the ontology; the payload is assembled by populate_factsheets_elements_view and served at scenario-bundles/populate_factsheets_elements/.

Field	Sourced from	Dynamic?
Sectors (under a division)	OEO graph via is defined by (OEO_00000504), factsheet/helper.py	✅ queried live
Sector divisions	OEO graph, build_sector_dropdowns_from_oeo, factsheet/helper.py	✅ queried live
Study descriptors	OEO graph, terms annotated oekg annotation (OEO_00020425) with a value starting study descriptor, factsheet/helper.py	✅ queried live
Technologies	OEO graph, served by populate_factsheets_elements_view	✅ queried live

Both builders are memoized at module level on first call, because the OEO only changes when the process restarts.

Sector divisions

build_sector_dropdowns_from_oeo returns (divisions, sectors). Two things about it are easy to get wrong and are pinned by factsheet/tests/test_sector_dropdowns.py:

• Divisions are modelled two ways in the OEO. Some carry their members as individuals (KSG, CRF 2006); others are classes whose members declare the division through an rdf:type restriction (NC/BR, EU legislation). An individual-only query misses the second kind entirely — which is what the former hardcoded list papered over. Each division reports which it is via kind (individuals or tree).
• A division with no members is still listed (NACE has none), and the division asserting is defined by about itself is filtered out of its own options — CRF 2006 offers 108, not 109.

The last entry is the Other division, which carries the full sector tree rather than a flat list. The legacy flat sectors list is still served alongside, so older consumers keep working.

Study descriptors

Served as study_descriptors, an array of [label, iri, definition] triples — the same shape as the former hardcoded StudyKeywords array, so consumers did not have to change their rendering. The annotation match is deliberately a prefix, because the OEO carries both study descriptor and the inconsistent study descriptor tag.

Three places consume the list, by two different routes — worth knowing before changing the shape:

• bundle edit checkboxes and overview chips — scenarioBundle.tsx, which already fetches the populate endpoint for its other fields and reads data.study_descriptors straight off that payload
• the all-bundles filter dialog — FactsheetFilterDialog.jsx
• the comparison board — comparisonBoardItems.jsx

The latter two use the useStudyDescriptors hook (scenarioBundleUtilityComponents/useStudyDescriptors.js), which holds a module-level cache and a shared in-flight promise, so N mounts trigger one request. It also exports getStudyDescriptors() for plain helper functions, returning [] until the fetch resolves.

customTable.jsx is not a consumer: it holds the selected-keyword state and passes it down to the filter dialog, which resolves the labels itself.

Related surfaces

• OEKG SPARQL explorer — the oekg app exposes a YASGUI query UI (oekg:main), linked from the Scenario Bundles navbar dropdown (base/templates/base/_header.html).
• OEKG chat — an external chatbot for asking questions about the OEKG, https://oekg-chat.openenergyplatform.org/. (Being linked from the Scenario Bundles nav and the overview page — see the wayfinder map.)
• OEKG Web-API — see OEKG API for the read-only SPARQL endpoint and Writing scenario bundles for the REST API that writes them. The singular scenario-bundle/scenario/manage-datasets/ route is superseded by that API and still served; it is described in the reference under Scenario Bundles (legacy).

How to extend this feature

• Add / change a form option list: decide OWL-graph vs SPARQL (above), add or adjust the query in factsheet/helper.py (ontology structure) or oekg/sparqlQuery.py (instance data), expose it through populate_factsheets_elements_view, and consume it in scenarioBundle.tsx.
• Add a bundle field: thread it through the create/get/update views in factsheet/views.py (JSON ⇄ triples) and the matching form tab in scenarioBundle.tsx.
• Rebuild the frontend: npm run build (Vite; output under assets/, served via django-vite). See the frontend workflow.
• Add an addressable sub-resource to the REST API: add a field table and a BundlePart entry, then a route. The machinery works against a subject plus a table rather than against the bundle, so a new part is a table entry and a route — not a module. See The API write path.

The API write path

The feature has two write paths, and everything above this section describes the first one: the UI's. Session-authenticated RPC endpoints in factsheet/ build the bundle triple by triple through rdflib, which is a thin client over Fuseki — so a create writing ~200 triples is ~200 HTTP requests, and it is not atomic.

The second is the OEKG REST API under /api/v0/scenario-bundles/, in the oekg app. It is a standalone surface built beside the UI's rather than on top of it, and it differs from the UI path in four ways a contributor needs to know before touching it:

• One request, one transaction. The API does not use rdflib as transport. It builds the graph in memory and sends one SPARQL update request, and one update request is one transaction across ;-separated operations. A write therefore either lands whole or not at all.
• The shape is enforced before the write. The canonical SHACL shape comes from the oekg repository as a build-time artifact — manage.py fetch_oekg_shapes — and the API validates the bundle's post-state in-process before committing. Validating a PATCH diff instead would pass vacuously, because nothing targets an untyped node.
• A write is judged by what it introduces. The pre-state is validated too, and only violations the write adds are refused. Without this the API could not write to any bundle the browser had created. A create has no pre-state and so stays strict.
• Writes are guarded by a version. Every mutating request carries the version it believes it is editing; the guard binds that version and the bundle's existence into the update's WHERE, and reads a write token back to tell a winner from a loser.

Two constraints hold for anything added here. The API must not import factsheet/oekg/connection.py, which parses the full OEO at module import; and the graph store and Postgres cannot share a transaction, so the graph commits first and a failed history write is reported rather than rolled back.

The write sequence

The sequence every mutating endpoint shares — exists, owned, precondition, validate the whole bundle, guard, read back, record — lives in one place, so an endpoint is a payload and a field table rather than a repetition of that order.

One write to a bundle, whatever part of it the request names.

Every mutating endpoint in this API does the same five things in the same order, and gets them wrong in the same ways if it does them itself:

1. the bundle has to exist, or there is nothing to write to;
2. the caller has to own it;
3. the caller has to say which version it is editing;
4. the whole bundle with the change in it has to satisfy the shape -- every constraint in the shape is bundle-local, so a part on its own is not a unit the shape can judge -- but only as far as this write is responsible for it: a violation the bundle already had is not this caller's to answer for, and refusing it would make an inherited defect unfixable through the API;
5. the write has to be guarded on that version from inside, and read back, because the store answers 200 whether or not the guard held.

So the sequence lives here once rather than in each view. open_bundle does 1, require_write does 2 and 3, and apply does 4 and 5 and records the history. They are three calls rather than one so a sub-resource view can check that its part exists between the first and the second, and keep the same order one level down. A view is then only the part that differs: which triples change.

A whole-bundle delete is the one write that does not end in apply, because there is no bundle left to validate or to version afterwards. It takes the same first two steps, adds confirm_deletion -- the retyped acronym, which only this write asks for -- and ends in destroy. Everything it does differently is written down there.

Refusals travel as exceptions. A helper that returns either a value or a refusal makes every call site test which it got, and one forgotten test is a refusal silently ignored -- so a refusal is raised and a view catches it in the same place it already catches the store's failures.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

BundleWrite dataclass

A bundle read, checked, and ready to be changed.

Source code in oekg/writes.py

@dataclass
class BundleWrite:
    """A bundle read, checked, and ready to be changed."""

    uid: str
    store: GraphStore
    pre_state: Graph
    version: BundleVersion
    actor: object = None
    post_state: Optional[Graph] = field(default=None)
    history_recorded: bool = True
    ownership_forgotten: bool = True

    def require_write(self, request) -> None:
        """Refuse unless this caller may write, and said which version.

        Raises ``Refused`` with a `403`, a `428` or a `412`.
        """
        if not may_write_bundle(request.user, self.uid):
            raise Refused(
                Response(
                    {
                        "detail": (
                            "Only an owner of this scenario bundle may change it. "
                            "A bundle with no recorded owner can be changed by an "
                            "administrator only."
                        )
                    },
                    status=status.HTTP_403_FORBIDDEN,
                )
            )
        refusal = precondition_refusal(request, self.version)
        if refusal is not None:
            raise Refused(refusal)

    def confirm_deletion(self, request) -> str:
        """Refuse unless the caller retyped this bundle's acronym. Returns it.

        Beside `require_write` because it is the same kind of thing -- what a
        request has to carry before it may proceed -- and only a whole-bundle
        delete asks for it. The acronym is read from the bundle rather than
        taken from the caller, which is the whole point.

        Raises ``Refused`` with a `400`.
        """
        stored = self.pre_state.value(bundle_iri(self.uid), DC.acronym)
        acronym = None if stored is None else str(stored)
        refusal = confirmation_refusal(request, acronym)
        if refusal is not None:
            raise Refused(refusal)
        return acronym

    @property
    def gaps(self) -> dict:
        """What was lost after the graph committed, if anything was.

        Empty when everything landed, so a client does not have to check
        something on every response to learn that the ordinary thing happened;
        these keys are there to name the exception.
        """
        lost = {}
        if not self.history_recorded:
            lost["history_recorded"] = False
        if not self.ownership_forgotten:
            lost["ownership_forgotten"] = False
        return lost

    def labels_of(self, iris: list) -> dict:
        """Labels for referenced nodes, answered from the bundle where possible."""
        return labels_of(self.store, iris, self.pre_state)

    def apply(
        self,
        *,
        removed: Graph,
        added: Graph,
        verb: str,
        resource_type: URIRef = BUNDLE_CLASS,
        resource_uuid: Optional[str] = None,
        guard: str = "",
    ) -> None:
        """Validate the post-state, write it under the guard, record it.

        ``guard`` is an extra pattern bound into the write's own ``WHERE``, for
        a condition the bundle's version cannot express -- a delete uses it to
        assert that nothing outside the bundle started citing the nodes it is
        about to remove. It shares the `409`, because the answer to either is
        the same: read again and retry.

        Raises ``Refused`` with a `400` if the shape objects **to something this
        write introduced**, and with a `409` if the guard did not hold. Nothing
        is written in either case, and nothing on this object is updated either
        -- a caller that catches a refusal must not find a post-state
        describing a write that did not happen.
        """
        if self.post_state is not None:
            raise RuntimeError(
                "This bundle has already been written. A second write needs a "
                "fresh read: this one still holds the state from before the "
                "first, and would silently undo it."
            )

        # Both sides through the same pruner: otherwise they differ by how they
        # were assembled rather than by what the write did, and that difference
        # reads as violations nobody introduced.
        post_state = bundle_subgraph(self.pre_state - removed + added, self.uid)
        introduced, inherited = introduced_violations(
            bundle_subgraph(self.pre_state, self.uid), post_state
        )
        if introduced:
            raise Refused(
                Response(
                    {
                        "detail": (
                            "This change would add violations of the OEKG shape."
                        ),
                        "violations": [v.as_dict() for v in introduced],
                        # Named rather than hidden: the bundle is not clean, and a
                        # caller should be able to learn that without being blamed
                        # for it.
                        "pre_existing_violations": inherited,
                    },
                    status=status.HTTP_400_BAD_REQUEST,
                )
            )

        token = mint_write_token()
        self.store.update(
            guarded_operation(
                self.store,
                self.uid,
                self.version,
                token,
                delete=removed,
                insert=added,
                condition=guard,
            )
        )
        if not write_applied(self.store, self.uid, token):
            raise Refused(
                Response(
                    {
                        "detail": (
                            "The bundle changed while this request was being "
                            "prepared, so nothing was written. Read it again, "
                            "apply the change to what you get back, and retry."
                        )
                    },
                    status=status.HTTP_409_CONFLICT,
                )
            )

        # Only now: everything above could still have refused, and a refusal
        # must leave this object describing the bundle as it still is.
        self.post_state = post_state
        before = self.version.number
        self.version = BundleVersion(before + 1, token)
        # The graph has committed. Nothing from here may turn a successful
        # write into an error; a lost history entry is reported beside the
        # success it qualifies, never instead of it.
        self.history_recorded = record_write(
            bundle_uid=self.uid,
            verb=verb,
            actor=self.actor,
            version_before=before,
            version_after=self.version.number,
            removed=removed,
            added=added,
            resource_type=resource_type,
            resource_uuid=resource_uuid,
        )

    def destroy(self, acronym: str) -> Removal:
        """Delete this bundle, its bookkeeping and its records of ownership.

        Not `apply`, and every difference is a consequence of there being no
        bundle afterwards:

        - **Nothing to validate.** The shape judges a bundle; an absent one is
          not a worse bundle, it is no bundle.
        - **No version bump.** The version node goes with the bundle it counts.
          Leaving it is what the browser's delete does (issue #2440), and it
          leaves a node asserting that a bundle is at version 4 when there is
          no bundle -- which the next write would then guard against and match.
        - **The read-back asks whether the bundle is gone**, not whether a
          token landed. A delete's own signal is absence, and absence needs no
          token to be unambiguous: nothing else in this API can make a bundle
          stop existing, so finding it still there can only mean the guard did
          not hold.
        - **What follows the commit is different too.** The ownership rows go,
          and the history records an event rather than a diff.

        Returns what the delete came to -- which nodes went and which were kept
        because something outside still cites them. The plan is made here
        rather than by the caller, so the one place that decides what a delete
        reaches is the one place that writes it.

        Raises ``Refused`` with a `409` if the guard did not hold -- either the
        bundle moved since it was read, or something outside it started citing
        a node this plan was about to delete. The advice is the same for both:
        read it again and retry.
        """
        removal = plan_bundle_removal(self.store, self.pre_state, self.uid)
        removed = Graph()
        removed += removal.removed
        removed += version_node_triples(self.uid, self.version)
        guard = version_guard(self.uid, self.version)
        if removal.guard:
            guard = f"{guard} {removal.guard}"

        self.store.update(self.store.guarded_modification(guard, delete=removed))
        if bundle_in(self.store, self.uid):
            raise Refused(
                Response(
                    {
                        "detail": (
                            "The bundle changed while this request was being "
                            "prepared, so nothing was deleted. Read it again, "
                            "check it is still the one you meant, and retry."
                        )
                    },
                    status=status.HTTP_409_CONFLICT,
                )
            )

        # The graph has committed. Nothing from here may turn a successful
        # delete into an error; what is lost is named beside the success it
        # qualifies, never instead of it.
        self.ownership_forgotten = forget_ownership(self.uid)
        self.history_recorded = record_bundle_deletion(
            bundle_uid=self.uid,
            acronym=acronym,
            actor=self.actor,
            version_before=self.version.number,
        )
        return removal

gaps property

What was lost after the graph committed, if anything was.

Empty when everything landed, so a client does not have to check something on every response to learn that the ordinary thing happened; these keys are there to name the exception.

apply(*, removed, added, verb, resource_type=BUNDLE_CLASS, resource_uuid=None, guard='')

Validate the post-state, write it under the guard, record it.

guard is an extra pattern bound into the write's own WHERE, for a condition the bundle's version cannot express -- a delete uses it to assert that nothing outside the bundle started citing the nodes it is about to remove. It shares the 409, because the answer to either is the same: read again and retry.

Raises Refused with a 400 if the shape objects to something this write introduced, and with a 409 if the guard did not hold. Nothing is written in either case, and nothing on this object is updated either -- a caller that catches a refusal must not find a post-state describing a write that did not happen.

Source code in oekg/writes.py

def apply(
    self,
    *,
    removed: Graph,
    added: Graph,
    verb: str,
    resource_type: URIRef = BUNDLE_CLASS,
    resource_uuid: Optional[str] = None,
    guard: str = "",
) -> None:
    """Validate the post-state, write it under the guard, record it.

    ``guard`` is an extra pattern bound into the write's own ``WHERE``, for
    a condition the bundle's version cannot express -- a delete uses it to
    assert that nothing outside the bundle started citing the nodes it is
    about to remove. It shares the `409`, because the answer to either is
    the same: read again and retry.

    Raises ``Refused`` with a `400` if the shape objects **to something this
    write introduced**, and with a `409` if the guard did not hold. Nothing
    is written in either case, and nothing on this object is updated either
    -- a caller that catches a refusal must not find a post-state
    describing a write that did not happen.
    """
    if self.post_state is not None:
        raise RuntimeError(
            "This bundle has already been written. A second write needs a "
            "fresh read: this one still holds the state from before the "
            "first, and would silently undo it."
        )

    # Both sides through the same pruner: otherwise they differ by how they
    # were assembled rather than by what the write did, and that difference
    # reads as violations nobody introduced.
    post_state = bundle_subgraph(self.pre_state - removed + added, self.uid)
    introduced, inherited = introduced_violations(
        bundle_subgraph(self.pre_state, self.uid), post_state
    )
    if introduced:
        raise Refused(
            Response(
                {
                    "detail": (
                        "This change would add violations of the OEKG shape."
                    ),
                    "violations": [v.as_dict() for v in introduced],
                    # Named rather than hidden: the bundle is not clean, and a
                    # caller should be able to learn that without being blamed
                    # for it.
                    "pre_existing_violations": inherited,
                },
                status=status.HTTP_400_BAD_REQUEST,
            )
        )

    token = mint_write_token()
    self.store.update(
        guarded_operation(
            self.store,
            self.uid,
            self.version,
            token,
            delete=removed,
            insert=added,
            condition=guard,
        )
    )
    if not write_applied(self.store, self.uid, token):
        raise Refused(
            Response(
                {
                    "detail": (
                        "The bundle changed while this request was being "
                        "prepared, so nothing was written. Read it again, "
                        "apply the change to what you get back, and retry."
                    )
                },
                status=status.HTTP_409_CONFLICT,
            )
        )

    # Only now: everything above could still have refused, and a refusal
    # must leave this object describing the bundle as it still is.
    self.post_state = post_state
    before = self.version.number
    self.version = BundleVersion(before + 1, token)
    # The graph has committed. Nothing from here may turn a successful
    # write into an error; a lost history entry is reported beside the
    # success it qualifies, never instead of it.
    self.history_recorded = record_write(
        bundle_uid=self.uid,
        verb=verb,
        actor=self.actor,
        version_before=before,
        version_after=self.version.number,
        removed=removed,
        added=added,
        resource_type=resource_type,
        resource_uuid=resource_uuid,
    )

confirm_deletion(request)

Refuse unless the caller retyped this bundle's acronym. Returns it.

Beside require_write because it is the same kind of thing -- what a request has to carry before it may proceed -- and only a whole-bundle delete asks for it. The acronym is read from the bundle rather than taken from the caller, which is the whole point.

Raises Refused with a 400.

Source code in oekg/writes.py

def confirm_deletion(self, request) -> str:
    """Refuse unless the caller retyped this bundle's acronym. Returns it.

    Beside `require_write` because it is the same kind of thing -- what a
    request has to carry before it may proceed -- and only a whole-bundle
    delete asks for it. The acronym is read from the bundle rather than
    taken from the caller, which is the whole point.

    Raises ``Refused`` with a `400`.
    """
    stored = self.pre_state.value(bundle_iri(self.uid), DC.acronym)
    acronym = None if stored is None else str(stored)
    refusal = confirmation_refusal(request, acronym)
    if refusal is not None:
        raise Refused(refusal)
    return acronym

destroy(acronym)

Delete this bundle, its bookkeeping and its records of ownership.

Not apply, and every difference is a consequence of there being no bundle afterwards:

• Nothing to validate. The shape judges a bundle; an absent one is not a worse bundle, it is no bundle.
• No version bump. The version node goes with the bundle it counts. Leaving it is what the browser's delete does (issue #2440), and it leaves a node asserting that a bundle is at version 4 when there is no bundle -- which the next write would then guard against and match.
• The read-back asks whether the bundle is gone, not whether a token landed. A delete's own signal is absence, and absence needs no token to be unambiguous: nothing else in this API can make a bundle stop existing, so finding it still there can only mean the guard did not hold.
• What follows the commit is different too. The ownership rows go, and the history records an event rather than a diff.

Returns what the delete came to -- which nodes went and which were kept because something outside still cites them. The plan is made here rather than by the caller, so the one place that decides what a delete reaches is the one place that writes it.

Raises Refused with a 409 if the guard did not hold -- either the bundle moved since it was read, or something outside it started citing a node this plan was about to delete. The advice is the same for both: read it again and retry.

Source code in oekg/writes.py

def destroy(self, acronym: str) -> Removal:
    """Delete this bundle, its bookkeeping and its records of ownership.

    Not `apply`, and every difference is a consequence of there being no
    bundle afterwards:

    - **Nothing to validate.** The shape judges a bundle; an absent one is
      not a worse bundle, it is no bundle.
    - **No version bump.** The version node goes with the bundle it counts.
      Leaving it is what the browser's delete does (issue #2440), and it
      leaves a node asserting that a bundle is at version 4 when there is
      no bundle -- which the next write would then guard against and match.
    - **The read-back asks whether the bundle is gone**, not whether a
      token landed. A delete's own signal is absence, and absence needs no
      token to be unambiguous: nothing else in this API can make a bundle
      stop existing, so finding it still there can only mean the guard did
      not hold.
    - **What follows the commit is different too.** The ownership rows go,
      and the history records an event rather than a diff.

    Returns what the delete came to -- which nodes went and which were kept
    because something outside still cites them. The plan is made here
    rather than by the caller, so the one place that decides what a delete
    reaches is the one place that writes it.

    Raises ``Refused`` with a `409` if the guard did not hold -- either the
    bundle moved since it was read, or something outside it started citing
    a node this plan was about to delete. The advice is the same for both:
    read it again and retry.
    """
    removal = plan_bundle_removal(self.store, self.pre_state, self.uid)
    removed = Graph()
    removed += removal.removed
    removed += version_node_triples(self.uid, self.version)
    guard = version_guard(self.uid, self.version)
    if removal.guard:
        guard = f"{guard} {removal.guard}"

    self.store.update(self.store.guarded_modification(guard, delete=removed))
    if bundle_in(self.store, self.uid):
        raise Refused(
            Response(
                {
                    "detail": (
                        "The bundle changed while this request was being "
                        "prepared, so nothing was deleted. Read it again, "
                        "check it is still the one you meant, and retry."
                    )
                },
                status=status.HTTP_409_CONFLICT,
            )
        )

    # The graph has committed. Nothing from here may turn a successful
    # delete into an error; what is lost is named beside the success it
    # qualifies, never instead of it.
    self.ownership_forgotten = forget_ownership(self.uid)
    self.history_recorded = record_bundle_deletion(
        bundle_uid=self.uid,
        acronym=acronym,
        actor=self.actor,
        version_before=self.version.number,
    )
    return removal

labels_of(iris)

Labels for referenced nodes, answered from the bundle where possible.

Source code in oekg/writes.py

def labels_of(self, iris: list) -> dict:
    """Labels for referenced nodes, answered from the bundle where possible."""
    return labels_of(self.store, iris, self.pre_state)

require_write(request)

Refuse unless this caller may write, and said which version.

Raises Refused with a 403, a 428 or a 412.

Source code in oekg/writes.py

def require_write(self, request) -> None:
    """Refuse unless this caller may write, and said which version.

    Raises ``Refused`` with a `403`, a `428` or a `412`.
    """
    if not may_write_bundle(request.user, self.uid):
        raise Refused(
            Response(
                {
                    "detail": (
                        "Only an owner of this scenario bundle may change it. "
                        "A bundle with no recorded owner can be changed by an "
                        "administrator only."
                    )
                },
                status=status.HTTP_403_FORBIDDEN,
            )
        )
    refusal = precondition_refusal(request, self.version)
    if refusal is not None:
        raise Refused(refusal)

open_bundle(request, uid)

Read a bundle, or refuse with a 404. Raises Refused.

Existence first, as it is for the two-step delete: one order for the whole API rather than one per endpoint. Reads are public, so answering 404 before authorisation reveals nothing a GET would not -- and a request for something that is not there should hear that, rather than being told its precondition is missing for a resource that does not exist.

Separate from require_write so a sub-resource view can check that its part exists in between, and keep the same order one level down.

Source code in oekg/writes.py

def open_bundle(request, uid: str) -> BundleWrite:
    """Read a bundle, or refuse with a `404`. Raises ``Refused``.

    **Existence first**, as it is for the two-step delete: one order for the
    whole API rather than one per endpoint. Reads are public, so answering 404
    before authorisation reveals nothing a `GET` would not -- and a request for
    something that is not there should hear that, rather than being told its
    precondition is missing for a resource that does not exist.

    Separate from ``require_write`` so a sub-resource view can check that *its*
    part exists in between, and keep the same order one level down.
    """
    if not bundle_exists(uid):
        raise Refused(
            Response(
                {"detail": f"No scenario bundle {uid}."},
                status=status.HTTP_404_NOT_FOUND,
            )
        )

    store = GraphStore.from_settings()
    pre_state = read_bundle(store, uid)
    if pre_state is None:
        raise Refused(
            Response(
                {"detail": f"No scenario bundle {uid}."},
                status=status.HTTP_404_NOT_FOUND,
            )
        )

    return BundleWrite(
        uid=uid,
        store=store,
        pre_state=pre_state,
        version=read_version(store, uid),
        actor=getattr(request, "user", None),
    )

refuse_bundle_renames(payload, known_labels)

refuse_renames for a whole-bundle payload, nested parts included.

The pair of bundles.bundle_referenced_iris: that one finds the IRIs a whole payload points at, this one refuses the ones it would rename. Two endpoints take a whole bundle -- the create and the replace -- and a loop written twice is a loop that will cover the parts in one of them and not in the other the next time a part is added.

Source code in oekg/writes.py

def refuse_bundle_renames(payload: dict, known_labels: dict) -> None:
    """`refuse_renames` for a whole-bundle payload, nested parts included.

    The pair of `bundles.bundle_referenced_iris`: that one finds the IRIs a
    whole payload points at, this one refuses the ones it would rename. Two
    endpoints take a whole bundle -- the create and the replace -- and a loop
    written twice is a loop that will cover the parts in one of them and not in
    the other the next time a part is added.
    """
    refuse_renames(payload, known_labels, BUNDLE_FIELDS)
    for part in BUNDLE_PARTS:
        for nested in payload.get(part.payload_key) or []:
            refuse_renames(nested, known_labels, part.fields)

refuse_renames(payload, known_labels, fields)

Refuse a payload that gives an existing node a different label.

The API offers no rename. Shared IRIs stay shared -- that is the point of a graph -- but a shared contact, organisation, funder or region is cited by other bundles, so letting one payload rewrite its label would change every one of them. Referencing such a node is allowed; renaming it is not, and the difference is the label sent alongside the iri.

Driven by the field table rather than by a list of field names, because a list stops covering a table the moment that table gains a node field, and the refusal then quietly passes writes it should refuse.

Source code in oekg/writes.py

def refuse_renames(payload: dict, known_labels: dict, fields: tuple) -> None:
    """Refuse a payload that gives an existing node a different label.

    **The API offers no rename.** Shared IRIs stay shared -- that is the point
    of a graph -- but a shared contact, organisation, funder or region is cited
    by other bundles, so letting one payload rewrite its label would change
    every one of them. Referencing such a node is allowed; renaming it is not,
    and the difference is the label sent alongside the iri.

    Driven by the field table rather than by a list of field names, because a
    list stops covering a table the moment that table gains a node field, and
    the refusal then quietly passes writes it should refuse.
    """
    conflicts = [
        {
            "iri": entry["iri"],
            "stored_label": known_labels[entry["iri"]],
            "sent_label": entry["label"],
        }
        for field in fields
        if field.kind == NODE
        for entry in (payload.get(field.name) or [])
        if entry.get("iri") in known_labels
        and entry["label"] != known_labels[entry["iri"]]
    ]
    if conflicts:
        raise Refused(
            Response(
                {
                    "detail": (
                        "A shared node cannot be renamed through this API. "
                        "Reference it by iri and send the label it already has, or "
                        "omit the iri to mint a new node."
                    ),
                    "conflicts": conflicts,
                },
                status=status.HTTP_400_BAD_REQUEST,
            )
        )

What a mutating request has to carry before it is allowed to proceed.

Two guards live here, and they defend different accidents. The version catches somebody changed this since you looked. The retyped acronym catches right verb, wrong identifier -- a pipeline looping over a list and reaching the wrong entry -- which no version can catch, because the version it names is the correct current version of the wrong bundle.

Only the whole-bundle delete asks for the second one. Ceremony is proportional to blast radius: everything else in this API can be written again.

If-Match is required on every mutating call and there is no opt-out. The alternative — optional, with same-field writes falling back to last-write-wins — was on the table with a worked example and declined: a client that does not know about the header would then overwrite somebody's change and neither of them would be told. The cost is named rather than discovered: deliberate overwriting is not reachable, and a repair pipeline that wants it must read first, every time.

Three refusals, because they mean three different things to a client:

• 428 you did not say which version you were editing;
• 412 you said one, and it is not the current version;
• 409 the server's own guard fired — the bundle moved between the read the validation needed and the write. That one is raised at the write, not here.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

confirmation_refusal(request, acronym)

Why this delete may not proceed on its confirmation, if it may not.

The acronym is retyped as a query parameter and compared exactly with the one stored. Exactly, because the whole value of the check is that it cannot be satisfied by a value the caller already had in hand for another bundle -- normalising it away would let NEMO-2030 confirm a delete of nemo 2030, which is the confusion the check exists to catch.

Both refusals are 400 rather than 412: nothing here is a precondition on the bundle's state. A missing or wrong token is a badly formed request, and the bundle is exactly as the caller last read it.

Source code in oekg/preconditions.py

def confirmation_refusal(request, acronym: str) -> Optional[Response]:
    """Why this delete may not proceed on its confirmation, if it may not.

    The acronym is retyped as a query parameter and compared **exactly** with
    the one stored. Exactly, because the whole value of the check is that it
    cannot be satisfied by a value the caller already had in hand for another
    bundle -- normalising it away would let `NEMO-2030` confirm a delete of
    `nemo 2030`, which is the confusion the check exists to catch.

    Both refusals are `400` rather than `412`: nothing here is a precondition
    on the bundle's state. A missing or wrong token is a badly formed request,
    and the bundle is exactly as the caller last read it.
    """
    if acronym is None:
        # Not reachable for a bundle this API created -- the shape requires an
        # acronym -- but the browser wrote most of the bundles in the graph and
        # the API judges a write by what it introduces, not by what it found.
        # Refusing here rather than inventing a substitute token keeps the
        # check meaning what it says; the way out is to give the bundle an
        # acronym with a `PATCH`, which is allowed precisely because the
        # missing one is a violation this caller did not introduce.
        return _bad_request(
            "This scenario bundle has no acronym, so there is nothing to "
            "confirm a delete with. Give it one with a PATCH first, then "
            "delete it."
        )
    given = request.query_params.get(CONFIRM)
    if given is None:
        return _bad_request(
            "Deleting a whole scenario bundle is irreversible, so it has to be "
            f"confirmed: repeat the bundle's acronym as ?{CONFIRM}=<acronym>. "
            "Read the bundle first -- the acronym is in the response, and so "
            "is the version this delete also needs."
        )
    if given != acronym:
        return _bad_request(
            f"The confirmation {given!r} is not this bundle's acronym, so "
            "nothing was deleted. Check that this is the bundle you meant to "
            "delete before retrying."
        )
    return None

precondition_refusal(request, version)

Why this request may not proceed on its precondition, if it may not.

Source code in oekg/preconditions.py

def precondition_refusal(request, version: BundleVersion) -> Optional[Response]:
    """Why this request may not proceed on its precondition, if it may not."""
    named = versions_named(request)
    if named is None:
        return Response(
            {
                "detail": (
                    "This write needs an If-Match header carrying the version "
                    "you read, so that it cannot silently overwrite a change "
                    f"made since. The bundle is at {version.etag}."
                )
            },
            status=status.HTTP_428_PRECONDITION_REQUIRED,
        )
    if str(version.number) not in named:
        return Response(
            {
                "detail": (
                    "The bundle is not at the version this request expects, so "
                    f"nothing was written. It is at {version.etag}. Read it "
                    "again and apply the change to what you get back."
                )
            },
            status=status.HTTP_412_PRECONDITION_FAILED,
        )
    return None

versions_named(request)

The versions an If-Match names, or None if it names none.

Lenient in what it accepts and strict in what it emits: a weak validator or a bare number is read as the version it plainly is, while a value that is not a version simply matches nothing and is refused as stale.

* names no version. It is a legal header value, and it satisfies the letter of the precondition while withholding the one thing the precondition is for, so it reads here as an absent header rather than as a match.

Source code in oekg/preconditions.py

def versions_named(request):
    """The versions an ``If-Match`` names, or ``None`` if it names none.

    Lenient in what it accepts and strict in what it emits: a weak validator or
    a bare number is read as the version it plainly is, while a value that is
    not a version simply matches nothing and is refused as stale.

    ``*`` names no version. It is a legal header value, and it satisfies the
    letter of the precondition while withholding the one thing the precondition
    is for, so it reads here as an absent header rather than as a match.
    """
    header = request.headers.get("If-Match")
    if header is None:
        return None
    named = []
    for entry in header.split(","):
        entry = entry.strip()
        if entry == "*":
            return None
        if entry.startswith("W/"):
            entry = entry[2:].strip()
        named.append(entry.strip('"'))
    return named

Who may write a scenario bundle, and what happens to that record.

The rule is asked by every mutating endpoint, because ownership is asked in half a dozen places -- sub-resources, the two-step delete, replace -- and a rule spelled out at each of them would drift at the first change. When the group system is connected to bundles, this is the place that learns about it.

The delete's counterpart lives here too: ownership records are removed with the bundle they describe, which the platform has never done.

Two properties are deliberate rather than incidental:

• Ownership is asked of the access-control model, not compared against a creator. Multi-owner bundles already exist -- an admin command adds a second owner -- so a single-owner comparison would be wrong today, not just after some future change.
• A bundle with no ownership record is administrator-only. Records of unknown provenance fail closed. That is the platform's existing behaviour for these bundles, kept on purpose rather than inherited by accident.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

forget_ownership(uid)

Remove the records saying who owns uid. Returns whether it worked.

Called after a bundle has been deleted, because ownership data that outlives the thing it describes is the platform's existing behaviour and is not defensible: the rows accumulate, they name a bundle nobody can read, and the next bundle minted at that identifier -- which cannot happen today, but is exactly the kind of thing a later change makes possible -- would inherit an owner nobody granted.

Never raises. The graph has already committed by the time this runs, so a failure here cannot be answered with an error without denying a delete that happened. A leftover row is inert: it grants access to nothing, because every endpoint checks the graph for the bundle first.

Source code in oekg/permissions.py

def forget_ownership(uid: str) -> bool:
    """Remove the records saying who owns ``uid``. Returns whether it worked.

    Called after a bundle has been deleted, because ownership data that
    outlives the thing it describes is the platform's existing behaviour and is
    not defensible: the rows accumulate, they name a bundle nobody can read,
    and the next bundle minted at that identifier -- which cannot happen today,
    but is exactly the kind of thing a later change makes possible -- would
    inherit an owner nobody granted.

    **Never raises.** The graph has already committed by the time this runs, so
    a failure here cannot be answered with an error without denying a delete
    that happened. A leftover row is inert: it grants access to nothing,
    because every endpoint checks the graph for the bundle first.
    """
    try:
        # A savepoint, for the same reason the history writer takes one: a
        # rejected statement caught here would otherwise poison the surrounding
        # transaction and surface at the next query, far from its cause.
        with transaction.atomic():
            ScenarioBundleAccessControl.objects.filter(bundle_id=uid).delete()
    except Exception:
        logger.exception(
            "OEKG bundle %s was deleted from the graph but its ownership rows "
            "could not be removed. They now name a bundle that does not exist.",
            uid,
        )
        return False
    return True

may_write_bundle(user, uid)

Whether user may change the bundle uid.

Source code in oekg/permissions.py

def may_write_bundle(user, uid: str) -> bool:
    """Whether ``user`` may change the bundle ``uid``."""
    if not getattr(user, "is_authenticated", False):
        return False
    if getattr(user, "is_admin", False):
        return True
    return ScenarioBundleAccessControl.user_has_access(user, uid)

Validation and the shape

Shape validation for scenario bundles, in one function.

The post-state is what gets validated -- the bundle as it would be after the write, assembled in memory, before anything is sent to the store. Not the diff: a diff carries no rdf:type, so no shape targets it and it conforms vacuously. That would not be weak validation, it would be a confident false pass.

The whole graph is unnecessary. The shape has no cross-bundle constraint, so one bundle's post-state is a complete unit.

The label subset is merged in for the same reason it exists: ex:CommonShape requires exactly one rdfs:label on every OEO term a bundle picks, and the payload carries labels only for the nodes it mints. Only the labels of terms the post-state actually names are merged -- see _with_the_labels_it_names.

One seam, one function: post-state graph in, violations out. The engine behind it is then swappable -- the fallback if the library ever stalls is Jena's own SHACL engine, which this stack already deploys.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

ShapeViolation dataclass

One thing the shape objects to, in the shape's own words.

Source code in oekg/validation.py

@dataclass(frozen=True)
class ShapeViolation:
    """One thing the shape objects to, in the shape's own words."""

    message: str
    focus_node: Optional[str] = None
    path: Optional[str] = None
    value: Optional[str] = None

    def as_dict(self) -> dict:
        return {
            "message": self.message,
            "focus_node": self.focus_node,
            "path": self.path,
            "value": self.value,
        }

introduced_violations(before, after)

What after violates that before did not, and how many it inherited.

A write is judged by what it adds. The alternative -- the post-state must conform, full stop -- sounds stricter and is, but it makes an existing defect unfixable through this API: the fields most often missing are the ones a person would supply by patching, and the patch would be refused for the very thing it came to fix. The promise that matters survives either way, because nothing invalid is written by this API.

Compared as a multiset, so a bundle already missing one required field may not come out missing two. And by violation identity -- message, focus node, path and value together -- so swapping one violation for another counts as introducing one, which comparing counts alone would miss.

Both graphs must be assembled the same way. Hand this the pruned pre-state, not the raw read, or the two differ by how they were built rather than by what the write did, and that shows up as violations nobody introduced.

Source code in oekg/validation.py

def introduced_violations(before: Graph, after: Graph) -> Tuple[List, int]:
    """What ``after`` violates that ``before`` did not, and how many it inherited.

    A write is judged by what it **adds**. The alternative -- the post-state must
    conform, full stop -- sounds stricter and is, but it makes an existing defect
    unfixable through this API: the fields most often missing are the ones a
    person would supply by patching, and the patch would be refused for the very
    thing it came to fix. The promise that matters survives either way, because
    nothing invalid is written *by this API*.

    Compared as a **multiset**, so a bundle already missing one required field
    may not come out missing two. And by violation identity -- message, focus
    node, path and value together -- so swapping one violation for another
    counts as introducing one, which comparing counts alone would miss.

    Both graphs must be assembled the same way. Hand this the pruned pre-state,
    not the raw read, or the two differ by how they were built rather than by
    what the write did, and that shows up as violations nobody introduced.
    """
    inherited = Counter(validate_post_state(before))
    arrived = Counter(validate_post_state(after))
    return list((arrived - inherited).elements()), sum(inherited.values())

validate_post_state(post_state)

Validate a bundle subgraph. An empty list means it conforms.

Source code in oekg/validation.py

def validate_post_state(post_state: Graph) -> List[ShapeViolation]:
    """Validate a bundle subgraph. An empty list means it conforms."""
    shape = shape_graph()
    conforms, report, _ = pyshacl_validate(
        _with_the_labels_it_names(post_state),
        shacl_graph=shape,
        advanced=True,
        inference="none",
    )
    if conforms:
        return []

    violations = [
        _violation(report, result)
        for result in report.subjects(RDF.type, SH.ValidationResult)
    ]
    # Sorted so the same invalid payload always reports in the same order:
    # pyshacl walks the report graph, whose iteration order is not stable.
    return sorted(violations, key=lambda v: (v.path or "", v.message))

The canonical SHACL shape, read as data.

The shape is the contract for what a scenario bundle is, and it is still moving in its own repository. So nothing here is copied into Python: the enumerations a payload may pick from and the messages a rejection carries are both read from the artifact manage.py fetch_oekg_shapes puts on disk.

That artifact is the cache key. A redeploy with a new pin changes the file, and the file's identity changes with it, so the cache invalidates itself and no process serves yesterday's enumerations.

Two things this module does NOT do:

• It does not fall back. A missing artifact raises rather than validating against nothing, because a validator that silently disappears is worse than one that is absent loudly.
• It does not translate. message_for returns the shape's own wording, so the API and the shape can never disagree about the same rule. The validator in use does not honour sh:resultMessage itself, and relying on an engine's leniency for the API's error text would be fragile anyway.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

ShapeUnavailable

Bases: Exception

The shape artifacts are not on disk, so nothing can be validated.

Source code in oekg/shape.py

class ShapeUnavailable(Exception):
    """The shape artifacts are not on disk, so nothing can be validated."""

constraint_message(property_iri)

The shape's own wording for the property shape constraining property_iri.

Used so a rejected pick is explained in the shape's words rather than in a second set written here, which could then disagree with it.

Source code in oekg/shape.py

def constraint_message(property_iri: str) -> Optional[str]:
    """The shape's own wording for the property shape constraining ``property_iri``.

    Used so a rejected pick is explained in the shape's words rather than in a
    second set written here, which could then disagree with it.
    """
    return _shape_enumerations()[1].get(property_iri)

enumeration(property_iri)

The IRIs a payload may pick for property_iri, per the shape's sh:in.

Empty for a property the shape does not constrain by enumeration -- which a caller must treat as "not enumerated", never as "nothing is allowed".

Source code in oekg/shape.py

def enumeration(property_iri: str) -> frozenset:
    """The IRIs a payload may pick for ``property_iri``, per the shape's sh:in.

    Empty for a property the shape does not constrain by enumeration -- which
    a caller must treat as "not enumerated", never as "nothing is allowed".
    """
    return _shape_enumerations()[0].get(property_iri, frozenset())

label_graph()

The rdfs:label subset the shape's targets need to satisfy it.

Source code in oekg/shape.py

def label_graph() -> Graph:
    """The rdfs:label subset the shape's targets need to satisfy it."""
    return _parsed(_fingerprint(Path(settings.OEKG_SHAPE_LABELS_PATH)))

labels_by_term()

The same subset as a lookup table: term IRI to its label literals.

The subset is flat -- one predicate, rdfs:label, on IRI subjects -- so a caller needing the labels of a handful of terms can ask for exactly those instead of merging all 2,054 triples to reach twenty. Read-only, because it is the cached artifact and not a copy of it.

The literals are a tuple rather than a single value on purpose: the artifact holds one label per term today, and a validator asking "how many labels does this term have" must see what is there rather than what the extraction promised.

Source code in oekg/shape.py

def labels_by_term() -> Mapping:
    """The same subset as a lookup table: term IRI to its label literals.

    The subset is flat -- one predicate, ``rdfs:label``, on IRI subjects -- so
    a caller needing the labels of a handful of terms can ask for exactly
    those instead of merging all 2,054 triples to reach twenty. Read-only,
    because it is the cached artifact and not a copy of it.

    The literals are a tuple rather than a single value on purpose: the
    artifact holds one label per term today, and a validator asking "how many
    labels does this term have" must see what is there rather than what the
    extraction promised.
    """
    return _labels_index(_fingerprint(Path(settings.OEKG_SHAPE_LABELS_PATH)))

message_for(source_shape)

The shape's own wording for a violation raised by source_shape.

Source code in oekg/shape.py

def message_for(source_shape) -> Optional[str]:
    """The shape's own wording for a violation raised by ``source_shape``."""
    if source_shape is None:
        return None
    message = shape_graph().value(source_shape, SH.resultMessage)
    return str(message) if message is not None else None

shape_graph()

The SHACL shape itself.

Source code in oekg/shape.py

def shape_graph() -> Graph:
    """The SHACL shape itself."""
    return _parsed(_fingerprint(Path(settings.OEKG_SHAPES_PATH)))

Versioning

A bundle's version: what it is, where it lives, and how a write guards on it.

Two writers on the same field is the failure no amount of shape validation catches -- a perfectly valid change silently destroying one made a second earlier -- and a programmatic API makes it likelier than the user interface does, because scripts retry, run in parallel and do not look at the screen first.

The version is a monotonic counter, not an opaque token, because two other parts of this API need it to be one: the two-step delete uses it as its confirmation, and the history uses it as the natural key of "which change produced this state".

The triple points at the bundle, not away from it. The bundle's shape is sh:closed with only rdf:type ignored, so <bundle> oekg:version 17 would make every bundle this API writes invalid. sh:closed constrains a focus node's outgoing properties, so the bundle as an object is untouched, and nothing in the shape targets the version node: it is untyped, and oekg:versionOf is not one of the predicates whose objects the shape validates. The guard therefore works without editing the shape -- and the version triples stay out of a read for free, because a read walks outward from the bundle and these point inward.

The version node's IRI is derived from the bundle's, so a compare-and-set can address it in the same request that tests it, with no lookup first.

Why a write token. SPARQL's DELETE/INSERT/WHERE guard applies only when its pattern matches, but the store answers 200 and changes nothing when it does not, so the write itself carries no signal. Reading the version back is not enough either: two writers guarding on version 17 both intend 18, so finding 18 afterwards does not tell a loser from a winner. Each write therefore stamps a token only its own writer could have chosen, and reads that back. It costs no extra round trip -- the token comes back with the version the write already has to read -- and it is the difference between reporting a conflict and reporting a success that did not happen.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

BundleVersion dataclass

The version a read saw, and the token the write before it left behind.

Both come from one query, because a write needs both: the number to guard on, and the token to clear so the next read-back stays unambiguous.

Source code in oekg/versioning.py

@dataclass(frozen=True)
class BundleVersion:
    """The version a read saw, and the token the write before it left behind.

    Both come from one query, because a write needs both: the number to guard
    on, and the token to clear so the next read-back stays unambiguous.
    """

    number: int
    token: Optional[str] = None

    @property
    def etag(self) -> str:
        return f'"{self.number}"'

guarded_operation(store, uid, version, token, delete=None, insert=None, condition='')

One update operation that applies delete/insert at version.

The version bump rides in the same operation as the change, so the two cannot come apart: one request is one transaction, and the guard is the request's own WHERE.

The guard is the version and the bundle's existence, plus whatever condition a caller adds. A condition the version cannot express has to be bound in here, because the version only moves when this bundle is written: a delete asserting that nothing outside still cites the nodes it is removing is answering a question about the rest of the graph, and no amount of versioning this bundle would notice that changing.

The cost of a second condition is that a failed guard no longer has one cause. That is why the refusal says "read it again and retry" rather than naming which half fired -- which is the right advice for either.

Source code in oekg/versioning.py

def guarded_operation(
    store: GraphStore,
    uid: str,
    version: BundleVersion,
    token: str,
    delete: Optional[Graph] = None,
    insert: Optional[Graph] = None,
    condition: str = "",
) -> str:
    """One update operation that applies ``delete``/``insert`` at ``version``.

    The version bump rides in the same operation as the change, so the two
    cannot come apart: one request is one transaction, and the guard is the
    request's own ``WHERE``.

    The guard is the version and the bundle's existence, plus whatever
    ``condition`` a caller adds. A condition the version cannot express has to
    be bound in here, because the version only moves when *this* bundle is
    written: a delete asserting that nothing outside still cites the nodes it
    is removing is answering a question about the rest of the graph, and no
    amount of versioning this bundle would notice that changing.

    The cost of a second condition is that a failed guard no longer has one
    cause. That is why the refusal says "read it again and retry" rather than
    naming which half fired -- which is the right advice for either.
    """
    node = version_iri(uid)
    # Copies: the version bookkeeping is added here, and a caller's own delta
    # graph should not come back carrying it.
    to_delete = _copy(delete)
    to_insert = _copy(insert)

    where = version_guard(uid, version)
    if version.number != UNVERSIONED:
        to_delete.add((node, VERSION, Literal(version.number)))
        if version.token is not None:
            # Left behind, the previous writer's token would accumulate one
            # triple per write and make the next read-back ambiguous the other
            # way round.
            to_delete.add((node, WRITE_TOKEN, Literal(version.token)))

    to_insert += version_triples(uid, version.number + 1, token)
    if condition:
        where = f"{where} {condition}"
    return store.guarded_modification(where, delete=to_delete, insert=to_insert)

mint_write_token()

A value no competing writer would choose, so a read-back can tell.

Source code in oekg/versioning.py

def mint_write_token() -> str:
    """A value no competing writer would choose, so a read-back can tell."""
    return str(uuid.uuid4())

read_version(store, uid)

The version a bundle is at. UNVERSIONED if it has no version node.

Source code in oekg/versioning.py

def read_version(store: GraphStore, uid: str) -> BundleVersion:
    """The version a bundle is at. ``UNVERSIONED`` if it has no version node."""
    node = version_iri(uid)
    rows = store.select(
        "SELECT ?number ?token WHERE { %s %s ?number "
        "OPTIONAL { %s %s ?token } }"
        % (node.n3(), VERSION.n3(), node.n3(), WRITE_TOKEN.n3())
    )
    if not rows:
        return BundleVersion(UNVERSIONED)
    return BundleVersion(int(rows[0]["number"]), rows[0].get("token"))

version_guard(uid, version)

The pattern that holds exactly while uid is a bundle at version.

Written once because two writes need it and they are not the same write: an ordinary change bumps the version inside this guard, and a whole-bundle delete removes the node the guard tests. Spelled out at both, the two would drift, and a guard that has drifted still answers 200.

The bundle's own existence is asserted alongside the version, and it is not redundant: the version node points AT the bundle, so a delete that removes only the bundle's outgoing triples -- which is exactly what the browser's delete does -- leaves the version node behind. On the version alone the guard would then match a bundle that is gone, and the write would resurrect it as untyped orphan triples.

Source code in oekg/versioning.py

def version_guard(uid: str, version: BundleVersion) -> str:
    """The pattern that holds exactly while ``uid`` is a bundle at ``version``.

    Written once because two writes need it and they are not the same write: an
    ordinary change bumps the version inside this guard, and a whole-bundle
    delete removes the node the guard tests. Spelled out at both, the two would
    drift, and a guard that has drifted still answers `200`.

    **The bundle's own existence is asserted alongside the version**, and it is
    not redundant: the version node points AT the bundle, so a delete that
    removes only the bundle's outgoing triples -- which is exactly what the
    browser's delete does -- leaves the version node behind. On the version
    alone the guard would then match a bundle that is gone, and the write would
    resurrect it as untyped orphan triples.
    """
    bundle = bundle_iri(uid)
    if version.number == UNVERSIONED:
        # Bootstrapping a bundle the user interface wrote. The guard is the
        # absence of a version node, so two first writes cannot both see it
        # missing and both apply.
        return "%s a %s . FILTER NOT EXISTS { ?node %s %s }" % (
            bundle.n3(),
            BUNDLE_CLASS.n3(),
            VERSION_OF.n3(),
            bundle.n3(),
        )
    return "%s a %s . %s %s %s ; %s %s ." % (
        bundle.n3(),
        BUNDLE_CLASS.n3(),
        version_iri(uid).n3(),
        VERSION_OF.n3(),
        bundle.n3(),
        VERSION.n3(),
        Literal(version.number).n3(),
    )

version_iri(uid)

The version node for a bundle -- derived, never looked up.

Source code in oekg/versioning.py

def version_iri(uid: str) -> URIRef:
    """The version node for a bundle -- derived, never looked up."""
    return OEKG[f"version/{uid}"]

version_node_triples(uid, version)

The version node as it currently stands -- what a delete has to remove.

Built from the version a read already holds rather than fetched again: the node has at most these three triples, and the read that produced version saw all of them.

Empty for an unversioned bundle, because there is no node yet. Deleting the triples anyway would be harmless and would also say, wrongly, that this API knows of a node it has never written.

Source code in oekg/versioning.py

def version_node_triples(uid: str, version: BundleVersion) -> Graph:
    """The version node as it currently stands -- what a delete has to remove.

    Built from the version a read already holds rather than fetched again: the
    node has at most these three triples, and the read that produced
    ``version`` saw all of them.

    Empty for an unversioned bundle, because there is no node yet. Deleting the
    triples anyway would be harmless and would also say, wrongly, that this API
    knows of a node it has never written.
    """
    if version.number == UNVERSIONED:
        return Graph()
    return version_triples(uid, version.number, version.token)

version_triples(uid, number, token=None)

The version node as triples, at number.

Source code in oekg/versioning.py

def version_triples(uid: str, number: int, token: Optional[str] = None) -> Graph:
    """The version node as triples, at ``number``."""
    triples = Graph()
    node = version_iri(uid)
    triples.add((node, VERSION_OF, bundle_iri(uid)))
    triples.add((node, VERSION, Literal(int(number))))
    if token is not None:
        triples.add((node, WRITE_TOKEN, Literal(token)))
    return triples

write_applied(store, uid, token)

Whether the write that stamped token is the one that applied.

The signal the guarded update cannot give. False means the bundle moved between the read the validation needed and the write -- or vanished entirely -- so nothing of this request was applied.

Source code in oekg/versioning.py

def write_applied(store: GraphStore, uid: str, token: str) -> bool:
    """Whether the write that stamped ``token`` is the one that applied.

    The signal the guarded update cannot give. ``False`` means the bundle moved
    between the read the validation needed and the write -- or vanished
    entirely -- so nothing of this request was applied.
    """
    return store.ask(
        "ASK { %s %s %s }"
        % (version_iri(uid).n3(), WRITE_TOKEN.n3(), Literal(token).n3())
    )

Transport to the graph store

The OEKG REST API's own transport to the graph store.

Deliberately not factsheet/oekg/connection.py. That module exposes the graph as an rdflib Graph over a SPARQLUpdateStore with autocommit=True, so every add() is its own committed transaction: a ~200-triple bundle costs ~200 requests and 30 s, and an abort halfway leaves half a bundle behind. It also parses the full ontology at import -- 1.3 GB resident and ~36 s, per process -- which no request path should pay for.

So the boundary is drawn one step earlier: keep rdflib for building a graph in memory, drop it as transport. Callers assemble triples in an rdflib.Graph (which is also what the validator will read), and this module ships them as one SPARQL request.

That one request is also one transaction, across ;-separated operations -- established by experiment against Fuseki 5.1.0 on TDB2, and re-proved by this app's own tests rather than taken on trust. It is why an update-in-place is expressible at all: a delete and an insert in one request either both apply or neither does.

Two safety properties are built in rather than left to callers:

• The store's own error bodies never escape. Fuseki answers a bad query with a parser dump that echoes the generated query back. It is logged, never raised.
• clear() refuses to touch the default graph. It exists for tests, and a helper that could DROP DEFAULT is one misconfigured endpoint away from erasing the production knowledge graph.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

GraphQueryRejected

Bases: GraphStoreError

The store refused a read. Its response body is logged, never carried.

Source code in oekg/graph_store.py

class GraphQueryRejected(GraphStoreError):
    """The store refused a read. Its response body is logged, never carried."""

    action = "query"

GraphStore dataclass

A query client and an update client, and no pretence of being a graph.

graph names the target: None is the default graph, which is what the platform reads and writes in production. Tests point it at a named graph of their own instead.

Source code in oekg/graph_store.py

@dataclass(frozen=True)
class GraphStore:
    """A query client and an update client, and no pretence of being a graph.

    ``graph`` names the target: ``None`` is the default graph, which is what
    the platform reads and writes in production. Tests point it at a named
    graph of their own instead.
    """

    query_url: str
    update_url: str
    graph: Optional[str] = None
    auth: Optional[tuple] = None
    timeout: float = DEFAULT_TIMEOUT_SECONDS
    session: requests.Session = field(
        default_factory=requests.Session, repr=False, compare=False
    )

    @classmethod
    def from_settings(cls, *, graph=USE_CONFIGURED_GRAPH) -> "GraphStore":
        """Build the store the platform is configured to talk to.

        ``graph`` defaults to ``settings.OEKG_GRAPH``, which is ``None`` -- the
        default graph, where the platform's bundles live. A test overrides that
        setting to work in a graph of its own without reaching into the views.
        Passing ``graph=None`` explicitly means the default graph and ignores
        the setting.
        """
        rdf = settings.RDF_DATABASES["knowledge"]
        if graph is USE_CONFIGURED_GRAPH:
            graph = getattr(settings, "OEKG_GRAPH", None)
        base = "http://{host}:{port}/{name}".format(
            host=rdf["host"], port=rdf["port"], name=rdf["name"]
        )
        # Credentials are sent whenever they are configured. The existing
        # connection module gates this on USE_DOCKER, which makes the transport
        # behave differently between environments for no stated reason.
        user, password = rdf.get("user"), rdf.get("password")
        return cls(
            query_url=f"{base}/query",
            update_url=f"{base}/update",
            graph=graph,
            auth=(user, password) if user and password else None,
        )

    # ------------------------------------------------------------------ reads

    def select(self, query: str) -> list:
        """Run a SELECT and return its bindings as plain dicts of strings."""
        payload = self._query(query, SELECT_RESULTS).json()
        return [
            {name: binding[name]["value"] for name in binding}
            for binding in payload["results"]["bindings"]
        ]

    def ask(self, query: str) -> bool:
        """Run an ASK."""
        return bool(self._query(query, SELECT_RESULTS).json()["boolean"])

    def construct(self, query: str) -> Graph:
        """Run a CONSTRUCT or DESCRIBE and return the triples as a graph."""
        response = self._query(query, CONSTRUCT_RESULTS)
        graph = Graph()
        graph.parse(data=response.text, format="turtle")
        return graph

    # ----------------------------------------------------------------- writes

    def update(self, *operations: str) -> None:
        """Send SPARQL update operations as **one** request, so one transaction.

        The operations are the caller's own SPARQL and are sent as written --
        **this method does not scope them to the target graph.** Use
        ``insert_data``, ``delete_data`` and ``guarded_modification`` to build
        scoped operations; reach for a hand-written one only where no builder
        exists yet, and scope it yourself. An unscoped ``INSERT DATA`` writes
        the default graph whatever this store targets.
        """
        if not operations:
            return
        self._post(
            self.update_url, {"update": " ;\n".join(operations)}, GraphUpdateRejected
        )

    def insert(self, triples: Graph) -> None:
        """Write every triple in ``triples`` in one request."""
        self.update(self.insert_data(triples))

    def insert_data(self, triples: Graph) -> str:
        """The INSERT DATA operation for ``triples``, scoped to the target graph.

        Returned rather than sent, so several operations can be combined into
        the single request that makes them atomic.
        """
        return f"INSERT DATA {{ {self._in_target_graph(triples)} }}"

    def delete_data(self, triples: Graph) -> str:
        """The DELETE DATA operation for ``triples``, scoped to the target graph."""
        return f"DELETE DATA {{ {self._in_target_graph(triples)} }}"

    def guarded_modification(
        self,
        where: str,
        delete: Optional[Graph] = None,
        insert: Optional[Graph] = None,
    ) -> str:
        """A DELETE/INSERT/WHERE operation, scoped to the target graph.

        ``where`` is the guard, and it is part of the write rather than a check
        in front of it: SPARQL applies the templates only if the pattern
        matches, so a compare-and-set cannot be overtaken between the test and
        the change. The price is that a guard that does not match is
        indistinguishable from one that does -- the store answers ``200`` and
        changes nothing either way -- so the caller has to read back for the
        signal. That is not a shortcoming of this method; it is what SPARQL
        update offers.

        Scoping is by ``WITH``, which makes the target graph the default for
        every unqualified pattern in all three clauses at once. Writing
        ``GRAPH`` blocks instead would put the same decision in three places,
        and forgetting one of them writes the default graph -- in production,
        the graph the platform serves.
        """
        clauses = []
        if delete is not None and len(delete):
            clauses.append(f"DELETE {{ {delete.serialize(format='nt')} }}")
        if insert is not None and len(insert):
            clauses.append(f"INSERT {{ {insert.serialize(format='nt')} }}")
        if not clauses:
            # A guard with nothing behind it would still be a valid request the
            # store answers 200 to, which is the one answer a caller must never
            # read as "the guard held".
            raise NothingToModifyError(
                "A guarded modification needs triples to delete or to insert."
            )
        prefix = f"WITH <{self.graph}>\n" if self.graph else ""
        return f"{prefix}{' '.join(clauses)} WHERE {{ {where} }}"

    def clear(self) -> None:
        """Remove this store's named graph. Refuses on the default graph.

        DROP rather than CLEAR: CLEAR empties a graph but leaves it in the
        store, so a persistent dataset would accumulate one empty graph per
        test per run.
        """
        if not self.graph:
            raise UnsafeClearError(
                "Refusing to clear the default graph: this store targets the "
                "graph the platform itself uses. Build a GraphStore with a "
                "named graph to clear it."
            )
        self.update(f"DROP SILENT GRAPH <{self.graph}>")

    # ---------------------------------------------------------------- probing

    def is_available(self) -> bool:
        """Whether this store can actually be read from and written to.

        Probed with a harmless no-op update rather than a read, because the two
        are not the same permission: Fuseki serves queries to anyone and answers
        updates with 401 unless credentials are valid. A read-only probe would
        report a store as usable and leave the write tests failing instead of
        skipping.

        Never raises -- the caller is deciding whether to skip.
        """
        try:
            self.session.post(
                self.update_url,
                data={"update": f"DROP SILENT GRAPH <{AVAILABILITY_PROBE_GRAPH}>"},
                auth=self.auth,
                timeout=AVAILABILITY_TIMEOUT_SECONDS,
            ).raise_for_status()
        except requests.RequestException:
            return False
        return True

    # --------------------------------------------------------------- internal

    def _in_target_graph(self, triples: Graph) -> str:
        body = triples.serialize(format="nt")
        if self.graph:
            return f"GRAPH <{self.graph}> {{ {body} }}"
        return body

    def _query(self, query: str, accept: str) -> requests.Response:
        parameters = {"query": query}
        if self.graph:
            # Scoping by protocol parameter rather than by rewriting the query,
            # so callers write plain SPARQL and the store decides where it runs.
            parameters["default-graph-uri"] = self.graph
        return self._post(self.query_url, parameters, GraphQueryRejected, accept=accept)

    def _post(
        self,
        url: str,
        data: dict,
        rejection: type,
        accept: Optional[str] = None,
    ) -> requests.Response:
        headers = {"Accept": accept} if accept else {}
        try:
            response = self.session.post(
                url, data=data, headers=headers, auth=self.auth, timeout=self.timeout
            )
        except requests.RequestException as error:
            raise GraphStoreUnavailable(
                f"The OEKG graph store at {url} could not be reached."
            ) from error

        if response.status_code >= 400:
            # The body is the store's, not ours: Fuseki's parser dump echoes the
            # generated query back, which would leak it to whoever sees the
            # error. Log it where operators can read it and raise without it.
            logger.error(
                "OEKG graph store refused a %s: HTTP %s from %s -- %s",
                rejection.action,
                response.status_code,
                url,
                response.text[:2000],
            )
            raise rejection(
                f"The OEKG graph store refused this {rejection.action} "
                f"(HTTP {response.status_code}). {rejection.detail}".strip()
            )
        return response

ask(query)

Run an ASK.

Source code in oekg/graph_store.py

def ask(self, query: str) -> bool:
    """Run an ASK."""
    return bool(self._query(query, SELECT_RESULTS).json()["boolean"])

clear()

Remove this store's named graph. Refuses on the default graph.

DROP rather than CLEAR: CLEAR empties a graph but leaves it in the store, so a persistent dataset would accumulate one empty graph per test per run.

Source code in oekg/graph_store.py

def clear(self) -> None:
    """Remove this store's named graph. Refuses on the default graph.

    DROP rather than CLEAR: CLEAR empties a graph but leaves it in the
    store, so a persistent dataset would accumulate one empty graph per
    test per run.
    """
    if not self.graph:
        raise UnsafeClearError(
            "Refusing to clear the default graph: this store targets the "
            "graph the platform itself uses. Build a GraphStore with a "
            "named graph to clear it."
        )
    self.update(f"DROP SILENT GRAPH <{self.graph}>")

construct(query)

Run a CONSTRUCT or DESCRIBE and return the triples as a graph.

Source code in oekg/graph_store.py

def construct(self, query: str) -> Graph:
    """Run a CONSTRUCT or DESCRIBE and return the triples as a graph."""
    response = self._query(query, CONSTRUCT_RESULTS)
    graph = Graph()
    graph.parse(data=response.text, format="turtle")
    return graph

delete_data(triples)

The DELETE DATA operation for triples, scoped to the target graph.

Source code in oekg/graph_store.py

def delete_data(self, triples: Graph) -> str:
    """The DELETE DATA operation for ``triples``, scoped to the target graph."""
    return f"DELETE DATA {{ {self._in_target_graph(triples)} }}"

from_settings(*, graph=USE_CONFIGURED_GRAPH) classmethod

Build the store the platform is configured to talk to.

graph defaults to settings.OEKG_GRAPH, which is None -- the default graph, where the platform's bundles live. A test overrides that setting to work in a graph of its own without reaching into the views. Passing graph=None explicitly means the default graph and ignores the setting.

Source code in oekg/graph_store.py

@classmethod
def from_settings(cls, *, graph=USE_CONFIGURED_GRAPH) -> "GraphStore":
    """Build the store the platform is configured to talk to.

    ``graph`` defaults to ``settings.OEKG_GRAPH``, which is ``None`` -- the
    default graph, where the platform's bundles live. A test overrides that
    setting to work in a graph of its own without reaching into the views.
    Passing ``graph=None`` explicitly means the default graph and ignores
    the setting.
    """
    rdf = settings.RDF_DATABASES["knowledge"]
    if graph is USE_CONFIGURED_GRAPH:
        graph = getattr(settings, "OEKG_GRAPH", None)
    base = "http://{host}:{port}/{name}".format(
        host=rdf["host"], port=rdf["port"], name=rdf["name"]
    )
    # Credentials are sent whenever they are configured. The existing
    # connection module gates this on USE_DOCKER, which makes the transport
    # behave differently between environments for no stated reason.
    user, password = rdf.get("user"), rdf.get("password")
    return cls(
        query_url=f"{base}/query",
        update_url=f"{base}/update",
        graph=graph,
        auth=(user, password) if user and password else None,
    )

guarded_modification(where, delete=None, insert=None)

A DELETE/INSERT/WHERE operation, scoped to the target graph.

where is the guard, and it is part of the write rather than a check in front of it: SPARQL applies the templates only if the pattern matches, so a compare-and-set cannot be overtaken between the test and the change. The price is that a guard that does not match is indistinguishable from one that does -- the store answers 200 and changes nothing either way -- so the caller has to read back for the signal. That is not a shortcoming of this method; it is what SPARQL update offers.

Scoping is by WITH, which makes the target graph the default for every unqualified pattern in all three clauses at once. Writing GRAPH blocks instead would put the same decision in three places, and forgetting one of them writes the default graph -- in production, the graph the platform serves.

Source code in oekg/graph_store.py

def guarded_modification(
    self,
    where: str,
    delete: Optional[Graph] = None,
    insert: Optional[Graph] = None,
) -> str:
    """A DELETE/INSERT/WHERE operation, scoped to the target graph.

    ``where`` is the guard, and it is part of the write rather than a check
    in front of it: SPARQL applies the templates only if the pattern
    matches, so a compare-and-set cannot be overtaken between the test and
    the change. The price is that a guard that does not match is
    indistinguishable from one that does -- the store answers ``200`` and
    changes nothing either way -- so the caller has to read back for the
    signal. That is not a shortcoming of this method; it is what SPARQL
    update offers.

    Scoping is by ``WITH``, which makes the target graph the default for
    every unqualified pattern in all three clauses at once. Writing
    ``GRAPH`` blocks instead would put the same decision in three places,
    and forgetting one of them writes the default graph -- in production,
    the graph the platform serves.
    """
    clauses = []
    if delete is not None and len(delete):
        clauses.append(f"DELETE {{ {delete.serialize(format='nt')} }}")
    if insert is not None and len(insert):
        clauses.append(f"INSERT {{ {insert.serialize(format='nt')} }}")
    if not clauses:
        # A guard with nothing behind it would still be a valid request the
        # store answers 200 to, which is the one answer a caller must never
        # read as "the guard held".
        raise NothingToModifyError(
            "A guarded modification needs triples to delete or to insert."
        )
    prefix = f"WITH <{self.graph}>\n" if self.graph else ""
    return f"{prefix}{' '.join(clauses)} WHERE {{ {where} }}"

insert(triples)

Write every triple in triples in one request.

Source code in oekg/graph_store.py

def insert(self, triples: Graph) -> None:
    """Write every triple in ``triples`` in one request."""
    self.update(self.insert_data(triples))

insert_data(triples)

The INSERT DATA operation for triples, scoped to the target graph.

Returned rather than sent, so several operations can be combined into the single request that makes them atomic.

Source code in oekg/graph_store.py

def insert_data(self, triples: Graph) -> str:
    """The INSERT DATA operation for ``triples``, scoped to the target graph.

    Returned rather than sent, so several operations can be combined into
    the single request that makes them atomic.
    """
    return f"INSERT DATA {{ {self._in_target_graph(triples)} }}"

is_available()

Whether this store can actually be read from and written to.

Probed with a harmless no-op update rather than a read, because the two are not the same permission: Fuseki serves queries to anyone and answers updates with 401 unless credentials are valid. A read-only probe would report a store as usable and leave the write tests failing instead of skipping.

Never raises -- the caller is deciding whether to skip.

Source code in oekg/graph_store.py

def is_available(self) -> bool:
    """Whether this store can actually be read from and written to.

    Probed with a harmless no-op update rather than a read, because the two
    are not the same permission: Fuseki serves queries to anyone and answers
    updates with 401 unless credentials are valid. A read-only probe would
    report a store as usable and leave the write tests failing instead of
    skipping.

    Never raises -- the caller is deciding whether to skip.
    """
    try:
        self.session.post(
            self.update_url,
            data={"update": f"DROP SILENT GRAPH <{AVAILABILITY_PROBE_GRAPH}>"},
            auth=self.auth,
            timeout=AVAILABILITY_TIMEOUT_SECONDS,
        ).raise_for_status()
    except requests.RequestException:
        return False
    return True

select(query)

Run a SELECT and return its bindings as plain dicts of strings.

Source code in oekg/graph_store.py

def select(self, query: str) -> list:
    """Run a SELECT and return its bindings as plain dicts of strings."""
    payload = self._query(query, SELECT_RESULTS).json()
    return [
        {name: binding[name]["value"] for name in binding}
        for binding in payload["results"]["bindings"]
    ]

update(*operations)

Send SPARQL update operations as one request, so one transaction.

The operations are the caller's own SPARQL and are sent as written -- this method does not scope them to the target graph. Use insert_data, delete_data and guarded_modification to build scoped operations; reach for a hand-written one only where no builder exists yet, and scope it yourself. An unscoped INSERT DATA writes the default graph whatever this store targets.

Source code in oekg/graph_store.py

def update(self, *operations: str) -> None:
    """Send SPARQL update operations as **one** request, so one transaction.

    The operations are the caller's own SPARQL and are sent as written --
    **this method does not scope them to the target graph.** Use
    ``insert_data``, ``delete_data`` and ``guarded_modification`` to build
    scoped operations; reach for a hand-written one only where no builder
    exists yet, and scope it yourself. An unscoped ``INSERT DATA`` writes
    the default graph whatever this store targets.
    """
    if not operations:
        return
    self._post(
        self.update_url, {"update": " ;\n".join(operations)}, GraphUpdateRejected
    )

GraphStoreError

Bases: Exception

Base class for every way talking to the graph store can fail.

action names what was refused and detail what the caller may assume about the graph's state afterwards. Both live on the class so the message is built from the failure itself rather than from an argument that has to agree with it.

Source code in oekg/graph_store.py

class GraphStoreError(Exception):
    """Base class for every way talking to the graph store can fail.

    ``action`` names what was refused and ``detail`` what the caller may assume
    about the graph's state afterwards. Both live on the class so the message
    is built from the failure itself rather than from an argument that has to
    agree with it.
    """

    action = "request"
    detail = ""

GraphStoreUnavailable

Bases: GraphStoreError

The store could not be reached at all -- DNS, refused, or timed out.

Source code in oekg/graph_store.py

class GraphStoreUnavailable(GraphStoreError):
    """The store could not be reached at all -- DNS, refused, or timed out."""

GraphUpdateRejected

Bases: GraphStoreError

The store refused a write. Its response body is logged, never carried.

Source code in oekg/graph_store.py

class GraphUpdateRejected(GraphStoreError):
    """The store refused a write. Its response body is logged, never carried."""

    action = "update"
    detail = "Nothing was changed: one request is one transaction."

NothingToModifyError

Bases: GraphStoreError

A guarded modification was asked for with no triples to change.

Source code in oekg/graph_store.py

class NothingToModifyError(GraphStoreError):
    """A guarded modification was asked for with no triples to change."""

UnsafeClearError

Bases: GraphStoreError

Refused to empty the default graph.

Source code in oekg/graph_store.py

class UnsafeClearError(GraphStoreError):
    """Refused to empty the default graph."""

Payloads and triples

One field table per resource drives both directions, so a field is declared once rather than written twice.

A scenario bundle and its parts, as triples and as payloads.

The field tables live here; the machinery that drives them lives in oekg.fields, because it is the same problem at every level and there are now three tables playing it: the bundle, the scenario factsheet and the study report. Each entry names the property the shape validates, so every field is traceable to the shape.

What is a part and what is a field is not a judgement call: the shape decides it. A thing carrying its own has-uuid is addressable and therefore a part with its own URL; a set of IRIs with labels stays a field on its parent. Scenario factsheets and study reports carry one, frameworks and models do not.

Two decisions worth stating here rather than leaving to be inferred:

• The server mints identifiers. No client-supplied identifier is accepted, so a pipeline cannot collide with, or overwrite, somebody else's bundle by choosing a value.
• Frameworks and models are minted per bundle. They are fields, not addressable resources, so nothing outside this bundle should reference them. The user interface mints them globally, which is why deleting one bundle today strips labels from every other bundle citing the same model.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

BundlePart dataclass

An addressable part of a bundle: what it is, and what it is made of.

One of these per resource the shape gives its own has-uuid. Everything that differs between a scenario factsheet and a study report is in here, which is what lets one implementation of the endpoints serve both -- and what makes adding the next such resource a table rather than a module.

Source code in oekg/bundles.py

@dataclass(frozen=True)
class BundlePart:
    """An addressable part of a bundle: what it is, and what it is made of.

    One of these per resource the shape gives its own has-uuid. Everything that
    differs between a scenario factsheet and a study report is in here, which is
    what lets one implementation of the endpoints serve both -- and what makes
    adding the next such resource a table rather than a module.
    """

    name: str  # how a refusal names it to a client
    node_class: URIRef
    fields: tuple
    mint_segment: str  # the IRI segment this API mints under
    payload_key: str  # the key it nests under on a bundle create
    detail_route: str  # the named URL of one of them
    sort_field: str  # the field a listing is ordered by
    # The key its payload nests dataset links under, or "" for a part that
    # carries none. A study report cites a publication, not data.
    nested_links: str = ""

build_bundle_graph(uid, payload, known_labels=None, address=None)

A whole bundle: its fields, its parts, and its scenarios' dataset links.

Nesting is accepted here and on the replace endpoint, and nowhere else on the write path: a bundle POST builds its scenarios and study reports with it, while a bundle PATCH cannot reach one. That asymmetry is what lets a pipeline create a whole bundle in one call without giving any call the power to drop its parts by omission.

address says where one nested dataset link points, given its payload. It is a callable rather than a URL because half of the answer comes from the router and the other half from the request the link arrived on -- and neither belongs in a module about triples.

Source code in oekg/bundles.py

def build_bundle_graph(
    uid: str, payload: dict, known_labels: dict = None, address=None
) -> Graph:
    """A whole bundle: its fields, its parts, and its scenarios' dataset links.

    Nesting is accepted here and on the replace endpoint, and nowhere else on
    the write path: a bundle `POST` builds its scenarios and study reports with
    it, while a bundle `PATCH` cannot reach one. That asymmetry is what lets a
    pipeline create a whole bundle in one call without giving any call the
    power to drop its parts by omission.

    ``address`` says where one nested dataset link points, given its payload.
    It is a callable rather than a URL because half of the answer comes from
    the router and the other half from the request the link arrived on -- and
    neither belongs in a module about triples.
    """
    graph = resource_triples(
        bundle_iri(uid), BUNDLE_CLASS, BUNDLE_FIELDS, payload, known_labels
    )
    for part in BUNDLE_PARTS:
        for nested in payload.get(part.payload_key) or []:
            pid = mint_identifier()
            graph += build_part_graph(part, bundle_iri(uid), pid, nested, known_labels)
            graph += build_nested_link_graphs(
                part, part_iri(part, pid), nested, address
            )
    return graph

build_nested_link_graphs(part, node, payload, address=None)

The dataset links nested inside one part's payload, if it takes any.

Only a scenario does. Asked of the part rather than of the key, so a part that never carries links needs no branch here and a payload that names them on a study report is refused by that serializer rather than ignored here.

Source code in oekg/bundles.py

def build_nested_link_graphs(
    part: BundlePart, node: URIRef, payload: dict, address=None
) -> Graph:
    """The dataset links nested inside one part's payload, if it takes any.

    Only a scenario does. Asked of the part rather than of the key, so a part
    that never carries links needs no branch here and a payload that names them
    on a study report is refused by that serializer rather than ignored here.
    """
    graph = Graph()
    for link in payload.get(part.nested_links) or []:
        graph += build_dataset_link_graph(
            node, mint_identifier(), link, address(link) if address else None
        )
    return graph

build_part_graph(part, bundle, pid, payload, known_labels=None, node=None)

One sub-resource, linked to its bundle and carrying its identity.

The uuid goes in twice on purpose: once as the literal the shape requires and the URL names, and once inside the minted IRI. The literal is the identity -- a part the user interface wrote has an IRI this API did not choose, and a lookup by literal finds it anyway.

node names an existing part this is rewriting, for the one caller that has one: a replace matches a part by the identifier its _meta carries, and that part may well live at an IRI this API did not mint.

Source code in oekg/bundles.py

def build_part_graph(
    part: BundlePart,
    bundle: URIRef,
    pid: str,
    payload: dict,
    known_labels: dict = None,
    node: Optional[URIRef] = None,
) -> Graph:
    """One sub-resource, linked to its bundle and carrying its identity.

    The uuid goes in twice on purpose: once as the literal the shape requires
    and the URL names, and once inside the minted IRI. The literal is the
    identity -- a part the user interface wrote has an IRI this API did not
    choose, and a lookup by literal finds it anyway.

    ``node`` names an existing part this is rewriting, for the one caller that
    has one: a replace matches a part by the identifier its `_meta` carries,
    and that part may well live at an IRI this API did not mint.
    """
    node = part_iri(part, pid) if node is None else node
    graph = resource_triples(node, part.node_class, part.fields, payload, known_labels)
    graph.add((bundle, HAS_PART, node))
    graph.add((node, HAS_UUID, Literal(pid)))
    return graph

bundle_delta(uid, payload, pre_state, known_labels=None)

resource_delta for a bundle.

Source code in oekg/bundles.py

def bundle_delta(
    uid: str, payload: dict, pre_state: Graph, known_labels: dict = None
) -> tuple:
    """``resource_delta`` for a bundle."""
    return resource_delta(
        bundle_iri(uid), BUNDLE_FIELDS, payload, pre_state, known_labels
    )

bundle_iri(uid)

The IRI a bundle lives at -- the same one the user interface reads.

Source code in oekg/bundles.py

def bundle_iri(uid: str) -> URIRef:
    """The IRI a bundle lives at -- the same one the user interface reads."""
    return OEKG[uid]

bundle_payload(graph, uid)

A bundle's own fields. Sub-resources are not in here.

They have their own URLs, so a bundle read names them rather than nesting them -- the same asymmetry as the write side, from the other direction. The view adds that naming; keeping it out of this function is what lets a read be sent straight back to POST, where scenarios and study_reports mean create these.

Source code in oekg/bundles.py

def bundle_payload(graph: Graph, uid: str) -> dict:
    """A bundle's own fields. **Sub-resources are not in here.**

    They have their own URLs, so a bundle read names them rather than nesting
    them -- the same asymmetry as the write side, from the other direction. The
    view adds that naming; keeping it out of this function is what lets a read
    be sent straight back to `POST`, where `scenarios` and `study_reports` mean
    *create these*.
    """
    return resource_payload(graph, bundle_iri(uid), BUNDLE_FIELDS)

bundle_referenced_iris(payload)

Existing node IRIs a whole-bundle payload points at, nesting included.

One field table knows one level, because only the caller knows which key holds what -- so the walk over the parts is here, where the parts are.

Source code in oekg/bundles.py

def bundle_referenced_iris(payload: dict) -> list:
    """Existing node IRIs a whole-bundle payload points at, nesting included.

    One field table knows one level, because only the caller knows which key
    holds what -- so the walk over the parts is here, where the parts are.
    """
    iris = referenced_node_iris(payload, BUNDLE_FIELDS)
    for part in BUNDLE_PARTS:
        for nested in payload.get(part.payload_key) or []:
            iris += referenced_node_iris(nested, part.fields)
    return iris

bundle_subgraph(graph, uid)

graph narrowed to the bundle, as deep as a read goes.

Applied to a patch's post-state, this is what makes "validate the post-state" mean the state that will actually read back: a node a patch unlinked is no longer part of the bundle, so it is no longer part of what the shape is asked about.

It walks to the same depth the read query does, because the two have to agree: the post-state validated must be the state a subsequent read returns, and a pruner that stopped shorter would hide a scenario's regions from the validator.

Source code in oekg/bundles.py

def bundle_subgraph(graph: Graph, uid: str) -> Graph:
    """``graph`` narrowed to the bundle, as deep as a read goes.

    Applied to a patch's post-state, this is what makes "validate the
    post-state" mean the state that will actually read back: a node a patch
    unlinked is no longer part of the bundle, so it is no longer part of what
    the shape is asked about.

    It walks to the same depth the read query does, because the two have to
    agree: the post-state validated must be the state a subsequent read
    returns, and a pruner that stopped shorter would hide a scenario's regions
    from the validator.
    """
    narrowed = Graph()
    reached = {bundle_iri(uid)}
    frontier = {bundle_iri(uid)}
    for _ in range(BUNDLE_DEPTH + 1):
        beyond = set()
        for subject in frontier:
            for predicate, obj in graph.predicate_objects(subject):
                narrowed.add((subject, predicate, obj))
                if isinstance(obj, URIRef) and obj not in reached:
                    reached.add(obj)
                    beyond.add(obj)
        frontier = beyond
    return narrowed

bundle_uid(iri)

The identifier inside a bundle's IRI -- the inverse of bundle_iri.

None for an IRI outside the namespace bundles live in, so a listing reports what it found rather than a segment chopped off the end of an address it does not recognise. The IRI itself is reported either way.

A uid this API did not mint can come back here: the user interface takes one from its client and concatenates it. Such a bundle is listed with the uid it has, and its own URL answers 404 -- which is the read side reporting an identity problem that predates it, not making one.

Source code in oekg/bundles.py

def bundle_uid(iri) -> Optional[str]:
    """The identifier inside a bundle's IRI -- the inverse of `bundle_iri`.

    ``None`` for an IRI outside the namespace bundles live in, so a listing
    reports what it found rather than a segment chopped off the end of an
    address it does not recognise. The IRI itself is reported either way.

    A uid this API did not mint can come back here: the user interface takes
    one from its client and concatenates it. Such a bundle is listed with the
    uid it has, and its own URL answers `404` -- which is the read side
    reporting an identity problem that predates it, not making one.
    """
    prefix = str(OEKG)
    text = str(iri)
    if not text.startswith(prefix):
        return None
    uid = text[len(prefix) :]
    return uid or None

find_part(graph, uid, part, pid)

The part of this bundle with identifier pid, if it has one.

By the has-uuid literal rather than by rebuilding the IRI, so a part written before this API existed is addressable too.

Source code in oekg/bundles.py

def find_part(graph: Graph, uid: str, part: BundlePart, pid: str) -> Optional[URIRef]:
    """The part of this bundle with identifier ``pid``, if it has one.

    By the has-uuid literal rather than by rebuilding the IRI, so a part
    written before this API existed is addressable too.
    """
    for node in part_nodes(graph, uid, part):
        if part_uid(graph, node) == pid:
            return node
    return None

part_iri(part, pid)

The IRI a sub-resource this API minted lives at.

Derived, but not the identity: the identity is the has-uuid literal, so a part the user interface wrote -- whose IRI this API did not choose -- is still reachable by the same identifier its URL carries.

Source code in oekg/bundles.py

def part_iri(part: BundlePart, pid: str) -> URIRef:
    """The IRI a sub-resource this API minted lives at.

    Derived, but **not** the identity: the identity is the has-uuid literal, so
    a part the user interface wrote -- whose IRI this API did not choose -- is
    still reachable by the same identifier its URL carries.
    """
    return OEKG[f"{part.mint_segment}/{pid}"]

part_nodes(graph, uid, part)

Every part of this kind hanging off this bundle, in the graph given.

Source code in oekg/bundles.py

def part_nodes(graph: Graph, uid: str, part: BundlePart) -> list:
    """Every part of this kind hanging off this bundle, in the graph given."""
    return [
        node
        for node in graph.objects(bundle_iri(uid), HAS_PART)
        if (node, RDF.type, part.node_class) in graph
    ]

part_payload(graph, node, part)

One sub-resource's fields.

Source code in oekg/bundles.py

def part_payload(graph: Graph, node: URIRef, part: BundlePart) -> dict:
    """One sub-resource's fields."""
    return resource_payload(graph, node, part.fields)

part_uid(graph, node)

The identifier a part node carries, as the shape requires it to.

Source code in oekg/bundles.py

def part_uid(graph: Graph, node: URIRef) -> Optional[str]:
    """The identifier a part node carries, as the shape requires it to."""
    value = graph.value(node, HAS_UUID)
    return None if value is None else str(value)

One field table per resource, driving triples in both directions.

A resource in this API -- a bundle, a scenario factsheet, a study report -- is a closed set of fields, some literal, some picked from one of the shape's own lists, some minted nodes. That is the same problem at every level, so the machinery is written against a subject and a table rather than against any one resource: give it the subject to hang triples off and the table saying what the fields are, and it can build a create, a patch's delta, and the read that sends either back.

Writing it once is what makes a read return exactly what a write accepts. Neither direction can drift from the other, because there is only one place that says what a field is.

Each table entry names the property the shape validates, so every field is traceable to the shape rather than to a convention somebody remembered.

The server mints identifiers, everywhere and without exception, so a pipeline cannot collide with -- or overwrite -- somebody else's resource by choosing a value. And minted nodes are typed: the shape's sh:class constraints require it, the user interface does not do it, and the live graph carries the violations that produced.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

field_named(fields, name)

The one entry in fields called name.

Source code in oekg/fields.py

def field_named(fields: tuple, name: str) -> ResourceField:
    """The one entry in ``fields`` called ``name``."""
    return _by_name(fields)[name]

field_triples(subject, field, value, known_labels=None)

The triples one field's value means, and nothing else.

A create needs every field's triples at once; a patch needs exactly the named field's, so that a field nobody mentioned is genuinely untouched rather than deleted and rewritten identically. Both ask here, so the two directions cannot come to disagree about what a field is.

Source code in oekg/fields.py

def field_triples(
    subject: URIRef, field: ResourceField, value, known_labels: dict = None
) -> Graph:
    """The triples one field's value means, and nothing else.

    A create needs every field's triples at once; a patch needs exactly the
    named field's, so that a field nobody mentioned is genuinely untouched
    rather than deleted and rewritten identically. Both ask here, so the two
    directions cannot come to disagree about what a field is.
    """
    known_labels = known_labels or {}
    graph = Graph()
    if field.kind == LITERAL:
        if value is not None and value != "":
            graph.add((subject, field.predicate, Literal(value)))
    elif field.kind == ENUM:
        for iri in value:
            graph.add((subject, field.predicate, URIRef(iri)))
    elif field.kind == DATES:
        for moment in value:
            graph.add((subject, field.predicate, Literal(moment)))
    elif field.kind == DATE:
        if value is not None:
            graph.add((subject, field.predicate, Literal(value)))
    elif field.kind == LINK:
        if value:
            # The URL is the node. Nothing else is written onto it: no shape
            # asks a reference for a label, and minting a second node to hold
            # one would leave the bundle pointing at something that is not the
            # document it cites.
            graph.add((subject, field.predicate, URIRef(value)))
            graph.add((URIRef(value), RDF.type, field.node_class))
    elif field.kind == NODE:
        for entry in value:
            iri = entry.get("iri")
            node = URIRef(iri) if iri else minted(field)
            graph.add((subject, field.predicate, node))
            graph.add((node, RDF.type, field.node_class))
            if iri and iri in known_labels:
                # An existing shared node is referenced, never rewritten. Its
                # label comes from the graph so the post-state is complete for
                # validation, and the write adds nothing to a node other
                # bundles depend on.
                graph.add((node, RDFS.label, Literal(known_labels[iri])))
            else:
                graph.add((node, RDFS.label, Literal(entry["label"])))
    elif field.kind == PART:
        for entry in value:
            node = minted(field)
            graph.add((subject, field.predicate, node))
            graph.add((node, RDF.type, field.node_class))
            graph.add((node, RDFS.label, Literal(entry["label"])))
            if entry.get("iri"):
                # has-iri is a STRING on these nodes, per the shape: it points
                # at a factsheet page, it is not the node's identity.
                graph.add((node, HAS_IRI, Literal(entry["iri"])))
    return graph

linked_field_triples(graph, subject, field)

The triples that currently attach field's values to subject.

What a patch of that field removes -- the links only. A referenced contact, organisation, funder or author is shared with other bundles, and a framework or model is on the same "unlink, never delete" footing: nothing here reaches into a node's own triples, so no patch can strip a label that another bundle is displaying.

The cost of that rule is an unlinked node left in the graph. It is unreachable from the bundle, so no read returns it and no shape is asked about it, and deciding which nodes may actually be removed is the typed containment walk's job -- a whole slice of its own, because the allowlist was already wrong once while it was being drafted. Unlinking fails safe; guessing does not.

Frameworks and models share the has-part predicate, so the type is what tells them apart. Deleting by predicate alone would take both.

Source code in oekg/fields.py

def linked_field_triples(graph: Graph, subject: URIRef, field: ResourceField) -> Graph:
    """The triples that currently attach ``field``'s values to ``subject``.

    What a patch of that field removes -- **the links only.** A referenced
    contact, organisation, funder or author is shared with other bundles, and a
    framework or model is on the same "unlink, never delete" footing: nothing
    here reaches into a node's own triples, so no patch can strip a label that
    another bundle is displaying.

    The cost of that rule is an unlinked node left in the graph. It is
    unreachable from the bundle, so no read returns it and no shape is asked
    about it, and deciding which nodes may actually be removed is the typed
    containment walk's job -- a whole slice of its own, because the allowlist
    was already wrong once while it was being drafted. Unlinking fails safe;
    guessing does not.

    Frameworks and models share the has-part predicate, so the type is what
    tells them apart. Deleting by predicate alone would take both.
    """
    triples = Graph()
    for obj in graph.objects(subject, field.predicate):
        if field.kind == PART and (obj, RDF.type, field.node_class) not in graph:
            continue
        triples.add((subject, field.predicate, obj))
    return triples

mint_identifier()

A new identifier for anything this API creates.

One function rather than one per resource: no client supplies an identifier anywhere in this API, and that rule is easier to keep true where there is a single place it is expressed.

Source code in oekg/fields.py

def mint_identifier() -> str:
    """A new identifier for anything this API creates.

    One function rather than one per resource: **no client supplies an
    identifier anywhere in this API**, and that rule is easier to keep true
    where there is a single place it is expressed.
    """
    return str(uuid.uuid4())

minted(field)

A new IRI for a node this API is creating, under the field's segment.

Source code in oekg/fields.py

def minted(field: ResourceField) -> URIRef:
    """A new IRI for a node this API is creating, under the field's segment."""
    return OEKG[f"{field.mint_segment}/{mint_identifier()}"]

referenced_node_iris(payload, fields)

Every existing node IRI this payload points at, for this table.

One table, one level. Nesting is the caller's business, because only the caller knows which key holds what -- a table that reached for a key by name would keep looking for it after being handed a table that has no such key.

Source code in oekg/fields.py

def referenced_node_iris(payload: dict, fields: tuple) -> list:
    """Every existing node IRI this payload points at, for **this** table.

    One table, one level. Nesting is the caller's business, because only the
    caller knows which key holds what -- a table that reached for a key by name
    would keep looking for it after being handed a table that has no such key.
    """
    iris = []
    for field in fields:
        if field.kind != NODE:
            continue
        for entry in payload.get(field.name) or []:
            if entry.get("iri"):
                iris.append(entry["iri"])
    return iris

resource_delta(subject, fields, payload, pre_state, known_labels=None)

What a patch of payload removes and what it adds. Nothing else.

Per named field, so a field the payload does not mention contributes to neither side and is genuinely untouched -- not deleted and rewritten identically, which would churn every minted node in the bundle.

A set-valued field that is named is replaced whole: its links go and the payload's take their place. Emptying such a field is therefore a real change, which the shape then judges -- an empty list on a field the shape requires is a rejection, never a silent wipe.

Source code in oekg/fields.py

def resource_delta(
    subject: URIRef,
    fields: tuple,
    payload: dict,
    pre_state: Graph,
    known_labels: dict = None,
) -> tuple:
    """What a patch of ``payload`` removes and what it adds. Nothing else.

    Per named field, so a field the payload does not mention contributes to
    neither side and is genuinely untouched -- not deleted and rewritten
    identically, which would churn every minted node in the bundle.

    A set-valued field that *is* named is replaced whole: its links go and the
    payload's take their place. Emptying such a field is therefore a real
    change, which the shape then judges -- an empty list on a field the shape
    requires is a rejection, never a silent wipe.
    """
    removed, added = Graph(), Graph()
    for name, value in payload.items():
        field = field_named(fields, name)
        removed += linked_field_triples(pre_state, subject, field)
        added += field_triples(subject, field, value, known_labels)
    return removed, added

resource_payload(graph, subject, fields)

Read one resource's triples back into exactly what a write would accept.

Source code in oekg/fields.py

def resource_payload(graph: Graph, subject: URIRef, fields: tuple) -> dict:
    """Read one resource's triples back into exactly what a write would accept."""
    payload = {}
    for field in fields:
        if field.kind == LITERAL:
            value = graph.value(subject, field.predicate)
            payload[field.name] = None if value is None else str(value)
        elif field.kind in (ENUM, DATES):
            payload[field.name] = sorted(
                str(obj) for obj in graph.objects(subject, field.predicate)
            )
        elif field.kind in (DATE, LINK):
            payload[field.name] = optional(graph.value(subject, field.predicate))
        elif field.kind == NODE:
            payload[field.name] = sorted(
                (
                    {
                        "iri": str(node),
                        "label": str(graph.value(node, RDFS.label) or ""),
                    }
                    for node in graph.objects(subject, field.predicate)
                ),
                key=lambda entry: entry["label"],
            )
        elif field.kind == PART:
            # Frameworks and models share has-part; their type tells them apart.
            payload[field.name] = sorted(
                (
                    {
                        "label": str(graph.value(node, RDFS.label) or ""),
                        "iri": optional(graph.value(node, HAS_IRI)),
                    }
                    for node in graph.objects(subject, field.predicate)
                    if (node, RDF.type, field.node_class) in graph
                ),
                key=lambda entry: entry["label"],
            )
    return payload

resource_triples(subject, node_class, fields, payload, known_labels=None)

Assemble the triples one resource's payload means.

The result is the post-state to validate and, unchanged, the thing to write: nothing is added between validating and writing.

known_labels maps an already-existing node IRI to the label it already carries. Shared IRIs stay shared -- that is the point of a graph -- so a referenced node keeps its own label and this never writes a second one onto it. Without that, an additive write would leave a contact other bundles cite carrying two labels, violating sh:maxCount 1 for all of them.

Source code in oekg/fields.py

def resource_triples(
    subject: URIRef,
    node_class: URIRef,
    fields: tuple,
    payload: dict,
    known_labels: dict = None,
) -> Graph:
    """Assemble the triples one resource's payload means.

    The result is the post-state to validate and, unchanged, the thing to
    write: nothing is added between validating and writing.

    ``known_labels`` maps an already-existing node IRI to the label it already
    carries. Shared IRIs stay shared -- that is the point of a graph -- so a
    referenced node keeps its own label and this never writes a second one onto
    it. Without that, an additive write would leave a contact other bundles
    cite carrying two labels, violating sh:maxCount 1 for all of them.
    """
    graph = Graph()
    graph.add((subject, RDF.type, node_class))
    for field in fields:
        if field.name in payload:
            graph += field_triples(subject, field, payload[field.name], known_labels)
    return graph

Reading a bundle out of the graph.

Bounded depth, not a transitive closure. A bundle reaches its scenarios in one hop, a scenario's study regions in two, and a region's reference in three -- which is the longest chain the shape allows, so three hops see a whole bundle and fewer do not. A closure would also be correct today and would stop needing thought; it would also be an unbounded walk on a public endpoint, reachable by anyone who can put a triple in the graph.

The version node is unreachable from here, because it points at the bundle rather than away from it. That is what keeps bookkeeping out of every payload without a single line of filtering.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

labels_of(store, iris, known=None)

The label each of these nodes already carries, for the ones that exist.

A write consults this so a referenced node keeps its own label and never gains a second one -- which would violate the shape for every bundle citing it, not just the one being written.

known is a graph already in hand; anything answered from it costs no query, so a request that has read its bundle asks the store only about nodes outside it.

Source code in oekg/reads.py

def labels_of(store: GraphStore, iris: list, known: Optional[Graph] = None) -> dict:
    """The label each of these nodes already carries, for the ones that exist.

    A write consults this so a referenced node keeps its own label and never
    gains a second one -- which would violate the shape for every bundle citing
    it, not just the one being written.

    ``known`` is a graph already in hand; anything answered from it costs no
    query, so a request that has read its bundle asks the store only about
    nodes outside it.
    """
    if not iris:
        return {}
    found, missing = {}, []
    for iri in iris:
        label = known.value(URIRef(iri), RDFS.label) if known is not None else None
        if label is None:
            missing.append(iri)
        else:
            found[iri] = str(label)
    if missing:
        values = " ".join(URIRef(iri).n3() for iri in missing)
        rows = store.select(
            "SELECT ?node ?label WHERE { VALUES ?node { %s } ?node %s ?label }"
            % (values, RDFS.label.n3())
        )
        found.update({row["node"]: row["label"] for row in rows})
    return found

read_bundle(store, uid)

A bundle and everything reachable from it, or None if there is none.

Source code in oekg/reads.py

def read_bundle(store: GraphStore, uid: str) -> Optional[Graph]:
    """A bundle and everything reachable from it, or ``None`` if there is none."""
    subgraph = store.construct(_query(uid))
    if (bundle_iri(uid), None, None) not in subgraph:
        return None
    return subgraph

History

What a write leaves behind, and how it reads back.

Two rules, and the second is the one that is easy to get wrong.

Every write records. Not "every write that destroys information" -- the judgement call is what makes an audit trail patchy, and "no entry" would then be ambiguous between created and never touched. So the rule is flat: the history is the log of writes.

The graph commits first, and a failed history write does not fail the request. The two stores cannot share a transaction. The data is the truth and the history is the note about it, so answering 500 for a write that succeeded would provoke exactly the retry that creates duplicates. The price is a possible gap in the audit trail, and it is named rather than hidden -- in the response, and in one structured log line -- because a phantom entry would be worse: it looks like truth.

What is stored is the changed triples, losslessly and untranslated. No field names, no serializer vocabulary, nothing that a later shape change could re-interpret. Field names are computed here at read time instead, from the same field table that builds the triples, so a rendering can never drift from what was stored -- it is derived from it on every read.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

changed_fields(uid, removed, added, resource_type=None, resource_uuid=None)

The diff, in the vocabulary a client writes in.

Computed on every read rather than stored, so it cannot drift from the triples it describes and a shape change never re-interprets an old row.

resource_type and resource_uuid are the class and identifier the write was about, both of which the entry already records. Together they say which vocabulary to render in and which subject in the diff is the thing being described: a change to a study report is named in the study report's field names, not in the bundle's, and the two tables share label while agreeing about almost nothing else. An entry naming no class is read as a bundle write, which is what every row written before sub-resources existed is.

A triple this cannot attribute to a field -- the type and label a minted contact brings with it, say -- is reported with a None field rather than dropped. Dropping it would make the summary look complete when it is not, which is the one thing a history must not do.

Every change names its predicate, attributed or not. Without it an unattributed entry would be a list of bare values with nothing saying what they were values of, which is only marginally better than dropping them.

Source code in oekg/history.py

def changed_fields(
    uid: str,
    removed: Graph,
    added: Graph,
    resource_type=None,
    resource_uuid: Optional[str] = None,
) -> list:
    """The diff, in the vocabulary a client writes in.

    Computed on every read rather than stored, so it cannot drift from the
    triples it describes and a shape change never re-interprets an old row.

    ``resource_type`` and ``resource_uuid`` are the class and identifier the
    write was about, both of which the entry already records. Together they say
    **which** vocabulary to render in and **which** subject in the diff is the
    thing being described: a change to a study report is named in the study
    report's field names, not in the bundle's, and the two tables share `label`
    while agreeing about almost nothing else. An entry naming no class is read
    as a bundle write, which is what every row written before sub-resources
    existed is.

    A triple this cannot attribute to a field -- the type and label a minted
    contact brings with it, say -- is reported with a ``None`` field rather
    than dropped. Dropping it would make the summary look complete when it is
    not, which is the one thing a history must not do.

    **Every change names its predicate**, attributed or not. Without it an
    unattributed entry would be a list of bare values with nothing saying what
    they were values of, which is only marginally better than dropping them.
    """
    node_class = URIRef(resource_type) if resource_type else BUNDLE_CLASS
    fields = FIELDS_BY_CLASS.get(node_class, ())
    subject_of = _subjects(uid, node_class, resource_uuid)
    changes = {}
    for side, graph in (("removed", removed), ("added", added)):
        for subject, predicate, obj in graph:
            key = (
                _field_name(subject, predicate, obj, graph, subject_of, fields),
                str(predicate),
            )
            entry = changes.setdefault(key, {"removed": [], "added": []})
            entry[side].append(str(obj))
    return [
        {
            "field": name,
            "predicate": predicate,
            "removed": sides["removed"],
            "added": sides["added"],
        }
        for (name, predicate), sides in sorted(
            changes.items(), key=lambda item: (item[0][0] or "", item[0][1])
        )
    ]

record_bundle_deletion(*, bundle_uid, acronym, actor, version_before)

Record that a bundle was deleted, and prune what it used to hold.

One event-only line, and no payload. Every other write stores the triples it changed, because a reader needs to know what the change was. A whole-bundle delete stores none, because the triples it removed are the whole bundle: a diff would make the ledger a copy of the thing that was deleted, and deleting would not delete. So the line records that it happened -- who, when, which identifier, which acronym, from which version -- and nothing about what was in it.

And the same reasoning reaches backwards. The bundle's earlier entries keep their structured columns, so the record of how it changed survives, but their payloads go: a history that kept them would let anyone rebuild a deleted bundle from the account of its own deletion. Legacy rows are pruned too, and they are the ones that matter most -- a browser write stored the bundle's whole state, not a diff.

Pruning and the event line are one transaction, because a pruned bundle with no line saying why reads as tampering, and a line with the payloads still under it has not deleted anything.

Never raises, for the same reason record_write does not: the graph write has already committed, and there is nothing left to undo.

Source code in oekg/history.py

def record_bundle_deletion(
    *,
    bundle_uid: str,
    acronym: str,
    actor,
    version_before: int,
) -> bool:
    """Record that a bundle was deleted, and prune what it used to hold.

    **One event-only line, and no payload.** Every other write stores the
    triples it changed, because a reader needs to know what the change was. A
    whole-bundle delete stores none, because the triples it removed are the
    whole bundle: a diff would make the ledger a copy of the thing that was
    deleted, and deleting would not delete. So the line records that it
    happened -- who, when, which identifier, which acronym, from which version
    -- and nothing about what was in it.

    **And the same reasoning reaches backwards.** The bundle's earlier entries
    keep their structured columns, so the record of *how it changed* survives,
    but their payloads go: a history that kept them would let anyone rebuild a
    deleted bundle from the account of its own deletion. Legacy rows are pruned
    too, and they are the ones that matter most -- a browser write stored the
    bundle's whole state, not a diff.

    Pruning and the event line are one transaction, because a pruned bundle
    with no line saying why reads as tampering, and a line with the payloads
    still under it has not deleted anything.

    **Never raises**, for the same reason `record_write` does not: the graph
    write has already committed, and there is nothing left to undo.
    """
    try:
        with transaction.atomic():
            entries = OEKG_Modifications.objects.filter(bundle_id=bundle_uid)
            entries.update(
                removed=None,
                added=None,
                old_state=EMPTY_LEGACY_PAYLOAD,
                new_state=EMPTY_LEGACY_PAYLOAD,
            )
            OEKG_Modifications.objects.create(
                bundle_id=bundle_uid,
                user=actor if getattr(actor, "is_authenticated", False) else None,
                verb=DELETE,
                resource_type=str(BUNDLE_CLASS),
                acronym=acronym,
                version_before=version_before,
                # There is no version after: the node counting them went with
                # the bundle. NULL says that; 0 would claim a version.
                version_after=None,
                old_state=EMPTY_LEGACY_PAYLOAD,
                new_state=EMPTY_LEGACY_PAYLOAD,
            )
    except Exception:
        logger.error(
            "oekg_history bundle=%s verb=%s user=%s acronym=%s version=%s->gone "
            "outcome=not_recorded",
            bundle_uid,
            DELETE,
            getattr(actor, "name", None) or "-",
            acronym,
            version_before,
            exc_info=True,
        )
        return False
    return True

record_write(*, bundle_uid, verb, actor, version_before, version_after, removed=None, added=None, resource_type=BUNDLE_CLASS, resource_uuid=None)

Record one write. Returns whether it was recorded.

Never raises, and that is load-bearing rather than polite: the caller's graph write has already committed, so there is nothing left to undo and nothing useful to tell the client beyond the fact that this note was lost. Anything raised from here would become a 500 for a write that succeeded -- the one outcome the whole ordering exists to avoid -- so the net is cast around every failure and not only around the database's.

Source code in oekg/history.py

def record_write(
    *,
    bundle_uid: str,
    verb: str,
    actor,
    version_before: int,
    version_after: int,
    removed: Optional[Graph] = None,
    added: Optional[Graph] = None,
    resource_type=BUNDLE_CLASS,
    resource_uuid: Optional[str] = None,
) -> bool:
    """Record one write. Returns whether it was recorded.

    **Never raises**, and that is load-bearing rather than polite: the caller's
    graph write has already committed, so there is nothing left to undo and
    nothing useful to tell the client beyond the fact that this note was lost.
    Anything raised from here would become a 500 for a write that succeeded --
    the one outcome the whole ordering exists to avoid -- so the net is cast
    around every failure and not only around the database's.
    """
    if verb not in VERBS:
        # Not a runtime guard against clients -- they cannot reach this. It
        # catches a caller inventing a verb the readers will not understand.
        raise ValueError(f"{verb!r} is not one of {', '.join(VERBS)}.")
    try:
        # The savepoint keeps a rejected insert from poisoning the surrounding
        # transaction. Without it a real database error would be caught here
        # and then raised again by the next query -- which would produce the
        # same 500 by a slower route.
        with transaction.atomic():
            OEKG_Modifications.objects.create(
                bundle_id=bundle_uid,
                user=actor if getattr(actor, "is_authenticated", False) else None,
                verb=verb,
                resource_type=str(resource_type) if resource_type else None,
                resource_uuid=resource_uuid,
                version_before=version_before,
                version_after=version_after,
                removed=_as_json(removed),
                added=_as_json(added),
                # Not null: the user interface's diff viewer reads these two
                # and would throw on one, taking its whole page down.
                old_state=EMPTY_LEGACY_PAYLOAD,
                new_state=EMPTY_LEGACY_PAYLOAD,
            )
    except Exception:
        # Deliberately every exception, not just the database's: serialising
        # the diff runs in here too, and an rdflib failure would otherwise
        # escape and undo the guarantee this function's contract makes.
        #
        # One structured line, the house format, so the gap is greppable. It is
        # the only place this becomes visible: there is no Django admin on this
        # platform to inspect the table through.
        logger.error(
            "oekg_history bundle=%s verb=%s user=%s version=%s->%s "
            "outcome=not_recorded",
            bundle_uid,
            verb,
            getattr(actor, "name", None) or "-",
            version_before,
            version_after,
            exc_info=True,
        )
        return False
    return True

Shared endpoint behaviour

What every endpoint of the OEKG API shares: its ceilings and its refusals.

Extracted when the second view module appeared and had to import five names from the first. That import is the signal: helpers every endpoint needs are not the bundle endpoint's property, and five more view modules are coming -- scenarios, study reports, dataset links, delete, replace. Left where they were, api_views would become a utility module by accident, and would change whenever any endpoint's conventions changed.

What belongs here: throttling, the refusals more than one endpoint gives, the existence check they all make first, and the reading of the one query parameter more than one endpoint answers. What does not: anything specific to one resource, which stays with that resource's view.

SPDX-FileCopyrightText: 2026 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-License-Identifier: AGPL-3.0-or-later

OekgAPIView

Bases: APIView

The base every OEKG endpoint shares: one ceiling, one way to refuse.

Handling Refused here rather than in each view means a new endpoint cannot forget to, and a refusal decided three calls deep still reaches the client as the response it was written as.

Source code in oekg/api_support.py

class OekgAPIView(APIView):
    """The base every OEKG endpoint shares: one ceiling, one way to refuse.

    Handling `Refused` here rather than in each view means a new endpoint
    cannot forget to, and a refusal decided three calls deep still reaches the
    client as the response it was written as.
    """

    throttle_classes = [ScenarioBundleThrottle, ScenarioBundleUserThrottle]

    # What `?expand=` may name at this endpoint. Empty means it may name
    # nothing, which is itself an answer -- see the listing.
    offers_expansions: tuple = ()

    def initial(self, request, *args, **kwargs):
        """Settle the request's own problems before the handler runs.

        `?expand=` is read **here**, not in each handler, and that is not
        tidiness: a handler that reads it after writing would create the
        resource, record the history, bump the version and *then* answer `400`
        -- a refusal that refused nothing. Reading it before dispatch makes
        "nothing was written" true of every `400` this API gives, and makes it
        true structurally rather than by each endpoint remembering.

        It is checked ahead of the resource's existence, unlike a payload's
        problems: an expansion nobody offers is wrong whether or not the
        bundle is there, and the store should not be read for a response that
        cannot be given.
        """
        super().initial(request, *args, **kwargs)
        self.expand = expansions(request, self.offers_expansions)

    def handle_exception(self, exc):
        if isinstance(exc, Refused):
            return exc.response
        return super().handle_exception(exc)

initial(request, *args, **kwargs)

Settle the request's own problems before the handler runs.

?expand= is read here, not in each handler, and that is not tidiness: a handler that reads it after writing would create the resource, record the history, bump the version and then answer 400 -- a refusal that refused nothing. Reading it before dispatch makes "nothing was written" true of every 400 this API gives, and makes it true structurally rather than by each endpoint remembering.

It is checked ahead of the resource's existence, unlike a payload's problems: an expansion nobody offers is wrong whether or not the bundle is there, and the store should not be read for a response that cannot be given.

Source code in oekg/api_support.py

def initial(self, request, *args, **kwargs):
    """Settle the request's own problems before the handler runs.

    `?expand=` is read **here**, not in each handler, and that is not
    tidiness: a handler that reads it after writing would create the
    resource, record the history, bump the version and *then* answer `400`
    -- a refusal that refused nothing. Reading it before dispatch makes
    "nothing was written" true of every `400` this API gives, and makes it
    true structurally rather than by each endpoint remembering.

    It is checked ahead of the resource's existence, unlike a payload's
    problems: an expansion nobody offers is wrong whether or not the
    bundle is there, and the store should not be read for a response that
    cannot be given.
    """
    super().initial(request, *args, **kwargs)
    self.expand = expansions(request, self.offers_expansions)

Refused

Bases: Exception

A refusal, raised where it is decided and returned as it stands.

It carries a whole Response rather than a detail, and OekgAPIView hands that back untouched. Both halves of that are deliberate:

• Raised, not returned. A helper that returned either a value or a refusal would make every call site test which it got, and one forgotten test is a refusal silently ignored.
• A Response, not an APIException. The obvious alternative is to subclass APIException and let the framework render it -- but the framework rewrites every scalar in an error body through ErrorDetail, so None comes out as the string "None" and a count as a string. This API's refusals carry structured data with nullable fields, so that would corrupt them. Measured, not assumed.

Source code in oekg/api_support.py

class Refused(Exception):
    """A refusal, raised where it is decided and returned as it stands.

    It carries a whole ``Response`` rather than a detail, and `OekgAPIView`
    hands that back untouched. Both halves of that are deliberate:

    - **Raised, not returned.** A helper that returned either a value or a
      refusal would make every call site test which it got, and one forgotten
      test is a refusal silently ignored.
    - **A Response, not an ``APIException``.** The obvious alternative is to
      subclass ``APIException`` and let the framework render it -- but the
      framework rewrites every scalar in an error body through
      ``ErrorDetail``, so ``None`` comes out as the string ``"None"`` and a
      count as a string. This API's refusals carry structured data with
      nullable fields, so that would corrupt them. Measured, not assumed.
    """

    def __init__(self, response: Response):
        super().__init__(getattr(response, "status_code", "refused"))
        self.response = response

ScenarioBundleThrottle

Bases: AnonRateThrottle

Reads are public, so the public endpoints need a ceiling of their own.

Source code in oekg/api_support.py

class ScenarioBundleThrottle(AnonRateThrottle):
    """Reads are public, so the public endpoints need a ceiling of their own."""

    scope = "oekg_bundles_anon"

bundle_exists(uid)

Whether there is a bundle at uid. Asked of the graph, not of a table.

Raises GraphStoreError rather than answering False when the store cannot be reached: "there is no such bundle" and "I could not find out" are different answers, and a caller has to be able to say 503 instead of 404.

An ASK rather than the read a GET does -- this asks whether the bundle is there, and pulling its whole subgraph across to answer that would make every caller pay for data it throws away.

Source code in oekg/api_support.py

def bundle_exists(uid: str) -> bool:
    """Whether there is a bundle at ``uid``. Asked of the graph, not of a table.

    Raises ``GraphStoreError`` rather than answering ``False`` when the store
    cannot be reached: "there is no such bundle" and "I could not find out" are
    different answers, and a caller has to be able to say 503 instead of 404.

    An ASK rather than the read a GET does -- this asks whether the bundle is
    there, and pulling its whole subgraph across to answer that would make
    every caller pay for data it throws away.
    """
    if not is_minted_identifier(uid):
        return False
    return bundle_in(GraphStore.from_settings(), uid)

bundle_in(store, uid)

The same question, asked of a store the caller already has.

Separate from bundle_exists because a delete has to ask it of its own store, inside its own request, and because the two must not drift: a write reads this back to find out whether its delete applied, and a query that had drifted would report a success that did not happen.

Source code in oekg/api_support.py

def bundle_in(store: GraphStore, uid: str) -> bool:
    """The same question, asked of a store the caller already has.

    Separate from `bundle_exists` because a delete has to ask it of **its own**
    store, inside its own request, and because the two must not drift: a write
    reads this back to find out whether its delete applied, and a query that
    had drifted would report a success that did not happen.
    """
    return store.ask("ASK { %s a %s }" % (bundle_iri(uid).n3(), BUNDLE_CLASS.n3()))

expansions(request, allowed)

The expansions this request asked for, refusing one nobody offers.

?expand= is the one query parameter more than one endpoint reads, so the rule for an unrecognised value lives here rather than being written again per endpoint with a slightly different wording. It is the same rule an unknown key gets on a write: refused, not ignored -- a client that misspells labels should be told, not handed the unresolved representation and left to work out why the labels are missing.

Comma-separated, so asking for two is asking once. Raises Refused.

Source code in oekg/api_support.py

def expansions(request, allowed: tuple) -> frozenset:
    """The expansions this request asked for, refusing one nobody offers.

    `?expand=` is the one query parameter more than one endpoint reads, so the
    rule for an unrecognised value lives here rather than being written again
    per endpoint with a slightly different wording. It is the same rule an
    unknown key gets on a write: refused, not ignored -- a client that
    misspells `labels` should be told, not handed the unresolved
    representation and left to work out why the labels are missing.

    Comma-separated, so asking for two is asking once. Raises ``Refused``.
    """
    asked = frozenset(
        value.strip()
        for value in (request.query_params.get("expand") or "").split(",")
        if value.strip()
    )
    unknown = sorted(asked - set(allowed))
    if unknown:
        raise Refused(
            Response(
                {
                    "detail": (
                        "%s is not something this endpoint can expand. It "
                        "offers %s."
                        % (
                            ", ".join(repr(value) for value in unknown),
                            ", ".join(repr(value) for value in allowed) or "nothing",
                        )
                    )
                },
                status=status.HTTP_400_BAD_REQUEST,
            )
        )
    return asked

is_minted_identifier(uid)

Whether uid could have come from this API.

Checked before a value reaches a query: an IRI-unsafe character makes rdflib refuse to build the IRI, which would surface as a 500 rather than the 404 it actually is. That refusal is also what keeps the query free of injection.

Source code in oekg/api_support.py

def is_minted_identifier(uid: str) -> bool:
    """Whether ``uid`` could have come from this API.

    Checked before a value reaches a query: an IRI-unsafe character makes
    rdflib refuse to build the IRI, which would surface as a 500 rather than
    the 404 it actually is. That refusal is also what keeps the query free of
    injection.
    """
    try:
        uuid.UUID(str(uid))
    except (ValueError, AttributeError, TypeError):
        return False
    return True

is_safe(request)

Whether this request only reads.

Asked by every endpoint that is public to read and closed to write, and asked here so all of them ask the same question. The safe methods are named and everything else is closed, rather than the other way round: a verb a later slice adds is then authenticated by default instead of public until somebody remembers. The price is that an unsupported verb answers 401 before it can answer 405, which is the cheaper of the two mistakes.

Source code in oekg/api_support.py

def is_safe(request) -> bool:
    """Whether this request only reads.

    Asked by every endpoint that is public to read and closed to write, and
    asked here so all of them ask the same question. The safe methods are
    named and everything else is closed, rather than the other way round: a
    verb a later slice adds is then authenticated by default instead of public
    until somebody remembers. The price is that an unsupported verb answers
    `401` before it can answer `405`, which is the cheaper of the two mistakes.
    """
    return request.method in ("GET", "HEAD", "OPTIONS")

API reference

For request/response shapes and endpoint URLs, see the Scenario Bundles feature page.

SPDX-FileCopyrightText: 2025 Adel Memariani https://github.com/adelmemariani © Otto-von-Guericke-Universität Magdeburg SPDX-FileCopyrightText: 2025 Adel Memariani https://github.com/adelmemariani © Otto-von-Guericke-Universität Magdeburg SPDX-FileCopyrightText: 2025 Adel Memariani https://github.com/adelmemariani © Otto-von-Guericke-Universität Magdeburg SPDX-FileCopyrightText: 2025 Adel Memariani https://github.com/adelmemariani © Otto-von-Guericke-Universität Magdeburg SPDX-FileCopyrightText: 2025 Adel Memariani https://github.com/adelmemariani © Otto-von-Guericke-Universität Magdeburg SPDX-FileCopyrightText: 2025 Bryan Lancien https://github.com/bmlancien © Reiner Lemoine Institut SPDX-FileCopyrightText: 2025 Christian Winger https://github.com/wingechr © Öko-Institut e.V. SPDX-FileCopyrightText: 2025 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-FileCopyrightText: 2025 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut SPDX-FileCopyrightText: 2025 Jonas Huber https://github.com/jh-RLI © Reiner Lemoine Institut

SPDX-License-Identifier: AGPL-3.0-or-later

add_entities_view(request, *args, **kwargs)

Add entities to OEKG. The minimum requirements for adding an entity are the type and label.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP GET request.	required
entity_type	str	The type(OEO class) of the entity.	required
entity_label	str	The label of the entity.	required
entity_iri	str	The IRI of the entity.	required

Source code in factsheet/views.py

@login_required
def add_entities_view(request, *args, **kwargs):
    """
    Add entities to OEKG. The minimum requirements for
    adding an entity are the type and label.

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        entity_type (str): The type(OEO class) of the entity.
        entity_label (str): The label of the entity.
        entity_iri (str): The IRI of the entity.
    """
    request_body = json.loads(request.body)
    entity_type = request_body["entity_type"]
    entity_label = request_body["entity_label"]
    entity_iri = request_body["entity_iri"]

    vocab = entity_type.split(".")[0]
    classId = entity_type.split(".")[1]
    prefix = ""
    if vocab == "OEO":
        prefix = "https://openenergyplatform.org/ontology/oeo/"
    if vocab == "OBO":
        prefix = "http://purl.obolibrary.org/obo/"

    entity_type_URI = URIRef(prefix + classId)

    entity_URI = URIRef("https://openenergyplatform.org/ontology/oekg/" + entity_iri)

    oekg.add((entity_URI, RDF.type, entity_type_URI))
    oekg.add((entity_URI, RDFS.label, Literal(entity_label)))

    response = JsonResponse(
        "A new entity added!", safe=False, content_type="application/json"
    )
    patch_response_headers(response, cache_timeout=1)
    return response

create_factsheet_view(request, *args, **kwargs)

Creates a scenario bundle based on user's data. Currently, the minimum requirement to create a bundle is the "acronym". The "acronym" must be unique. If the provided acronym already exists in the OEKG, then the function returns a "Duplicate error".

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP GET request.	required
uid	str	The unique ID for the bundle.	required
acronym	str	The acronym for the bundle.	required
abstract	str	The abstract for the bundle.	required
institution	list of objects	The institutions for the bundle.	required
funding_source	list of objects	The funding sources for the bundle.	required
contact_person	list of objects	The contact persons for the bundle.	required
sector_divisions	list of objects	The sector divisions for the bundle.	required
sectors	list of objects	The sectors for the bundle.	required
technologies	list of objects	The technologies for the bundle.	required
study_keywords	list of strings	The study keywords for the bundle.	required
scenarios	list of objects	The scenarios for the bundle.	required
models	list of strings	The models for the bundle.	required
frameworks	list of strings	The frameworks for the bundle.	required
publications	list[object]	A list of n publications related to the bundle study_name (str): The study name for the bundle. date_of_publication (str): The date of publication for the bundle. report_title (str): The report title for the bundle. report_doi (str): The report_doi for the bundle. place_of_publication (str): The place of publication for the bundle. link_to_study_report (str): The link to study for the bundle. authors (list of objects): The authors for the bundle.	required

Returns:

Type	Description
	"Factsheet saved" if successful, "Duplicate error" if the bundle's
	acronym exists.

Source code in factsheet/views.py

@require_POST
def create_factsheet_view(request, *args, **kwargs):
    """
    Creates a scenario bundle based on user's data. Currently, the minimum requirement
    to create a bundle is the "acronym". The "acronym" must be unique. If the provided
    acronym already exists in the OEKG, then the function returns a "Duplicate error".

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        uid (str): The unique ID for the bundle.
        acronym (str): The acronym for the bundle.
        abstract (str): The abstract for the bundle.
        institution (list of objects): The institutions for the bundle.
        funding_source (list of objects): The funding sources for the bundle.
        contact_person (list of objects): The contact persons for the bundle.
        sector_divisions (list of objects): The sector divisions for the bundle.
        sectors (list of objects): The sectors for the bundle.
        technologies (list of objects): The technologies for the bundle.
        study_keywords (list of strings): The study keywords for the bundle.
        scenarios (list of objects): The scenarios for the bundle.
        models (list of strings): The models for the bundle.
        frameworks (list of strings): The frameworks for the bundle.
        publications (list[object]): A list of n publications related to the bundle
            study_name (str): The study name for the bundle.
            date_of_publication (str): The date of publication for the bundle.
            report_title (str): The report title for the bundle.
            report_doi (str): The report_doi for the bundle.
            place_of_publication (str): The place of publication for the bundle.
            link_to_study_report (str): The link to study for the bundle.
            authors (list of objects): The authors for the bundle.

    Returns:
        "Factsheet saved" if successful, "Duplicate error" if the bundle's
        acronym exists.

    """

    if not request.user.is_authenticated:
        return HttpResponseForbidden("User not authenticated")

    request_body = json.loads(request.body)
    name = request_body["name"]  # noqa
    uid = request_body["uid"]
    acronym = request_body["acronym"]
    study_name = request_body["study_name"]
    abstract = request_body["abstract"]
    institution = request_body["institution"]
    funding_source = request_body["funding_source"]
    contact_person = request_body["contact_person"]
    sector_divisions = request_body["sector_divisions"]
    sectors = request_body["sectors"]
    # expanded_sectors = request_body["expanded_sectors"]  # noqa
    # energy_carriers = request_body['energy_carriers']
    # expanded_energy_carriers = request_body['expanded_energy_carriers']
    # energy_transformation_processes = request_body['energy_transformation_processes']
    # expanded_energy_transformation_processes = request_body['expanded_energy_transformation_processes'] # noqa
    technologies = request_body["technologies"]
    study_keywords = request_body["study_keywords"]
    scenarios = request_body["scenarios"]
    publications = request_body["publications"]
    models = request_body["models"]
    frameworks = request_body["frameworks"]

    Duplicate_study_factsheet = False

    for s, p, o in oekg.triples((None, RDF.type, OEO.OEO_00020227)):
        study_acronym = oekg.value(s, DC.acronym)
        if str(clean_name(acronym)) == str(study_acronym):
            Duplicate_study_factsheet = True

    if Duplicate_study_factsheet == True:  # noqa
        response = JsonResponse(
            "Factsheet exists", safe=False, content_type="application/json"
        )
        patch_response_headers(response, cache_timeout=1)
        return response
    else:
        bundle = Graph()

        study_URI = URIRef("https://openenergyplatform.org/ontology/oekg/" + uid)
        bundle.add((study_URI, RDF.type, OEO.OEO_00020227))

        if acronym != "":
            bundle.add((study_URI, DC.acronym, Literal(remove_non_printable(acronym))))
        if study_name != "":
            bundle.add(
                (
                    study_URI,
                    RDFS.label,
                    Literal(remove_non_printable(study_name)),
                )
            )
        if abstract != "":
            bundle.add(
                (study_URI, DC.abstract, Literal(remove_non_printable(abstract)))
            )

        _publications = json.loads(publications) if publications is not None else []
        for item in _publications:
            publications_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/publication/" + item["id"]
            )
            # OEO_00020012
            bundle.add((publications_URI, OEO.OEO_00390095, Literal(item["id"])))
            bundle.add((publications_URI, RDF.type, OEO.OEO_00020012))
            bundle.add((study_URI, OBO.BFO_0000051, publications_URI))
            if item["report_title"] != "":
                bundle.add(
                    (
                        publications_URI,
                        RDFS.label,
                        Literal(remove_non_printable(item["report_title"])),
                    )
                )

            _authors = item["authors"]
            for author in _authors:
                author_URI = URIRef(
                    "https://openenergyplatform.org/ontology/oekg/" + author["iri"]
                )
                bundle.add((author_URI, RDF.type, OEO.OEO_00000064))
                bundle.add((publications_URI, OEO.OEO_00000506, author_URI))

            if item["doi"] != "":
                bundle.add((publications_URI, OEO.OEO_00390098, Literal(item["doi"])))

            if (
                item["date_of_publication"] != "01-01-1900"
                and item["date_of_publication"] != ""
            ):
                bundle.add(
                    (
                        publications_URI,
                        OEO.OEO_00390096,
                        Literal(item["date_of_publication"], datatype=XSD.dateTime),
                    )
                )

            if item["link_to_study_report"] != "":
                bundle.add(
                    (URIRef(item["link_to_study_report"]), RDF.type, OEO.OEO_00000353)
                )
                bundle.add(
                    (
                        publications_URI,
                        OEO.OEO_00390078,
                        URIRef(item["link_to_study_report"]),
                    )
                )

            bundle.add((study_URI, OBO.BFO_0000051, publications_URI))

        _scenarios = json.loads(scenarios) if scenarios is not None else []
        for item in _scenarios:
            if item["acronym"] != "":
                scenario_URI = URIRef(
                    "https://openenergyplatform.org/ontology/oekg/scenario/"
                    + item["id"]
                )
                bundle.add((study_URI, OBO.BFO_0000051, scenario_URI))
                bundle.add(
                    (
                        scenario_URI,
                        DC.acronym,
                        Literal(remove_non_printable(item["acronym"])),
                    )
                )
                if item["name"] != "":
                    bundle.add(
                        (
                            scenario_URI,
                            RDFS.label,
                            Literal(remove_non_printable(item["name"])),
                        )
                    )
                    bundle.add((scenario_URI, RDF.type, OEO.OEO_00000365))
                if item["abstract"] != "":
                    bundle.add(
                        (
                            scenario_URI,
                            DC.abstract,
                            Literal(remove_non_printable(item["abstract"])),
                        )
                    )

                bundle.add((scenario_URI, OEO.OEO_00390095, Literal(item["id"])))

                if "regions" in item:
                    for region in item["regions"]:
                        region_URI = URIRef(region["iri"])
                        scenario_region = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/region/"
                            + region["iri"].rsplit("/", 1)[1]
                        )
                        bundle.add((scenario_region, RDF.type, OEO.OEO_00020032))
                        bundle.add(
                            (scenario_region, RDFS.label, Literal(region["name"]))
                        )
                        bundle.add((scenario_region, OEO.OEO_00390078, region_URI))
                        bundle.add((scenario_URI, OEO.OEO_00020220, scenario_region))

                if "interacting_regions" in item:
                    for interacting_region in item["interacting_regions"]:
                        interacting_region_URI = URIRef(interacting_region["iri"])
                        scenario_interacting_region = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/"
                            + interacting_region["iri"]
                        )

                        bundle.add(
                            (scenario_interacting_region, RDF.type, OEO.OEO_00020036)
                        )
                        bundle.add(
                            (
                                scenario_interacting_region,
                                RDFS.label,
                                Literal(interacting_region["name"]),
                            )
                        )
                        bundle.add(
                            (
                                scenario_interacting_region,
                                OEO.OEO_00390078,
                                interacting_region_URI,
                            )
                        )
                        bundle.add(
                            (
                                scenario_URI,
                                OEO.OEO_00020222,
                                scenario_interacting_region,
                            )
                        )

                if "scenario_years" in item:
                    for scenario_year in item["scenario_years"]:
                        bundle.add(
                            (
                                scenario_URI,
                                OEO.OEO_00020440,
                                Literal(scenario_year["name"], datatype=XSD.dateTime),
                            )
                        )

                if "descriptors" in item:
                    for descriptor in item["descriptors"]:
                        descriptor = URIRef(descriptor["class"])
                        bundle.add((scenario_URI, OEO.OEO_00390073, descriptor))

                # TODO: Jonas Huber: Update to avoid duplicated table name entries
                if "input_datasets" in item:
                    for input_dataset in item["input_datasets"]:
                        # TODO- set in settings
                        input_dataset_URI = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/input_datasets/"  # noqa
                            + input_dataset["key"]
                        )
                        bundle.add((input_dataset_URI, RDF.type, OEO.OEO_00030029))
                        bundle.add(
                            (
                                input_dataset_URI,
                                RDFS.label,
                                Literal(
                                    remove_non_printable(
                                        input_dataset["value"]["label"]
                                    )
                                ),
                            )
                        )
                        bundle.add(
                            (
                                input_dataset_URI,
                                OEO.OEO_00390094,
                                Literal(input_dataset["value"]["url"]),
                            )
                        )
                        bundle.add(
                            (
                                input_dataset_URI,
                                OEKG["has_id"],
                                Literal(input_dataset["idx"]),
                            )
                        )
                        bundle.add(
                            (
                                input_dataset_URI,
                                OEO.OEO_00390095,
                                Literal(input_dataset["key"]),
                            )
                        )
                        bundle.add((scenario_URI, OEO.OEO_00020437, input_dataset_URI))

                # TODO: Jonas Huber: Update to avoid duplicated table name entries
                if "output_datasets" in item:
                    for output_dataset in item["output_datasets"]:
                        output_dataset_URI = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/output_datasets/"  # noqa
                            + output_dataset["key"]
                        )
                        bundle.add((output_dataset_URI, RDF.type, OEO.OEO_00030030))
                        bundle.add(
                            (
                                output_dataset_URI,
                                RDFS.label,
                                Literal(
                                    remove_non_printable(
                                        output_dataset["value"]["label"]
                                    )
                                ),
                            )
                        )
                        bundle.add(
                            (
                                output_dataset_URI,
                                OEO.OEO_00390094,
                                Literal(output_dataset["value"]["url"]),
                            )
                        )
                        bundle.add(
                            (
                                output_dataset_URI,
                                OEKG["has_id"],
                                Literal(output_dataset["idx"]),
                            )
                        )
                        bundle.add(
                            (
                                output_dataset_URI,
                                OEO.OEO_00390095,
                                Literal(output_dataset["key"]),
                            )
                        )
                        bundle.add((scenario_URI, OEO.OEO_00020436, output_dataset_URI))

        institutions = json.loads(institution) if institution is not None else []
        for item in institutions:
            institution_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/" + item["iri"]
            )
            bundle.add((study_URI, OEO.OEO_00000510, institution_URI))

        funding_sources = (
            json.loads(funding_source) if funding_source is not None else []
        )
        for item in funding_sources:
            funding_source_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/" + item["iri"]
            )
            bundle.add((study_URI, OEO.OEO_00000509, funding_source_URI))
        contact_persons = (
            json.loads(contact_person) if contact_person is not None else []
        )
        for item in contact_persons:
            contact_person_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/" + item["iri"]
            )
            bundle.add((study_URI, OEO.OEO_00000508, contact_person_URI))

        _sector_divisions = (
            json.loads(sector_divisions) if sector_divisions is not None else []
        )
        for item in _sector_divisions:
            sector_divisions_URI = URIRef(item["class"])
            bundle.add((study_URI, PROP_BASED_ON_SECTOR_DIVISION, sector_divisions_URI))

        _sectors = json.loads(sectors) if sectors is not None else []
        for item in _sectors:
            sector_URI = URIRef(item["class"])
            bundle.add((study_URI, OEO.OEO_00020439, sector_URI))

        _technologies = json.loads(technologies) if technologies is not None else []
        for item in _technologies:
            technology_URI = URIRef(item["class"])
            bundle.add((study_URI, OEO.OEO_00020438, technology_URI))

        _models = json.loads(models) if models is not None else []
        for item in _models:
            model_id = item.get("id")
            if item.get("acronym"):
                model_acronym = item.get("acronym")
            else:
                model_acronym = item.get("name")
            model_url = item.get("url")

            if not model_id or not model_acronym or not model_url:
                continue  # Skip this item if any critical field is empty

            model_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/models/" + str(model_id)
            )
            bundle.add((model_URI, RDF.type, OEO.OEO_00000277))
            bundle.add(
                (
                    model_URI,
                    RDFS.label,
                    Literal(remove_non_printable(model_acronym)),
                )
            )
            bundle.add(
                (
                    model_URI,
                    OEO.OEO_00390094,
                    Literal(model_url),
                )
            )
            bundle.add((study_URI, OBO.BFO_0000051, model_URI))

        _frameworks = json.loads(frameworks) if frameworks is not None else []
        for item in _frameworks:
            framework_id = item.get("id")
            if item.get("acronym"):
                framework_acronym = item.get("acronym")
            else:
                framework_acronym = item.get("name")
            framework_url = item.get("url")

            if not framework_id or not framework_acronym or not framework_url:
                continue  # Skip this item if any critical field is empty

            framework_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/frameworks/"
                + str(framework_id)
            )

            bundle.add((framework_URI, RDF.type, OEO.OEO_00000172))

            if framework_acronym:
                bundle.add(
                    (
                        framework_URI,
                        RDFS.label,
                        Literal(remove_non_printable(framework_acronym)),
                    )
                )

            if framework_url:
                bundle.add(
                    (
                        framework_URI,
                        OEO.OEO_00390094,
                        Literal(framework_url),
                    )
                )

            bundle.add((study_URI, OBO.BFO_0000051, framework_URI))

        _study_keywords = (
            json.loads(study_keywords) if study_keywords is not None else []
        )
        # TODO:  Literal(keyword) should be URiRef
        if _study_keywords != []:
            for keyword in _study_keywords:
                bundle.add((study_URI, OEO.OEO_00390071, Literal(keyword)))

        for s, p, o in bundle.triples((None, None, None)):
            oekg.add((s, p, o))

        response = JsonResponse(
            "Factsheet saved", safe=False, content_type="application/json"
        )
        result = set_ownership(bundle_uid=uid, user=request.user)
        logger.info(result)
        patch_response_headers(response, cache_timeout=1)

        return response

dataset_entry(node)

THis function is part of the comparison backend. It helps to retieve data table ID's from the OEKG input / ouput data relations.

This function especially helps to retrieve the table name from a vaity of enties, some of them are referancing external sources which are not directly usable in our comparison backend.

:param node: A predicate from the OEKG related to a Node in the graph which is related to an input or output dataset.

:return: A dict with the dataset label, url and if possible the table name.

Source code in factsheet/views.py

def dataset_entry(node):
    """
    THis function is part of the comparison backend. It helps to retieve data table
    ID's from the OEKG input / ouput data relations.

    This function especially helps to retrieve the table name from a vaity of enties,
    some of them are referancing external sources which are not directly usable in our
    comparison backend.

    :param node: A predicate from the OEKG related to a Node in the graph which is
                related to an input or output dataset.

    :return: A dict with the dataset label, url and if possible the table name.
    """
    label = oekg.value(node, RDFS.label)
    url_term = oekg.value(node, OEO.OEO_00390094)  # may be None
    url_str = str(url_term) if url_term else None

    entry = {
        "label": str(label) if label else "",
        "url": url_str,
        **parse_dataset_iri(url_str),
    }

    # If your KG stores the table name explicitly, prefer it over parsing URLs:
    table_name = oekg.value(node, OEO.OEO_00000504)
    if table_name:
        entry["kind"] = "oep_table"
        entry["table_name"] = str(table_name)

    return entry

delete_factsheet_by_id_view(request, *args, **kwargs)

Removes a scenario bundle based on the provided ID.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP GET request.	required
id	str	The unique ID for the bundle.	required

Source code in factsheet/views.py

@only_if_user_is_owner_of_scenario_bundle
@login_required
def delete_factsheet_by_id_view(request, *args, **kwargs):
    """
    Removes a scenario bundle based on the provided ID.

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        id (str): The unique ID for the bundle.

    """
    id = request.GET.get("id")
    study_URI = URIRef("https://openenergyplatform.org/ontology/oekg/" + id)

    for s, p, o in oekg.triples((study_URI, OBO.BFO_0000051, None)):
        oekg.remove((o, None, None))
    oekg.remove((study_URI, None, None))

    response = JsonResponse(
        "factsheet removed!", safe=False, content_type="application/json"
    )
    patch_response_headers(response, cache_timeout=1)
    return response

get_entities_by_type_view(request, *args, **kwargs)

Returns all entities (from OEKG) with a certain type. The type should be supplied by the user.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP GET request.	required
entity_type	str	The type(OEO class) of the entity.	required

Source code in factsheet/views.py

def get_entities_by_type_view(request, *args, **kwargs):
    """
    Returns all entities (from OEKG) with a certain type.
    The type should be supplied by the user.

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        entity_type (str): The type(OEO class) of the entity.
    """
    entity_type = request.GET.get("entity_type")
    vocab = entity_type.split(".")[0]
    classId = entity_type.split(".")[1]
    prefix = ""
    if vocab == "OEO":
        prefix = "https://openenergyplatform.org/ontology/oeo/"
    if vocab == "OBO":
        prefix = "http://purl.obolibrary.org/obo/"

    entity_URI = URIRef(prefix + classId)

    entities = []
    for s, p, o in oekg.triples((None, RDF.type, entity_URI)):
        sl = oekg.value(s, RDFS.label)
        entities.append({"name": sl, "id": sl, "iri": str(s).split("/")[-1]})

    response = JsonResponse(entities, safe=False, content_type="application/json")
    patch_response_headers(response, cache_timeout=1)
    return response

populate_factsheets_elements_view(request, *args, **kwargs)

This function populates the elements required for creating or updating a factsheet. For example: Elements returned form this function populate dropdown elements which help the user to select sectors, technologies, scenario descriptors etc.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP GET request.	required

Returns:

Name	Type	Description
JsonResponse		A JSON response containing the elements for factsheet creation or update.

Source code in factsheet/views.py

def populate_factsheets_elements_view(request, *args, **kwargs):
    """
    This function populates the elements required for creating or updating a factsheet.
    For example: Elements returned form this function populate dropdown elements which
    help the user to select sectors, technologies, scenario descriptors etc.

    Args:
        request (HttpRequest): The incoming HTTP GET request.

    Returns:
        JsonResponse: A JSON response containing the elements for factsheet creation or
                    update.
    """
    scenario_class = oeo_owl.search_one(iri=OEO.OEO_00000364)
    scenario_subclasses = get_all_sub_classes(scenario_class)

    technology_class = oeo_owl.search_one(iri=OEO.OEO_00000407)
    technology_subclasses = get_all_sub_classes(technology_class)

    # energy_carrier_class = oeo_owl.search_one(iri="http://openenergy-platform.org/ontology/oeo/OEO_00020039") # noqa
    # energy_carriers = get_all_sub_classes(energy_carrier_class)

    # energy_transformation_process_class = oeo_owl.search_one(iri="http://openenergy-platform.org/ontology/oeo/OEO_00020003") # noqa
    # energy_transformation_processes = get_all_sub_classes(energy_transformation_process_class) # noqa

    sector_divisions_list, sectors_list = build_sector_dropdowns_from_oeo(oeo)
    elements = {}
    # elements['energy_carriers'] = [energy_carriers]
    # elements['energy_transformation_processes'] = [energy_transformation_processes]
    elements["sector_divisions"] = sector_divisions_list
    elements["sectors"] = sectors_list
    elements["scenario_descriptors"] = scenario_subclasses
    elements["technologies"] = technology_subclasses
    elements["study_descriptors"] = build_study_descriptors_from_oeo(oeo)

    # for s, p, o in oeo.triples(( None, RDFS.subClassOf, OEO.OEO_00020003 )):
    #     sl = oeo.value(s, RDFS.label)
    #     parent = {
    #         'value': str(sl),
    #         'label': sl,
    #         'class': s
    #     }
    #     children = []
    #     for s1, p, o in oeo.triples(( None, RDFS.subClassOf, s )):
    #         sl1 = oeo.value(s1, RDFS.label)
    #         children2 = []
    #         for s2, p, o in oeo.triples(( None, RDFS.subClassOf, s1 )):
    #             sl2 = oeo.value(s2, RDFS.label)
    #             children3 = []
    #             for s3, p, o in oeo.triples(( None, RDFS.subClassOf, s2 )):
    #                 sl3 = oeo.value(s3, RDFS.label)
    #                 children3.append({
    #                     'value': str(sl) + str(sl1) + str(sl2) + str(sl3),
    #                     'label': sl3,
    #                     'class': s3
    #                 })

    #             if children3 != []:
    #                 children2.append({
    #                     'value': str(sl) + str(sl1) + str(sl2),
    #                     'label': sl2,
    #                     'class': s2,
    #                     'children': children3
    #                 })
    #             else:
    #                 children2.append({
    #                     'value': str(sl) + str(sl1) + str(sl2),
    #                     'class': s2,
    #                     'label': sl2,
    #                 })

    #         if children2 != []:
    #             children.append({
    #             'value': str(sl) + str(sl1),
    #             'label': sl1,
    #             'class': s1,
    #             'children': children2
    #             })
    #         else:
    #             children.append({
    #             'value': str(sl) + str(sl1),
    #             'class': s1,
    #             'label': sl1
    #             })

    #     if children != []:
    #         parent['children'] = children

    #     energy_transformation_processes.append(parent)

    # energy_carriers = []
    # for s, p, o in oeo.triples(( None, RDFS.subClassOf, OEO.OEO_00020039 )):
    #     sl = oeo.value(s, RDFS.label)
    #     parent = {
    #         'value': str(sl),
    #         'label': sl,
    #         'class': s
    #     }
    #     children = []
    #     for s1, p, o in oeo.triples(( None, RDFS.subClassOf, s )):
    #         sl1 = oeo.value(s1, RDFS.label)
    #         children2 = []
    #         for s2, p, o in oeo.triples(( None, RDFS.subClassOf, s1 )):
    #             sl2 = oeo.value(s2, RDFS.label)
    #             children3 = []
    #             for s3, p, o in oeo.triples(( None, RDFS.subClassOf, s2 )):
    #                 sl3 = oeo.value(s3, RDFS.label)
    #                 children3.append({
    #                     'value': str(sl) + "^^" + str(sl1) + "^^" + str(sl2) + "^^" + str(sl3), # noqa
    #                     'label': sl3,
    #                     'class': s3
    #                 })

    #             if children3 != []:
    #                 children2.append({
    #                     'value': str(sl) + "^^" + str(sl1) + "^^" + str(sl2),
    #                     'label': sl2,
    #                     'class': s2,
    #                     'children': children3
    #                 })
    #             else:
    #                 children2.append({
    #                     'value': str(sl) + "^^" + str(sl1) + "^^" + str(sl2),
    #                     'label': sl2,
    #                 })

    #         if children2 != []:
    #             children.append({
    #             'value': str(sl) + "^^" + str(sl1),
    #             'label': sl1,
    #             'class': s1,
    #             'children': children2
    #             })
    #         else:
    #             children.append({
    #             'value': str(sl) + "^^" + str(sl1),
    #             'class': s1,
    #             'label': sl1
    #             })

    #     if children != []:
    #         parent['children'] = children

    #     energy_carriers.append(parent)

    response = JsonResponse(elements, safe=False, content_type="application/json")
    patch_response_headers(response, cache_timeout=1)

    return response

update_an_entity_view(request, *args, **kwargs)

Updates an entity in OEKG. The minimum requirements for updating an entity are the type, the old label, and the new label.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP GET request.	required
entity_type	str	The type(OEO class) of the entity.	required
entity_label	str	The label of the entity.	required
new_entity_label	str	The new label of the entity.	required
entity_id	str	The IRI of the entity.	required

Source code in factsheet/views.py

@login_required
def update_an_entity_view(request, *args, **kwargs):
    """
    Updates an entity in OEKG. The minimum requirements for
    updating an entity are the type, the old label, and the
    new label.

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        entity_type (str): The type(OEO class) of the entity.
        entity_label (str): The label of the entity.
        new_entity_label (str): The new label of the entity.
        entity_id (str): The IRI of the entity.
    """
    request_body = json.loads(request.body)
    entity_type = request_body["entity_type"]
    entity_label = request_body["entity_label"]
    new_entity_label = request_body["new_entity_label"]
    entity_id = request_body["entity_iri"]

    vocab = entity_type.split(".")[0]
    classId = entity_type.split(".")[1]
    prefix = ""
    if vocab == "OEO":
        prefix = "https://openenergyplatform.org/ontology/oeo/"
    if vocab == "OBO":
        prefix = "http://purl.obolibrary.org/obo/"

    entity_type_URI = URIRef(prefix + classId)  # noqa
    entity_IRI = URIRef("https://openenergyplatform.org/ontology/oekg/" + (entity_id))

    oekg.add((entity_IRI, RDFS.label, Literal(new_entity_label)))
    oekg.remove((entity_IRI, RDFS.label, Literal(entity_label)))

    response = JsonResponse(
        "entity updated!", safe=False, content_type="application/json"
    )
    patch_response_headers(response, cache_timeout=1)
    return response

update_factsheet_view(request, *args, **kwargs)

Updates a scenario bundle based on user's data.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The incoming HTTP GET request.	required
uid	str	The unique ID for the bundle.	required
acronym	str	The acronym for the bundle.	required
abstract	str	The abstract for the bundle.	required
institution	list of objects	The institutions for the bundle.	required
funding_source	list of objects	The funding sources for the bundle.	required
contact_person	list of objects	The contact persons for the bundle.	required
sector_divisions	list of objects	The sector divisions for the bundle.	required
sectors	list of objects	The sectors for the bundle.	required
technologies	list of objects	The technologies for the bundle.	required
study_keywords	list of strings	The study keywords for the bundle.	required
scenarios	list of objects	The scenarios for the bundle.	required
models	list of strings	The models for the bundle.	required
frameworks	list of strings	The frameworks for the bundle.	required
publications	list[object]	A list of n publications related to the bundle study_name (str): The study name for the bundle. date_of_publication (str): The date of publication for the bundle. report_title (str): The report title for the bundle. report_doi (str): The report_doi for the bundle. place_of_publication (str): The place of publication for the bundle. link_to_study_report (str): The link to study for the bundle. authors (list of objects): The authors for the bundle.	required

Source code in factsheet/views.py

@login_required
@only_if_user_is_owner_of_scenario_bundle
def update_factsheet_view(request, *args, **kwargs):
    """
    Updates a scenario bundle based on user's data.

    Args:
        request (HttpRequest): The incoming HTTP GET request.
        uid (str): The unique ID for the bundle.
        acronym (str): The acronym for the bundle.
        abstract (str): The abstract for the bundle.
        institution (list of objects): The institutions for the bundle.
        funding_source (list of objects): The funding sources for the bundle.
        contact_person (list of objects): The contact persons for the bundle.
        sector_divisions (list of objects): The sector divisions for the bundle.
        sectors (list of objects): The sectors for the bundle.
        technologies (list of objects): The technologies for the bundle.
        study_keywords (list of strings): The study keywords for the bundle.
        scenarios (list of objects): The scenarios for the bundle.
        models (list of strings): The models for the bundle.
        frameworks (list of strings): The frameworks for the bundle.
        publications (list[object]): A list of n publications related to the bundle
            study_name (str): The study name for the bundle.
            date_of_publication (str): The date of publication for the bundle.
            report_title (str): The report title for the bundle.
            report_doi (str): The report_doi for the bundle.
            place_of_publication (str): The place of publication for the bundle.
            link_to_study_report (str): The link to study for the bundle.
            authors (list of objects): The authors for the bundle.
    """
    request_body = json.loads(request.body)
    fsData = request_body["fsData"]
    # id = request_body["id"]  # noqa
    uid = request_body["uid"]
    # name = request_body["name"]  # noqa
    studyName = request_body["study_name"]
    acronym = request_body["acronym"]
    abstract = request_body["abstract"]
    institution = request_body["institution"]
    funding_source = request_body["funding_source"]
    contact_person = request_body["contact_person"]
    sector_divisions = request_body["sector_divisions"]
    sectors = request_body["sectors"]
    # expanded_sectors = request_body["expanded_sectors"]  # noqa
    # energy_carriers = request_body['energy_carriers']
    # expanded_energy_carriers = request_body['expanded_energy_carriers']
    # energy_transformation_processes = request_body['energy_transformation_processes']
    # expanded_energy_transformation_processes = request_body['expanded_energy_transformation_processes'] # noqa
    technologies = request_body["technologies"]
    study_keywords = request_body["study_keywords"]
    scenarios = request_body["scenarios"]
    models = request_body["models"]
    frameworks = request_body["frameworks"]
    publications = request_body["publications"]

    Duplicate_study_factsheet = False

    for s, p, o in oekg.triples((None, RDF.type, OEO.OEO_00020227)):
        study_acronym = oekg.value(s, DC.acronym)
        if str(clean_name(acronym)) == str(study_acronym) and str(
            clean_name(acronym)
        ) != str(fsData["acronym"]):
            Duplicate_study_factsheet = True

    if Duplicate_study_factsheet == True:  # noqa
        response = JsonResponse(
            "Factsheet exists", safe=False, content_type="application/json"
        )
        patch_response_headers(response, cache_timeout=1)
        return response

    if Duplicate_study_factsheet == False:  # noqa
        study_URI = URIRef("https://openenergyplatform.org/ontology/oekg/" + uid)

        old_bundle = Graph()
        for s, p, o in oekg.triples((study_URI, None, None)):
            old_bundle.add((s, p, o))
        for s, p, o in oekg.triples((study_URI, OBO.BFO_0000051, None)):
            for s1, p1, o1 in oekg.triples((o, None, None)):
                old_bundle.add((s1, p1, o1))

        new_bundle = Graph()
        new_bundle.add((study_URI, RDF.type, OEO.OEO_00020227))

        _publications = json.loads(publications) if publications is not None else []
        for item in _publications:
            publications_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/publication/" + item["id"]
            )

            new_bundle.add((publications_URI, OEO.OEO_00390095, Literal(item["id"])))
            new_bundle.add((publications_URI, RDF.type, OEO.OEO_00020012))
            new_bundle.add((study_URI, OBO.BFO_0000051, publications_URI))
            if item["report_title"] != "":
                new_bundle.add(
                    (publications_URI, RDFS.label, Literal(item["report_title"]))
                )

            _authors = item["authors"]

            if _authors:
                for author in _authors:
                    if author["name"]:
                        author_URI = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/"
                            + author["iri"]
                        )
                        new_bundle.add((author_URI, RDF.type, OEO.OEO_00000064))
                        new_bundle.add((publications_URI, OEO.OEO_00000506, author_URI))

            if item["doi"] != "":
                new_bundle.add(
                    (publications_URI, OEO.OEO_00390098, Literal(item["doi"]))
                )

            if (
                item["date_of_publication"] != "1900"
                and item["date_of_publication"] != ""
            ):
                new_bundle.add(
                    (
                        publications_URI,
                        OEO.OEO_00390096,
                        Literal(item["date_of_publication"], datatype=XSD.dateTime),
                    )
                )

            if item["link_to_study_report"] != "":
                new_bundle.add(
                    (URIRef(item["link_to_study_report"]), RDF.type, OEO.OEO_00000353)
                )
                new_bundle.add(
                    (
                        publications_URI,
                        OEO.OEO_00390078,
                        URIRef(item["link_to_study_report"]),
                    )
                )

            new_bundle.add((study_URI, OBO.BFO_0000051, publications_URI))

            # remove old date in publication
            # iterate to make sure it can only have unique publication date
            for _s, _p, _o in oekg.triples((publications_URI, OEO.OEO_00390096, None)):
                oekg.remove((_s, _p, _o))

        _scenarios = json.loads(scenarios) if scenarios is not None else []
        for item in _scenarios:
            if item["acronym"] != "":
                scenario_URI = URIRef(
                    "https://openenergyplatform.org/ontology/oekg/scenario/"
                    + item["id"]
                )

                for s, p, o in oekg.triples((scenario_URI, None, None)):
                    oekg.remove((o, p, o))

                new_bundle.add((scenario_URI, OEO.OEO_00390095, Literal(item["id"])))
                new_bundle.add((scenario_URI, RDF.type, OEO.OEO_00000365))
                # TODO Acronmy wird lavbel
                new_bundle.add(
                    (
                        scenario_URI,
                        DC.acronym,
                        Literal(remove_non_printable(item["acronym"])),
                    )
                )
                if item["name"] != "":
                    new_bundle.add(
                        (
                            scenario_URI,
                            RDFS.label,
                            Literal(remove_non_printable(item["name"])),
                        )
                    )
                if item["abstract"] != "" and item["abstract"] != None:  # noqa
                    new_bundle.add(
                        (
                            scenario_URI,
                            DC.abstract,
                            Literal(remove_non_printable(item["abstract"])),
                        )
                    )

                if "regions" in item:
                    for region in item["regions"]:
                        region_URI = URIRef(region["iri"])
                        scenario_region = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/region/"
                            + region["iri"].rsplit("/", 1)[1]
                        )
                        new_bundle.add((scenario_region, RDF.type, OEO.OEO_00020032))
                        new_bundle.add(
                            (
                                scenario_region,
                                RDFS.label,
                                Literal(region["name"]),
                            )
                        )
                        new_bundle.add(
                            (
                                scenario_region,
                                OEO.OEO_00390078,
                                region_URI,
                            )
                        )
                        new_bundle.add(
                            (scenario_URI, OEO.OEO_00020220, scenario_region)
                        )
                        new_bundle.add(
                            (scenario_URI, OEO.OEO_00390078, scenario_region)
                        )

                if "interacting_regions" in item:
                    for interacting_region in item["interacting_regions"]:
                        interacting_region_URI = URIRef(interacting_region["iri"])
                        scenario_interacting_region = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/"
                            + interacting_region["iri"]
                        )

                        new_bundle.add(
                            (scenario_interacting_region, RDF.type, OEO.OEO_00020036)
                        )
                        new_bundle.add(
                            (
                                scenario_interacting_region,
                                RDFS.label,
                                Literal(interacting_region["name"]),
                            )
                        )
                        new_bundle.add(
                            (
                                scenario_interacting_region,
                                OEO.OEO_00390078,
                                interacting_region_URI,
                            )
                        )

                        new_bundle.add(
                            (
                                scenario_URI,
                                OEO.OEO_00020222,
                                scenario_interacting_region,
                            )
                        )
                # TODO Value does not have datatype xsd:dateTime
                if "scenario_years" in item:
                    for scenario_year in item["scenario_years"]:
                        new_bundle.add(
                            (
                                scenario_URI,
                                OEO.OEO_00020440,
                                Literal(scenario_year["name"], datatype=XSD.dateTime),
                            )
                        )

                if "descriptors" in item:
                    for descriptor in item["descriptors"]:
                        descriptor = URIRef(descriptor["class"])
                        new_bundle.add((scenario_URI, OEO.OEO_00390073, descriptor))

                # TODO: Jonas Huber: Update to avoid duplicated table name entries
                # TODO: Predicate is not allowed (closed shape)
                if "input_datasets" in item:
                    for input_dataset in item["input_datasets"]:
                        input_dataset_URI = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/input_datasets/"  # noqa
                            + input_dataset["key"]
                        )

                        for s, p, o in oekg.triples((input_dataset_URI, None, None)):
                            oekg.remove((o, p, o))

                        new_bundle.add((input_dataset_URI, RDF.type, OEO.OEO_00030029))
                        new_bundle.add(
                            (
                                input_dataset_URI,
                                RDFS.label,
                                Literal(input_dataset["value"]["label"]),
                            )
                        )
                        new_bundle.add(
                            (
                                input_dataset_URI,
                                OEO.OEO_00390094,
                                Literal(input_dataset["value"]["url"]),
                            )
                        )
                        new_bundle.add(
                            (
                                input_dataset_URI,
                                OEKG["has_id"],
                                Literal(input_dataset["idx"]),
                            )
                        )
                        new_bundle.add(
                            (
                                input_dataset_URI,
                                OEO.OEO_00390095,
                                Literal(input_dataset["key"]),
                            )
                        )
                        new_bundle.add(
                            (scenario_URI, OEO.OEO_00020437, input_dataset_URI)
                        )

                # TODO: Jonas Huber: Update to avoid duplicated table name entries
                if "output_datasets" in item:
                    for output_dataset in item["output_datasets"]:
                        output_dataset_URI = URIRef(
                            "https://openenergyplatform.org/ontology/oekg/output_datasets/"  # noqa: E501
                            + output_dataset["key"]
                        )
                        new_bundle.add((output_dataset_URI, RDF.type, OEO.OEO_00030030))
                        new_bundle.add(
                            (
                                output_dataset_URI,
                                RDFS.label,
                                Literal(output_dataset["value"]["label"]),
                            )
                        )
                        new_bundle.add(
                            (
                                output_dataset_URI,
                                OEO.OEO_00390094,
                                Literal(output_dataset["value"]["url"]),
                            )
                        )
                        new_bundle.add(
                            (
                                output_dataset_URI,
                                OEKG["has_id"],
                                Literal(output_dataset["idx"]),
                            )
                        )
                        new_bundle.add(
                            (
                                output_dataset_URI,
                                OEO.OEO_00390095,
                                Literal(output_dataset["key"]),
                            )
                        )
                        new_bundle.add(
                            (scenario_URI, OEO.OEO_00020436, output_dataset_URI)
                        )

                new_bundle.add((study_URI, OBO.BFO_0000051, scenario_URI))

        if acronym != "":
            new_bundle.add(
                (study_URI, DC.acronym, Literal(remove_non_printable(acronym)))
            )

        new_bundle.add(
            (study_URI, RDFS.label, Literal(remove_non_printable(studyName)))
        )

        institutions = json.loads(institution) if institution is not None else []
        for item in institutions:
            institution_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/" + item["iri"]
            )
            new_bundle.add((study_URI, OEO.OEO_00000510, institution_URI))

        funding_sources = (
            json.loads(funding_source) if funding_source is not None else []
        )
        for item in funding_sources:
            funding_source_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/" + item["iri"]
            )
            new_bundle.add((study_URI, OEO.OEO_00000509, funding_source_URI))

        if abstract != "":
            new_bundle.add(
                (study_URI, DC.abstract, Literal(remove_non_printable(abstract)))
            )

        contact_persons = (
            json.loads(contact_person) if contact_person is not None else []
        )
        for item in contact_persons:
            contact_person_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/" + item["iri"]
            )
            new_bundle.add((study_URI, OEO.OEO_00000508, contact_person_URI))

        _sector_divisions = (
            json.loads(sector_divisions) if sector_divisions is not None else []
        )
        for item in _sector_divisions:
            sector_divisions_URI = URIRef(item["class"])
            new_bundle.add(
                (study_URI, PROP_BASED_ON_SECTOR_DIVISION, sector_divisions_URI)
            )

        _sectors = json.loads(sectors) if sectors is not None else []
        for item in _sectors:
            sector_URI = URIRef(item["class"])
            new_bundle.add((study_URI, OEO.OEO_00020439, sector_URI))

        _technologies = json.loads(technologies) if technologies is not None else []
        for item in _technologies:
            technology_URI = URIRef(item["class"])
            new_bundle.add((study_URI, OEO.OEO_00020438, technology_URI))

        _models = json.loads(models) if models is not None else []
        for item in _models:
            model_id = item.get("id")

            if item.get("acronym"):
                model_acronym = item.get("acronym")
            else:
                model_acronym = item.get("name")
            model_url = item.get("url")

            if not model_id or not model_acronym or not model_url:
                continue  # Skip this item if any critical field is empty

            model_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/models/" + str(model_id)
            )
            new_bundle.add((model_URI, RDF.type, OEO.OEO_00000277))

            new_bundle.add(
                (
                    model_URI,
                    RDFS.label,
                    Literal(remove_non_printable(model_acronym)),
                )
            )

            new_bundle.add(
                (
                    model_URI,
                    OEO.OEO_00390094,
                    Literal(model_url),
                )
            )

            new_bundle.add((study_URI, OBO.BFO_0000051, model_URI))

            # remove old labels
            # iterate to make sure only current selection is available
            for _s, _p, _o in oekg.triples((model_URI, RDFS.label, None)):
                oekg.remove((_s, _p, _o))

            # remove old iri´s
            # iterate to make sure only current selection is available
            for _s, _p, _o in oekg.triples((model_URI, OEO.OEO_00390094, None)):
                oekg.remove((_s, _p, _o))

        _frameworks = json.loads(frameworks) if frameworks is not None else []
        for item in _frameworks:
            framework_id = item.get("id")
            if item.get("acronym"):
                framework_acronym = item.get("acronym")
            else:
                framework_acronym = item.get("name")
            framework_url = item.get("url")

            if not framework_id or not framework_url:
                continue  # Skip this item if any critical field is empty

            framework_URI = URIRef(
                "https://openenergyplatform.org/ontology/oekg/frameworks/"
                + str(framework_id)
            )

            new_bundle.add((framework_URI, RDF.type, OEO.OEO_00000172))
            if framework_acronym:
                new_bundle.add(
                    (
                        framework_URI,
                        RDFS.label,
                        Literal(remove_non_printable(framework_acronym)),
                    )
                )
            if framework_url:
                new_bundle.add(
                    (
                        framework_URI,
                        OEO.OEO_00390094,
                        Literal(framework_url),
                    )
                )

            new_bundle.add((study_URI, OBO.BFO_0000051, framework_URI))

            # remove old labels
            # iterate to make sure only current selection is available
            for _s, _p, _o in oekg.triples((framework_URI, RDFS.label, None)):
                oekg.remove((_s, _p, _o))

            # remove old iri´s
            # iterate to make sure only current selection is available
            for _s, _p, _o in oekg.triples((framework_URI, OEO.OEO_00390094, None)):
                oekg.remove((_s, _p, _o))

        _study_keywords = (
            json.loads(study_keywords) if study_keywords is not None else []
        )
        for keyword in _study_keywords:
            new_bundle.add((study_URI, OEO.OEO_00390071, URIRef(keyword)))

        iso_old_bundle = to_isomorphic(old_bundle)
        iso_new_bundle = to_isomorphic(new_bundle)

        in_both, in_first, in_second = graph_diff(iso_old_bundle, iso_new_bundle)

        in_first_json = str(in_first.serialize(format="json-ld"))  # noqa
        in_second_json = str(in_second.serialize(format="json-ld"))  # noqa

        # remove old bundle from oekg
        for s, p, o in oekg.triples((study_URI, OBO.BFO_0000051, None)):
            oekg.remove((o, None, None))
        oekg.remove((study_URI, None, None))

        for s, p, o in oekg.triples((study_URI, OBO.BFO_0000051, None)):
            oekg.remove((o, None, None))
        oekg.remove((study_URI, None, None))

        # add updated bundle to oekg
        for s, p, o in new_bundle.triples((None, None, None)):
            oekg.add((s, p, o))

        OEKG_Modifications_instance = OEKG_Modifications(  # noqa
            bundle_id=uid,
            user=login_models.myuser.objects.filter(name=request.user).first(),
            old_state=in_first.serialize(format="json-ld"),
            new_state=in_second.serialize(format="json-ld"),
        )
        OEKG_Modifications_instance.save()

        response = JsonResponse(
            "factsheet updated!", safe=False, content_type="application/json"
        )
        patch_response_headers(response, cache_timeout=1)
        return response