From 34dac9e93bdcb975ed32936f496aecd08493207c Mon Sep 17 00:00:00 2001 From: Etienne Hoffmann Date: Tue, 14 Jul 2026 22:40:51 +0200 Subject: [PATCH 1/5] Fix IRI validator rejecting URLs with ports IRI._sanitize rejected any http(s) URL containing a port (e.g. http://host:8991/path): it counted the scheme's own colon plus the authority port colon as "too many" via `raw.count(":") > 1`. This also broke reads, since query(convert_bindings=True) / triples_get convert an xsd:anyURI literal to an IRI, so any stored ported URL made the read itself raise. Fix: after confirming the scheme, strip it, split off the authority (up to the first '/', '#', or '?'), remove a trailing ':' port, and only then reject a stray ':' in the authority or remainder. Ports are accepted; the previously-rejected ":"-for-"#" malformations still raise. All existing tests/test_iri.py cases pass. Found by kapps_semantic_middleware, whose Service ontology stores middleware endpoint URLs (svc:address, svc:endpoint) as xsd:anyURI host:port values. See CHANGELOG.md. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 41 +++++++++++++++++++++++++++++++++ graph_db_interface/utils/iri.py | 19 ++++++++++++++- 2 files changed, 59 insertions(+), 1 deletion(-) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..065876d --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,41 @@ +# Changelog + +## Unreleased + +### Fixed + +- **`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 `:` 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. diff --git a/graph_db_interface/utils/iri.py b/graph_db_interface/utils/iri.py index a2e7b53..f25f1a9 100644 --- a/graph_db_interface/utils/iri.py +++ b/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 + # ':' 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})" ) From 65157c5ea12e382381c9dd7bf015fa754edc22d1 Mon Sep 17 00:00:00 2001 From: Etienne Hoffmann Date: Wed, 15 Jul 2026 13:54:24 +0200 Subject: [PATCH 2/5] Generalize triples_update to atomic add/remove/replace (unequal lengths) triples_update previously raised unless old_triples and new_triples had equal length, limiting it to 1:1 replacement. It already builds a single DELETE ... INSERT ... WHERE SPARQL transaction, which handles additions, removals, and unequal-size replacements equally, so the length restriction is removed and the existence pre-check is skipped for empty old_triples (a pure insert has nothing to check). The single-transaction atomicity is required for SHACL correctness: replacing a cardinality-constrained property (e.g. a possession handover under a "possessed by exactly one resource" shape) must apply removal + insertion together so the intermediate property-absent state is never validated. Enables kapps_ogm.OGM.commit to add/remove/replace atomically. See CHANGELOG.md. Requested by kapps_semantic_middleware. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 20 +++++++++++++++++++ graph_db_interface/queries/triple_multi.py | 23 +++++++++++++++------- 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 065876d..83e4fc0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,26 @@ ## 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 - **`IRI` rejected any URL containing a port** (`graph_db_interface/utils/iri.py`, diff --git a/graph_db_interface/queries/triple_multi.py b/graph_db_interface/queries/triple_multi.py index 6879b7d..c609346 100644 --- a/graph_db_interface/queries/triple_multi.py +++ b/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." From e659da113fd152e6fce0156600de02d611fe0d8c Mon Sep 17 00:00:00 2001 From: Etienne Hoffmann Date: Tue, 28 Jul 2026 12:17:39 +0200 Subject: [PATCH 3/5] Keep blank-node identity across triples_update MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A BNode passed in both old_triples and new_triples is the same node, but the DELETE and INSERT patterns were built from two separate blank-node -> variable maps, so it rendered as ?oldbn1 and ?newbn1 — and every new-side variable was bound by BIND(BNODE()), minting a replacement. Updating one property of an existing blank node therefore unlinked it and orphaned every triple the caller had not listed in old_triples, silently and with a success return. Use one shared map across both pattern sets, and emit BIND(BNODE()) only for blank nodes exclusive to new_triples. Pure additions, pure removals, unequal-length replacements, the IRI-only path and the single-transaction atomicity are all unchanged. tests/test_triple_update_query.py asserts on the generated SPARQL and needs no live repository. Verified red before / green after: the two identity tests fail on the previous implementation, the two control tests pass on both. Refs JaFeKl/graph_db_interface#6 Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 36 ++++++ graph_db_interface/queries/triple_multi.py | 28 +++-- tests/test_triple_update_query.py | 122 +++++++++++++++++++++ 3 files changed, 177 insertions(+), 9 deletions(-) create mode 100644 tests/test_triple_update_query.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 83e4fc0..2acf2c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,42 @@ ### Fixed +- **`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 { :hasValue ?oldbn1 . ?oldbn1 :hasValue 12.1 } + INSERT { :hasValue ?newbn1 . ?newbn1 :hasValue 1.4 } + WHERE { :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`). diff --git a/graph_db_interface/queries/triple_multi.py b/graph_db_interface/queries/triple_multi.py index c609346..aa47c94 100644 --- a/graph_db_interface/queries/triple_multi.py +++ b/graph_db_interface/queries/triple_multi.py @@ -400,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: @@ -420,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) diff --git a/tests/test_triple_update_query.py b/tests/test_triple_update_query.py new file mode 100644 index 0000000..9efd7c6 --- /dev/null +++ b/tests/test_triple_update_query.py @@ -0,0 +1,122 @@ +"""Query-construction tests for `triples_update`. + +These do not need a live GraphDB: they stub `query` and assert on the SPARQL text. +The blank-node behaviour they pin cannot be observed through the return value, and the +existing update tests in `test_graph_manipulation.py` all require a repository. +""" + +import logging +import re + +import pytest +from rdflib import BNode, Literal + +from graph_db_interface import GraphDB, IRI + + +SUB = IRI("https://example.org/Belt1") +PRED = IRI("https://example.org/hasSpeed") +VALUE = IRI("https://example.org/hasValue") + + +@pytest.fixture +def stub_db() -> GraphDB: + """A GraphDB that records the last query instead of sending it. + + Built with `__new__` because `__init__` authenticates against a live server. + """ + db = GraphDB.__new__(GraphDB) + db.logger = logging.getLogger("stub") + db.named_graph = None + db._repository = "stub" + db._blank_ids = set() + db.queries = [] + db.query = lambda query, update=False, **kwargs: (db.queries.append(query), True)[1] + db.all_triple_exists = lambda triples, named_graph=None: True + return db + + +def _variables(block: str) -> list[str]: + return re.findall(r"\?\w+", block) + + +def _split_blocks(query: str) -> dict[str, str]: + match = re.search( + r"DELETE \{(?P.*?)\}\s*INSERT \{(?P.*?)\}\s*WHERE \{(?P.*?)\}\s*$", + query, + re.DOTALL, + ) + assert match is not None, f"unexpected query shape:\n{query}" + return match.groupdict() + + +def test_blank_node_on_both_sides_is_one_variable(stub_db: GraphDB): + """A retained blank node must keep its identity across the update. + + Regression: the DELETE and INSERT patterns used to be rendered from two separate + blank-node maps, so the same node got two variables and the new one was minted by + BIND(BNODE()) — replacing the node and orphaning anything not listed in old_triples. + """ + node = BNode("genid-retained") + + stub_db.triples_update( + old_triples=[(SUB, PRED, node), (node, VALUE, Literal(12.1))], + new_triples=[(SUB, PRED, node), (node, VALUE, Literal(1.4))], + ) + blocks = _split_blocks(stub_db.queries[-1]) + + delete_vars = set(_variables(blocks["delete"])) + insert_vars = set(_variables(blocks["insert"])) + assert len(delete_vars) == 1 + assert delete_vars == insert_vars, "the retained node must be the same variable" + assert "BIND(BNODE()" not in blocks["where"], "a retained node must not be minted" + + +def test_blank_node_only_in_new_triples_is_minted(stub_db: GraphDB): + """A genuinely new anonymous node still gets a fresh store node.""" + fresh = BNode("genid-fresh") + + stub_db.triples_update( + old_triples=[(SUB, PRED, Literal("gone"))], + new_triples=[(SUB, PRED, fresh), (fresh, VALUE, Literal(1.4))], + ) + blocks = _split_blocks(stub_db.queries[-1]) + + insert_vars = set(_variables(blocks["insert"])) + assert len(insert_vars) == 1 + assert blocks["where"].count("BIND(BNODE()") == 1 + assert insert_vars.pop() in blocks["where"] + + +def test_mixed_retained_and_new_blank_nodes(stub_db: GraphDB): + """Only the new-side node is minted; the retained one stays bound by the WHERE.""" + retained, fresh = BNode("genid-retained"), BNode("genid-fresh") + + stub_db.triples_update( + old_triples=[(SUB, PRED, retained), (retained, VALUE, Literal(12.1))], + new_triples=[ + (SUB, PRED, retained), + (retained, VALUE, Literal(1.4)), + (SUB, PRED, fresh), + (fresh, VALUE, Literal(9.9)), + ], + ) + blocks = _split_blocks(stub_db.queries[-1]) + + assert blocks["where"].count("BIND(BNODE()") == 1 + retained_var = set(_variables(blocks["delete"])) + assert len(retained_var) == 1 + assert retained_var <= set(_variables(blocks["insert"])) + assert len(set(_variables(blocks["insert"]))) == 2 + + +def test_no_blank_nodes_is_unchanged(stub_db: GraphDB): + """The ordinary IRI-only path emits no variables and no BIND.""" + stub_db.triples_update( + old_triples=[(SUB, VALUE, Literal(12.1))], + new_triples=[(SUB, VALUE, Literal(1.4))], + ) + query = stub_db.queries[-1] + + assert "?" not in query + assert "BIND(BNODE()" not in query From c90c9400fa67630ec486c05c55add9bfe6ebb6e6 Mon Sep 17 00:00:00 2001 From: Etienne Hoffmann Date: Thu, 30 Jul 2026 10:31:15 +0200 Subject: [PATCH 4/5] Reuse a per-thread HTTP session so requests stop re-handshaking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _make_request dispatched through the module-level requests helpers (getattr(requests, method)(...)), which build a Session, use it for one request, and close it. Closing discards the urllib3 connection pool, so every call paid a fresh TCP connect and a full TLS handshake. Against a remote HTTPS endpoint that dominated the cost of a call: a trivial ASK measured ~17ms, only ~4ms of which was the request. The kapps_semantic_middleware suite issues 129 requests for a single integration test, so the waste compounded — 158s for 146 tests, of which 87-92% was HTTP wall time. The client now holds a persistent session per thread. 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 — a web framework serving requests while the embedding code queries. urllib3's pools are thread-safe; the session around them is not. Each thread pays one handshake, then reuses its own pool. close() releases the pools for every thread that used the client. No signature or response-handling change. Per-request cost ~17ms -> ~4ms. This repo's own suite 12.3s -> 6.6s; the kapps_semantic_middleware suite 158s -> 53s, tests unchanged. The regression tests count urllib3 connection creations rather than timing anything, so they are deterministic and do not flake on a fast or slow network. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 33 +++++++++++++++ graph_db_interface/graph_db.py | 37 +++++++++++++++- tests/test_connection_reuse.py | 77 ++++++++++++++++++++++++++++++++++ 3 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 tests/test_connection_reuse.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2acf2c2..4b558d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,39 @@ ### 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`). diff --git a/graph_db_interface/graph_db.py b/graph_db_interface/graph_db.py index aefd197..895d41c 100644 --- a/graph_db_interface/graph_db.py +++ b/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, diff --git a/tests/test_connection_reuse.py b/tests/test_connection_reuse.py new file mode 100644 index 0000000..c2ceaf0 --- /dev/null +++ b/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" From da083736f24c1c4c7f6169603e36790db12bcbf8 Mon Sep 17 00:00:00 2001 From: Etienne Hoffmann Date: Thu, 30 Jul 2026 18:34:14 +0200 Subject: [PATCH 5/5] Pin the asymmetric triples_update contract with tests 65157c5 widened triples_update from 1:1 replacement to general atomic add/remove/replace, but nothing in this suite held that open: the only existing update test is symmetric 2-in/2-out, so reinstating if len(old_triples) != len(new_triples): raise InvalidInputError(...) would have passed here and broken only downstream, in kapps_ogm, whose OGM.commit feeds a set-difference diff straight into this method. Two tests, both directions -- more added than removed and the reverse -- asserting the resulting graph state rather than just the return value. Verified they fail (8/8 across the named-graph parametrisations) with the length guard temporarily restored. Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_graph_manipulation.py | 42 ++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/test_graph_manipulation.py b/tests/test_graph_manipulation.py index 3386a26..3842ea3 100644 --- a/tests/test_graph_manipulation.py +++ b/tests/test_graph_manipulation.py @@ -301,6 +301,48 @@ def test_update_multiple_triples(db: GraphDB, named_graph: str): assert result is True +ASYM_SUB = "http://example.org#asymmetric_subject" +ASYM_PRED = "http://example.org#asymmetric_predicate" + + +def test_update_adds_more_triples_than_it_removes(db: GraphDB, named_graph: str): + """An update is not restricted to 1:1 replacement. + + `kapps_ogm.OGM.commit` feeds the output of a diff straight to `triples_update`, and a + diff is a set difference — the removed and added lists are almost never the same + length. An equal-length constraint here therefore breaks every such commit, so this + pins the widened contract rather than leaving it to a downstream consumer to discover. + """ + old = [(ASYM_SUB, ASYM_PRED, 1)] + new = [(ASYM_SUB, ASYM_PRED, 2), (ASYM_SUB, ASYM_PRED, 3), (ASYM_SUB, ASYM_PRED, 4)] + + assert db.triples_add(old, named_graph=named_graph) is True + + assert db.triples_update(old_triples=old, new_triples=new, named_graph=named_graph) is True + + assert db.triple_exists(old[0], named_graph=named_graph) is False + for triple in new: + assert db.triple_exists(triple, named_graph=named_graph) is True + + assert db.triples_delete(new, named_graph=named_graph) is True + + +def test_update_removes_more_triples_than_it_adds(db: GraphDB, named_graph: str): + """The reverse direction of the same widening — a diff that drops more than it adds.""" + old = [(ASYM_SUB, ASYM_PRED, 5), (ASYM_SUB, ASYM_PRED, 6), (ASYM_SUB, ASYM_PRED, 7)] + new = [(ASYM_SUB, ASYM_PRED, 8)] + + assert db.triples_add(old, named_graph=named_graph) is True + + assert db.triples_update(old_triples=old, new_triples=new, named_graph=named_graph) is True + + for triple in old: + assert db.triple_exists(triple, named_graph=named_graph) is False + assert db.triple_exists(new[0], named_graph=named_graph) is True + + assert db.triples_delete(new, named_graph=named_graph) is True + + def test_iri_exists(db: GraphDB, named_graph: str): # add a new triple to the default graph result = db.triple_add(