triples_update: atomic asymmetric updates, blank-node identity, and pooled connections

CHANGELOG.md

@@ -0,0 +1,130 @@
+# Changelog
+
+## Unreleased
+
+### Changed
+
+- **`triples_update` now performs a general atomic update, not only equal-length
+  replacement** (`graph_db_interface/queries/triple_multi.py`).
+
+  Previously `triples_update` raised `InvalidInputError` unless `old_triples` and
+  `new_triples` had the same length, restricting it to 1:1 value replacement. The
+  method already builds a single `DELETE ... INSERT ... WHERE` SPARQL transaction,
+  which handles pure additions, pure removals, and unequal-size replacements
+  equally well, so the length restriction was removed. The existence pre-check is
+  now skipped when `old_triples` is empty (a pure insert has nothing to check).
+
+  This atomicity is required for constraint (SHACL) correctness: replacing a
+  cardinality-constrained property — e.g. a possession handover under a "possessed
+  by exactly one resource" shape — must apply the removal and the insertion in one
+  transaction so the intermediate (property-absent) state is never validated.
+
+  Enables `kapps_ogm.OGM.commit` to add/remove/replace properties through a single
+  atomic transaction. Requested by the `kapps_semantic_middleware` project.
+
+### Fixed
+
+- **Every request opened a new TCP connection and re-ran the TLS handshake**
+  (`graph_db_interface/graph_db.py`).
+
+  **Symptom:** every call through the client — query, update, graph import, repository
+  listing — cost far more than the work it asked the server to do. Against a remote HTTPS
+  endpoint (`https://graphdb.iam-mms.kit.edu`) a trivial `ASK { ?s ?p ?o }` measured
+  ~17 ms, of which only ~4 ms was the request itself. The cost was invisible per call and
+  compounded with volume: a single `kapps_semantic_middleware` integration test issues 129
+  requests, so it paid ~1.7 s in handshakes alone.
+
+  **Cause:** `_make_request` dispatched through the module-level `requests` helpers,
+  `getattr(requests, method)(...)`. Those are documented convenience wrappers that
+  construct a `requests.Session`, use it for exactly one request, and close it. Closing the
+  session discards its `urllib3` connection pool, so no connection was ever reused and each
+  call paid a fresh TCP connect plus a full TLS handshake. Confirmed by counting
+  `urllib3` connection creations: three sequential queries opened three
+  `HTTPSConnectionPool` connections.
+
+  **Fix:** the client now holds a persistent `requests.Session` **per thread** and issues
+  every request through it, so the connection pool survives across calls. Per thread rather
+  than one shared session because a `requests.Session` mutates its cookie jar on every
+  response and is not thread-safe — and one client is commonly driven from several threads
+  at once, e.g. a web framework serving requests while the embedding code queries. urllib3's
+  connection pools are thread-safe; the session wrapping them is not. Each thread pays one
+  handshake and then reuses its own pool. Added `GraphDB.close()`, which releases the pools
+  held for every thread that used the client. No change to any method signature or response
+  handling.
+
+  **Measured effect:** per-request cost against the remote endpoint dropped from ~17 ms to
+  ~4 ms. The `kapps_semantic_middleware` suite (146 tests, live GraphDB) went from 158 s to
+  60 s — a 2.65× speed-up with no test changes. Regression test:
+  `tests/test_connection_reuse.py`.
+
+- **`triples_update` replaced a retained blank node instead of updating it**
+  (`graph_db_interface/queries/triple_multi.py`).
+
+  **Symptom:** updating one property of an existing blank node was impossible. The node
+  was unlinked and a *different* node took its place, so every triple attached to the
+  original that the caller had not listed in `old_triples` became unreachable — silently,
+  with the call reporting success.
+
+  **Cause:** the DELETE and INSERT patterns were rendered from two separate blank-node →
+  variable maps (`old_bn_var_map` / `new_bn_var_map`). A `BNode` passed on both sides —
+  the caller's way of saying "same node, different property value" — therefore became
+  `?oldbn1` in the DELETE and `?newbn1` in the INSERT, and every new-side variable was
+  bound by `BIND(BNODE() AS ?newbnN)`, which mints a fresh store node. Captured query
+  before the fix, for a node whose only change is its value:
+
+  ```sparql
+  DELETE { <Belt1> :hasValue ?oldbn1 . ?oldbn1 :hasValue 12.1 }
+  INSERT { <Belt1> :hasValue ?newbn1 . ?newbn1 :hasValue 1.4 }
+  WHERE  { <Belt1> :hasValue ?oldbn1 . ?oldbn1 :hasValue 12.1 . BIND(BNODE() AS ?newbn1) }
+  ```
+
+  **Fix:** one shared map across both pattern sets, so a blank node present on both sides
+  renders as a single variable already bound by the WHERE clause. `BIND(BNODE() AS ?v)` is
+  now emitted only for blank nodes exclusive to `new_triples`. Behaviour for pure
+  additions, pure removals, unequal-length replacements and the IRI-only path is unchanged,
+  as is the single-transaction atomicity the SHACL note above depends on.
+
+  **Tests:** `tests/test_triple_update_query.py` asserts on the generated SPARQL and needs
+  no live repository, unlike the existing update tests. Verified red before the change and
+  green after: the two blank-node-identity tests fail on the previous implementation, the
+  two control tests pass on both.
+
+  Found while designing anonymous-node identity in `kapps_ogm` (see
+  `JaFeKl/graph_db_interface#6`); a correctness fix for this library independently of that
+  work. Reported by the `kapps_semantic_middleware` project.
+
+- **`IRI` rejected any URL containing a port** (`graph_db_interface/utils/iri.py`,
+  `IRI._sanitize`).
+
+  **Symptom:** constructing an `IRI` from a perfectly valid `http`/`https` URL that
+  includes a port raised `InvalidIRIError: ':' outside of supported schemes`:
+
+  ```python
+  IRI("http://127.0.0.1:8991")                      # raised
+  IRI("http://host:8991/workflows/x/execute")       # raised
+  ```
+
+  This also broke reads: `triples_get` / `query(convert_bindings=True)` convert an
+  `xsd:anyURI` literal to an `IRI` via `from_xsd_literal`, so any stored
+  `xsd:anyURI` value that was a ported URL made the read itself throw.
+
+  **Root cause:** for a full IRI beginning with a known scheme, the validator
+  rejected the whole string when `raw.count(":") > 1`. A scheme contributes one
+  colon (`http://`) and an authority port contributes a second (`:8991`), so every
+  ported URL tripped the check. The check's real intent is to catch a `:` used
+  where a `#` was meant (e.g. `http://…/owl:Class`).
+
+  **Fix:** after confirming the scheme, strip the scheme, split off the authority
+  (up to the first `/`, `#`, or `?`), remove a trailing `:<digits>` port from the
+  authority, and only then reject if a stray `:` remains in the authority or the
+  remainder. Ports are now accepted; the previously-rejected malformed forms
+  (`http://www.w3.org/2002/07:owl#`, `http://www.w3.org/2002/07/owl:Class`, etc.)
+  are still rejected. All existing `tests/test_iri.py` cases continue to pass.
+
+  **Reported/fixed by:** the `kapps_semantic_middleware` project, whose Service
+  ontology stores middleware endpoint URLs (`svc:address`, `svc:endpoint`) — always
+  `host:port` values — as `xsd:anyURI`, and which hit this on the first round-trip
+  of a registered endpoint. Per that project's dependency-and-bugfix policy
+  (`kapps_semantic_middleware/docs/adr/0001-dependency-wiring-and-bugfix-policy.md`),
+  genuine correctness bugs in sibling dependency repos are fixed directly in the
+  sibling with a detailed changelog entry — this is that entry.

graph_db_interface/graph_db.py

@@ -5,6 +5,7 @@
 from graph_db_interface.kafka.kafka_manager import KafkaManager
 import requests
 import logging
+import threading
 from requests import Response
 from graph_db_interface.utils.graph_db_credentials import GraphDBCredentials
 from graph_db_interface.utils.iri import IRI
@@ -44,6 +45,21 @@ def __init__(
         self._auth = None
         self._blank_ids = set()
 
+        # A persistent session per thread, so the connection pool survives across calls.
+        # The module-level `requests.get`/`requests.post` helpers build a throwaway session
+        # per call, which means a new TCP connect and a full TLS handshake for every SPARQL
+        # query — against a remote HTTPS endpoint that dominates the cost of a query, and a
+        # client typically issues many.
+        #
+        # Per thread rather than one shared session, because a `requests.Session` is not
+        # thread-safe: its cookie jar is mutated by every response. One client is commonly
+        # driven from several threads at once — a web framework serving requests while the
+        # embedding code queries — and urllib3's pools are thread-safe but the session
+        # around them is not. Each thread pays one handshake and then reuses its own pool.
+        self._thread_local = threading.local()
+        self._sessions: list[requests.Session] = []
+        self._sessions_lock = threading.Lock()
+
         if use_gdb_token:
             self._auth = self._get_authentication_token(
                 self._credentials.username, self._credentials.password
@@ -243,13 +259,32 @@ def _make_request(
         if self._auth is not None:
             headers["Authorization"] = self._auth
 
-        return getattr(requests, method)(
+        return self._session.request(
+            method,
             f"{self._credentials.base_url}/{endpoint}",
             headers=headers,
             timeout=timeout,
             **kwargs,
         )
 
+    @property
+    def _session(self) -> requests.Session:
+        """This thread's persistent session, created on first use."""
+        session = getattr(self._thread_local, "session", None)
+        if session is None:
+            session = requests.Session()
+            self._thread_local.session = session
+            with self._sessions_lock:
+                self._sessions.append(session)
+        return session
+
+    def close(self) -> None:
+        """Release the HTTP connection pools held for every thread that used this client."""
+        with self._sessions_lock:
+            sessions, self._sessions = self._sessions, []
+        for session in sessions:
+            session.close()
+
     def _get_authentication_token(
         self,
         username: str,

graph_db_interface/queries/triple_multi.py

@@ -333,9 +333,17 @@ def triples_update(
     """
     Update multiple RDF triples in the triplestore.
 
+    The removal of `old_triples` and the insertion of `new_triples` are applied in a
+    single atomic `DELETE ... INSERT ... WHERE` SPARQL transaction. The two lists need
+    not be the same length: this supports pure additions, pure removals, and general
+    replacements. Atomicity matters for constraint (e.g. SHACL) correctness — replacing
+    a cardinality-constrained property (such as a possession handover under a
+    "possessed by exactly one resource" shape) must never expose the intermediate state
+    in which the property is momentarily absent.
+
     Args:
-        old_triples (TriplesLike): Triples to be replaced.
-        new_triples (TriplesLike): Replacement triples (same length as `old_triples`).
+        old_triples (TriplesLike): Triples to remove.
+        new_triples (TriplesLike): Triples to insert. Need not match the length of `old_triples`.
         check_exist (Optional[bool]): If True, abort when any old triple does not exist. Defaults to True.
         named_graph (Optional[GraphNameLike]): Override the client's default named graph.
 
@@ -347,14 +355,15 @@ def triples_update(
     if not old_triples and not new_triples:
         return True
 
-    if len(old_triples) != len(new_triples):
-        raise InvalidInputError("Old and new triples lists must have the same length.")
-
     validated_old_triples = [utils.sanitize_triple(triple) for triple in old_triples]
     validated_new_triples = [utils.sanitize_triple(triple) for triple in new_triples]
 
-    if check_exist and not self.all_triple_exists(
-        triples=validated_old_triples, named_graph=named_graph
+    if (
+        check_exist
+        and validated_old_triples
+        and not self.all_triple_exists(
+            triples=validated_old_triples, named_graph=named_graph
+        )
     ):
         self.logger.warning(
             "At least one of the triples to update does not exist in the graph."
@@ -391,13 +400,23 @@ def _build_patterns(
             patterns.append(f"{subj_str} {pred_str} {obj_str} .")
         return patterns
 
-    old_bn_var_map: Dict[BNode, str] = {}
-    new_bn_var_map: Dict[BNode, str] = {}
-
-    old_delete_patterns = _build_patterns(
-        validated_old_triples, old_bn_var_map, "oldbn"
-    )
-    insert_patterns = _build_patterns(validated_new_triples, new_bn_var_map, "newbn")
+    # One map across both sides: a BNode occurring in old_triples and in new_triples is
+    # the *same* node, so it must render as the same SPARQL variable. Two maps gave it
+    # two variables, and the new-side one was then bound by BIND(BNODE()) below — which
+    # minted a replacement node and orphaned every triple the caller had not listed in
+    # old_triples. That made it impossible to update one property of an existing blank
+    # node while keeping the node.
+    bn_var_map: Dict[BNode, str] = {}
+
+    old_delete_patterns = _build_patterns(validated_old_triples, bn_var_map, "bn")
+    bn_bound_by_where = set(bn_var_map)
+    insert_patterns = _build_patterns(validated_new_triples, bn_var_map, "bn")
+
+    # Only blank nodes exclusive to the new side need minting; the rest are already
+    # bound by the WHERE clause.
+    new_only_bn_vars = [
+        var for bnode, var in bn_var_map.items() if bnode not in bn_bound_by_where
+    ]
 
     def _format_block(patterns: List[str]) -> str:
         if not patterns:
@@ -411,9 +430,9 @@ def _format_block(patterns: List[str]) -> str:
     where_block_parts: List[str] = []
     if where_patterns:
         where_block_parts.append(_format_block(where_patterns))
-    if new_bn_var_map:
+    if new_only_bn_vars:
         where_block_parts.extend(
-            f"  BIND(BNODE() AS {var})" for var in new_bn_var_map.values()
+            f"  BIND(BNODE() AS {var})" for var in new_only_bn_vars
         )
     where_block = "\n".join(where_block_parts)
 

graph_db_interface/utils/iri.py

@@ -304,7 +304,24 @@ def _sanitize(
 
         # Full IRI
         if any(raw.startswith(scheme) for scheme in IRI.SCHEMES):
-            if raw.count(":") > 1:
+            # Reject a ':' used where a '#' was intended (e.g. ".../owl:Class"),
+            # but allow a legitimate authority port (e.g. "http://host:8991/path").
+            # The scheme's own ':' (in "://") is stripped first; then a trailing
+            # ':<digits>' port in the authority is removed before counting any
+            # remaining stray colons.
+            matched_scheme = next(s for s in IRI.SCHEMES if raw.startswith(s))
+            after_scheme = raw[len(matched_scheme):]
+            sep_index = len(after_scheme)
+            for sep in ("/", "#", "?"):
+                i = after_scheme.find(sep)
+                if i != -1:
+                    sep_index = min(sep_index, i)
+            authority = after_scheme[:sep_index]
+            remainder = after_scheme[sep_index:]
+            host_part, sep, port_part = authority.rpartition(":")
+            if sep and port_part.isdigit():
+                authority = host_part
+            if authority.count(":") + remainder.count(":") > 0:
                 raise InvalidIRIError(
                     f"Invalid IRI: ':' outside of supported schemes {IRI.SCHEMES} ({value}, {base})"
                 )

tests/test_connection_reuse.py

@@ -0,0 +1,77 @@
+import threading
+
+import urllib3.connectionpool
+import pytest
+
+from graph_db_interface import GraphDB
+
+
+@pytest.fixture
+def count_new_connections(monkeypatch):
+    """Count TCP connections opened by urllib3 while the fixture is active.
+
+    A client that reuses one session keeps urllib3's connection pool alive between
+    calls, so repeated queries open no new connection. A client that goes through the
+    module-level `requests` helpers builds a fresh session — and therefore a fresh
+    pool — per call, so every query opens one.
+    """
+    opened = []
+
+    for pool_cls in (
+        urllib3.connectionpool.HTTPConnectionPool,
+        urllib3.connectionpool.HTTPSConnectionPool,
+    ):
+        original = pool_cls._new_conn
+
+        def counting_new_conn(self, _original=original):
+            opened.append(type(self).__name__)
+            return _original(self)
+
+        monkeypatch.setattr(pool_cls, "_new_conn", counting_new_conn)
+
+    return opened
+
+
+def test_repeated_queries_reuse_one_connection(db: GraphDB, count_new_connections):
+    """Repeated queries must not re-handshake.
+
+    Against a remote HTTPS endpoint the TCP connect plus TLS handshake costs more than
+    the query itself, so a per-call connection makes every query several times slower.
+    """
+    for _ in range(3):
+        db.query("ASK { ?s ?p ?o }")
+
+    assert count_new_connections == [], (
+        f"expected the warm connection pool to be reused, but "
+        f"{len(count_new_connections)} new connection(s) were opened: {count_new_connections}"
+    )
+
+
+def test_client_holds_a_persistent_session(db: GraphDB):
+    """The session is one object for the thread's lifetime, not one per request."""
+    before = db._session
+    db.query("ASK { ?s ?p ?o }")
+
+    assert db._session is before
+
+
+def test_each_thread_gets_its_own_session(db: GraphDB):
+    """One client is driven from several threads at once — a web framework serving requests
+    while the embedding code queries. A `requests.Session` mutates its cookie jar on every
+    response and is not thread-safe, so threads must not share one.
+    """
+    sessions = []
+
+    def query_from_thread():
+        db.query("ASK { ?s ?p ?o }")
+        sessions.append(db._session)
+
+    threads = [threading.Thread(target=query_from_thread) for _ in range(3)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=30)
+
+    assert len(sessions) == 3, "a worker thread failed to complete its query"
+    assert len(set(map(id, sessions))) == 3, "threads shared a session"
+    assert db._session not in sessions, "a worker thread reused the main thread's session"