Skip to content

Commit

Permalink
Fallback to old Store.bind signature on TypeError
Browse files Browse the repository at this point in the history
If `Store.bind` raises a `TypeError`, and the string coversion of this
TypeError contains `override`, then log a warning and call `Store.bind`
without an override.

This is done so that stores that do not accept `override` on
`Store.bind` still work, but at the cost of still having the bug that
was fixed by introducing the `override` parameter.

Also added a private flag which can be used to disable the fix entirely
and never use `override` when calling `Store.bind`.
  • Loading branch information
aucampia committed Jul 14, 2022
1 parent ac8ef91 commit ddb0561
Show file tree
Hide file tree
Showing 2 changed files with 235 additions and 4 deletions.
26 changes: 22 additions & 4 deletions rdflib/namespace/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -346,6 +346,8 @@ def _ipython_key_completions_(self) -> List[str]:
if TYPE_CHECKING:
from rdflib._type_checking import _NamespaceSetString

_with_bind_override_fix = True


class NamespaceManager(object):
"""Class for managing prefix => namespace mappings
Expand Down Expand Up @@ -618,6 +620,22 @@ def expand_curie(self, curie: str) -> Union[URIRef, None]:
f"Prefix \"{curie.split(':')[0]}\" not bound to any namespace."
)

def _store_bind(self, prefix: str, namespace: URIRef, override: bool) -> None:
if not _with_bind_override_fix:
return self.store.bind(prefix, namespace)
try:
return self.store.bind(prefix, namespace, override=override)
except TypeError as error:
if "override" in str(error):
logger.warning(
"caught a TypeError, "
"retrying call to %s.bind without override, "
"see https://github.com/RDFLib/rdflib/issues/1880 for more info",
type(self.store),
exc_info=True,
)
return self.store.bind(prefix, namespace)

def bind(
self,
prefix: Optional[str],
Expand Down Expand Up @@ -652,7 +670,7 @@ def bind(
if bound_namespace and bound_namespace != namespace:

if replace:
self.store.bind(prefix, namespace, override=override)
self._store_bind(prefix, namespace, override=override)
insert_trie(self.__trie, str(namespace))
return
# prefix already in use for different namespace
Expand All @@ -673,16 +691,16 @@ def bind(
if not self.store.namespace(new_prefix):
break
num += 1
self.store.bind(new_prefix, namespace, override=override)
self._store_bind(new_prefix, namespace, override=override)
else:
bound_prefix = self.store.prefix(namespace)
if bound_prefix is None:
self.store.bind(prefix, namespace, override=override)
self._store_bind(prefix, namespace, override=override)
elif bound_prefix == prefix:
pass # already bound
else:
if override or bound_prefix.startswith("_"): # or a generated prefix
self.store.bind(prefix, namespace, override=override)
self._store_bind(prefix, namespace, override=override)

insert_trie(self.__trie, str(namespace))

Expand Down
213 changes: 213 additions & 0 deletions test/test_graph/test_graph_store.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,213 @@
"""
Tests for usage of the Store interface from Graph/NamespaceManager.
"""

import itertools
import logging
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
Optional,
Sequence,
Tuple,
Type,
Union,
)

import pytest

import rdflib.namespace
from rdflib.graph import Graph
from rdflib.namespace import Namespace
from rdflib.plugins.stores.memory import Memory
from rdflib.store import Store
from rdflib.term import URIRef

if TYPE_CHECKING:
from _pytest.mark.structures import ParameterSet

NamespaceBindings = Dict[str, URIRef]


def check_ns(graph: Graph, expected_bindings: NamespaceBindings) -> None:
actual_graph_bindings = list(graph.namespaces())
logging.info("actual_bindings = %s", actual_graph_bindings)
assert list(expected_bindings.items()) == actual_graph_bindings
store: Store = graph.store
actual_store_bindings = list(store.namespaces())
assert list(expected_bindings.items()) == actual_store_bindings
for prefix, uri in expected_bindings.items():
assert store.prefix(uri) == prefix
assert store.namespace(prefix) == uri


class MemoryWithoutBindOverride(Memory):
def bind(self, prefix, namespace) -> None: # type: ignore[override]
return super().bind(prefix, namespace, False)


class GraphWithoutBindOverrideFix(Graph):
def bind(self, prefix, namespace, override=True, replace=False) -> None:
old_value = rdflib.namespace._with_bind_override_fix
rdflib.namespace._with_bind_override_fix = False
try:
return super().bind(prefix, namespace, override, replace)
finally:
rdflib.namespace._with_bind_override_fix = old_value


GraphFactory = Callable[[], Graph]
GraphOperation = Callable[[Graph], None]
GraphOperations = Sequence[GraphOperation]

EGNS = Namespace("http://example.com/namespace#")
EGNS_V0 = EGNS["v0"]
EGNS_V1 = EGNS["v1"]
EGNS_V2 = EGNS["v2"]


def make_test_graph_store_bind_cases(
store_type: Type[Store] = Memory,
graph_type: Type[Graph] = Graph,
) -> Iterable[Union[Tuple[Any, ...], "ParameterSet"]]:
"""
Generate test cases for test_graph_store_bind.
"""

def graph_factory():
return graph_type(bind_namespaces="none", store=store_type())

id_prefix = f"{store_type.__name__}-{graph_type.__name__}"

def _p(
graph_factory: GraphFactory,
ops: GraphOperations,
expected_bindings: NamespaceBindings,
expected_bindings_overrides: Optional[
Dict[Tuple[Type[Graph], Type[Store]], NamespaceBindings]
] = None,
*,
id: Optional[str] = None,
):
if expected_bindings_overrides is not None:
expected_bindings = expected_bindings_overrides.get(
(graph_type, store_type), expected_bindings
)
if id is not None:
return pytest.param(graph_factory, ops, expected_bindings, id=id)
else:
return (graph_factory, ops, expected_bindings)

yield from [
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
],
{"eg": EGNS_V0},
id=f"{id_prefix}-default-args",
),
# reused-prefix
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("eg", EGNS_V1, override=False),
],
{"eg": EGNS_V0, "eg1": EGNS_V1},
id=f"{id_prefix}-reused-prefix-override-false-replace-false",
),
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("eg", EGNS_V1),
],
{"eg": EGNS_V0, "eg1": EGNS_V1},
id=f"{id_prefix}-reused-prefix-override-true-replace-false",
),
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("eg", EGNS_V1, override=False, replace=True),
],
{"eg": EGNS_V0},
{(GraphWithoutBindOverrideFix, Memory): {"eg": EGNS_V1}},
id=f"{id_prefix}-reused-prefix-override-false-replace-true",
),
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("eg", EGNS_V1, replace=True),
],
{"eg": EGNS_V1},
{(Graph, MemoryWithoutBindOverride): {"eg": EGNS_V0}},
id=f"{id_prefix}-reused-prefix-override-true-replace-true",
),
# reused-namespace
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("egv0", EGNS_V0, override=False),
],
{"eg": EGNS_V0},
id=f"{id_prefix}-reused-namespace-override-false-replace-false",
),
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("egv0", EGNS_V0),
],
{"egv0": EGNS_V0},
{(Graph, MemoryWithoutBindOverride): {"eg": EGNS_V0}},
id=f"{id_prefix}-reused-namespace-override-true-replace-false",
),
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("egv0", EGNS_V0, override=False, replace=True),
],
{"eg": EGNS_V0},
id=f"{id_prefix}-reused-namespace-override-false-replace-true",
),
_p(
graph_factory,
[
lambda g: g.bind("eg", EGNS_V0),
lambda g: g.bind("egv0", EGNS_V0, replace=True),
],
{"egv0": EGNS_V0},
{(Graph, MemoryWithoutBindOverride): {"eg": EGNS_V0}},
id=f"{id_prefix}-reused-namespace-override-true-replace-true",
),
]


@pytest.mark.parametrize(
["graph_factory", "ops", "expected_bindings"],
itertools.chain(
make_test_graph_store_bind_cases(),
make_test_graph_store_bind_cases(store_type=MemoryWithoutBindOverride),
make_test_graph_store_bind_cases(graph_type=GraphWithoutBindOverrideFix),
),
)
def test_graph_store_bind(
graph_factory: GraphFactory,
ops: GraphOperations,
expected_bindings: NamespaceBindings,
) -> None:
"""
The expected sequence of graph operations results in the expected namespace bindings.
"""
graph = graph_factory()
for op in ops:
op(graph)
check_ns(graph, expected_bindings)

0 comments on commit ddb0561

Please sign in to comment.