Skip to content

Commit

Permalink
Merge pull request #1680 from ajnelson-nist/add_identified_node
Browse files Browse the repository at this point in the history
Add IdentifiedNode abstract intermediary class
  • Loading branch information
nicholascar authored Jan 20, 2022
2 parents 107504c + 9189df8 commit 3dbc848
Show file tree
Hide file tree
Showing 3 changed files with 28 additions and 31 deletions.
3 changes: 2 additions & 1 deletion rdflib/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@
__all__ = [
"URIRef",
"BNode",
"IdentifiedNode",
"Literal",
"Variable",
"Namespace",
Expand Down Expand Up @@ -155,7 +156,7 @@
Literal work, eq, __neq__, __lt__, etc.
"""

from rdflib.term import URIRef, BNode, Literal, Variable
from rdflib.term import URIRef, BNode, IdentifiedNode, Literal, Variable

from rdflib.graph import Dataset, Graph, ConjunctiveGraph

Expand Down
31 changes: 17 additions & 14 deletions rdflib/term.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
__all__ = [
"bind",
"Node",
"IdentifiedNode",
"Identifier",
"URIRef",
"BNode",
Expand Down Expand Up @@ -218,7 +219,21 @@ def startswith(self, prefix, start=..., end=...) -> bool:
__hash__ = str.__hash__


class URIRef(Identifier):
class IdentifiedNode(Identifier):
"""
An abstract class, primarily defined to identify Nodes that are not Literals.
The name "Identified Node" is not explicitly defined in the RDF specification, but can be drawn from this section: https://www.w3.org/TR/rdf-concepts/#section-URI-Vocabulary
"""

def __getnewargs__(self):
return (str(self),)

def toPython(self) -> str:
return str(self)


class URIRef(IdentifiedNode):
"""
RDF URI Reference: http://www.w3.org/TR/rdf-concepts/#section-Graph-URIref
"""
Expand Down Expand Up @@ -250,9 +265,6 @@ def __new__(cls, value: str, base: Optional[str] = None):
rt = str.__new__(cls, value, "utf-8") # type: ignore[call-overload]
return rt

def toPython(self) -> str:
return str(self)

def n3(self, namespace_manager=None) -> str:
"""
This will do a limited check for valid URIs,
Expand Down Expand Up @@ -284,9 +296,6 @@ def defrag(self):
def __reduce__(self):
return (URIRef, (str(self),))

def __getnewargs__(self):
return (str(self),)

def __repr__(self):
if self.__class__ is URIRef:
clsName = "rdflib.term.URIRef"
Expand Down Expand Up @@ -386,7 +395,7 @@ def _generator():
return _generator


class BNode(Identifier):
class BNode(IdentifiedNode):
"""
Blank Node: http://www.w3.org/TR/rdf-concepts/#section-blank-nodes
Expand Down Expand Up @@ -418,15 +427,9 @@ def __new__(
# http://www.w3.org/TR/2004/REC-rdf-testcases-20040210/#nodeID
return Identifier.__new__(cls, value) # type: ignore[return-value]

def toPython(self) -> str:
return str(self)

def n3(self, namespace_manager=None):
return "_:%s" % self

def __getnewargs__(self):
return (str(self),)

def __reduce__(self):
return (BNode, (str(self),))

Expand Down
25 changes: 9 additions & 16 deletions test/test_typing.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,6 @@
import rdflib.plugins.sparql.processor


# TODO Question - is there a usable type name or class name for
# 'typing.Union[rdflib.BNode, rdflib.URIRef]'?
# Conversation to resolve:
# https://github.com/RDFLib/rdflib/issues/1526
example_BlankNodeOrIRI = Union[rdflib.BNode, rdflib.URIRef]


def test_rdflib_query_exercise() -> None:
"""
The purpose of this test is to exercise a selection of rdflib features under "mypy --strict" review.
Expand Down Expand Up @@ -63,10 +56,10 @@ def test_rdflib_query_exercise() -> None:
graph.add((kb_https_uriref, predicate_q, literal_two))
graph.add((kb_bnode, predicate_p, literal_one))

expected_nodes_using_predicate_q: Set[example_BlankNodeOrIRI] = {
expected_nodes_using_predicate_q: Set[rdflib.IdentifiedNode] = {
kb_https_uriref
}
computed_nodes_using_predicate_q: Set[example_BlankNodeOrIRI] = set()
computed_nodes_using_predicate_q: Set[rdflib.IdentifiedNode] = set()
for triple in graph.triples((None, predicate_q, None)):
computed_nodes_using_predicate_q.add(triple[0])
assert expected_nodes_using_predicate_q == computed_nodes_using_predicate_q
Expand All @@ -78,13 +71,13 @@ def test_rdflib_query_exercise() -> None:
}
"""

expected_one_usage: Set[example_BlankNodeOrIRI] = {
expected_one_usage: Set[rdflib.IdentifiedNode] = {
kb_bnode,
kb_http_uriref,
kb_https_uriref,
kb_urn_uriref,
}
computed_one_usage: Set[example_BlankNodeOrIRI] = set()
computed_one_usage: Set[rdflib.IdentifiedNode] = set()
for one_usage_result in graph.query(one_usage_query):
computed_one_usage.add(one_usage_result[0])
assert expected_one_usage == computed_one_usage
Expand All @@ -103,14 +96,14 @@ def test_rdflib_query_exercise() -> None:

expected_two_usage: Set[
Tuple[
example_BlankNodeOrIRI,
example_BlankNodeOrIRI,
rdflib.IdentifiedNode,
rdflib.IdentifiedNode,
]
] = {(kb_https_uriref, predicate_p), (kb_https_uriref, predicate_q)}
computed_two_usage: Set[
Tuple[
example_BlankNodeOrIRI,
example_BlankNodeOrIRI,
rdflib.IdentifiedNode,
rdflib.IdentifiedNode,
]
] = set()
for two_usage_result in graph.query(two_usage_query):
Expand All @@ -122,7 +115,7 @@ def test_rdflib_query_exercise() -> None:
prepared_one_usage_query = rdflib.plugins.sparql.processor.prepareQuery(
one_usage_query, initNs=nsdict
)
computed_one_usage_from_prepared_query: Set[example_BlankNodeOrIRI] = set()
computed_one_usage_from_prepared_query: Set[rdflib.IdentifiedNode] = set()
for prepared_one_usage_result in graph.query(prepared_one_usage_query):
computed_one_usage_from_prepared_query.add(prepared_one_usage_result[0])
assert expected_one_usage == computed_one_usage_from_prepared_query
Expand Down

0 comments on commit 3dbc848

Please sign in to comment.