summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--rdflib/extras/external_graph_libs.py30
-rw-r--r--rdflib/namespace/__init__.py2
-rw-r--r--rdflib/plugins/serializers/nt.py4
-rw-r--r--rdflib/plugins/serializers/trig.py17
-rw-r--r--rdflib/serializer.py13
-rw-r--r--rdflib/tools/chunk_serializer.py3
-rw-r--r--rdflib/tools/defined_namespace_creator.py34
7 files changed, 69 insertions, 34 deletions
diff --git a/rdflib/extras/external_graph_libs.py b/rdflib/extras/external_graph_libs.py
index 11d2ca1e..f50994b5 100644
--- a/rdflib/extras/external_graph_libs.py
+++ b/rdflib/extras/external_graph_libs.py
@@ -1,5 +1,6 @@
#!/usr/bin/env python
# encoding: utf-8
+from __future__ import annotations
"""Convert (to and) from rdflib graphs to other well known graph libraries.
@@ -13,6 +14,10 @@ see ../../test/test_extras_external_graph_libs.py for conditional tests
"""
import logging
+from typing import TYPE_CHECKING, Any, Dict, List
+
+if TYPE_CHECKING:
+ from rdflib.graph import Graph
logger = logging.getLogger(__name__)
@@ -22,9 +27,9 @@ def _identity(x):
def _rdflib_to_networkx_graph(
- graph,
+ graph: Graph,
nxgraph,
- calc_weights,
+ calc_weights: bool,
edge_attrs,
transform_s=_identity,
transform_o=_identity,
@@ -70,7 +75,7 @@ def _rdflib_to_networkx_graph(
def rdflib_to_networkx_multidigraph(
- graph, edge_attrs=lambda s, p, o: {"key": p}, **kwds
+ graph: Graph, edge_attrs=lambda s, p, o: {"key": p}, **kwds
):
"""Converts the given graph into a networkx.MultiDiGraph.
@@ -124,8 +129,8 @@ def rdflib_to_networkx_multidigraph(
def rdflib_to_networkx_digraph(
- graph,
- calc_weights=True,
+ graph: Graph,
+ calc_weights: bool = True,
edge_attrs=lambda s, p, o: {"triples": [(s, p, o)]},
**kwds,
):
@@ -187,8 +192,8 @@ def rdflib_to_networkx_digraph(
def rdflib_to_networkx_graph(
- graph,
- calc_weights=True,
+ graph: Graph,
+ calc_weights: bool = True,
edge_attrs=lambda s, p, o: {"triples": [(s, p, o)]},
**kwds,
):
@@ -250,9 +255,9 @@ def rdflib_to_networkx_graph(
def rdflib_to_graphtool(
- graph,
- v_prop_names=[str("term")],
- e_prop_names=[str("term")],
+ graph: Graph,
+ v_prop_names: List[str] = [str("term")],
+ e_prop_names: List[str] = [str("term")],
transform_s=lambda s, p, o: {str("term"): s},
transform_p=lambda s, p, o: {str("term"): p},
transform_o=lambda s, p, o: {str("term"): o},
@@ -313,7 +318,8 @@ def rdflib_to_graphtool(
True
"""
- import graph_tool as gt
+ # pytype error: Can't find module 'graph_tool'.
+ import graph_tool as gt # pytype: disable=import-error
g = gt.Graph()
@@ -323,7 +329,7 @@ def rdflib_to_graphtool(
eprops = [(epn, g.new_edge_property("object")) for epn in e_prop_names]
for epn, eprop in eprops:
g.edge_properties[epn] = eprop
- node_to_vertex = {}
+ node_to_vertex: Dict[Any, Any] = {}
for s, p, o in graph:
sv = node_to_vertex.get(s)
if sv is None:
diff --git a/rdflib/namespace/__init__.py b/rdflib/namespace/__init__.py
index 93f692ad..5bfac7c6 100644
--- a/rdflib/namespace/__init__.py
+++ b/rdflib/namespace/__init__.py
@@ -592,7 +592,7 @@ class NamespaceManager(object):
return self.__cache_strict[uri]
- def expand_curie(self, curie: str) -> Union[URIRef, None]:
+ def expand_curie(self, curie: str) -> URIRef:
"""
Expand a CURIE of the form <prefix:element>, e.g. "rdf:type"
into its full expression:
diff --git a/rdflib/plugins/serializers/nt.py b/rdflib/plugins/serializers/nt.py
index 8efa0af8..5faf9efb 100644
--- a/rdflib/plugins/serializers/nt.py
+++ b/rdflib/plugins/serializers/nt.py
@@ -33,7 +33,7 @@ class NTSerializer(Serializer):
base: Optional[str] = None,
encoding: Optional[str] = "utf-8",
**args,
- ):
+ ) -> None:
if base is not None:
warnings.warn("NTSerializer does not support base.")
if encoding != "utf-8":
@@ -57,7 +57,7 @@ class NT11Serializer(NTSerializer):
Serializer.__init__(self, store) # default to utf-8
-def _nt_row(triple: _TripleType):
+def _nt_row(triple: _TripleType) -> str:
if isinstance(triple[2], Literal):
return "%s %s %s .\n" % (
# type error: "Node" has no attribute "n3"
diff --git a/rdflib/plugins/serializers/trig.py b/rdflib/plugins/serializers/trig.py
index 3b8b2aab..18bee3f2 100644
--- a/rdflib/plugins/serializers/trig.py
+++ b/rdflib/plugins/serializers/trig.py
@@ -3,12 +3,15 @@ Trig RDF graph serializer for RDFLib.
See <http://www.w3.org/TR/trig/> for syntax specification.
"""
-from typing import IO, TYPE_CHECKING, Optional, Union
+from typing import IO, TYPE_CHECKING, Dict, List, Optional, Tuple, Union
from rdflib.graph import ConjunctiveGraph, Graph
from rdflib.plugins.serializers.turtle import TurtleSerializer
from rdflib.term import BNode, Node
+if TYPE_CHECKING:
+ from rdflib.graph import _ContextType, _SubjectType
+
__all__ = ["TrigSerializer"]
@@ -31,7 +34,7 @@ class TrigSerializer(TurtleSerializer):
super(TrigSerializer, self).__init__(store)
- def preprocess(self):
+ def preprocess(self) -> None:
for context in self.contexts:
# do not write unnecessary prefix (ex: for an empty default graph)
if len(context) == 0:
@@ -48,9 +51,12 @@ class TrigSerializer(TurtleSerializer):
self._contexts[context] = (self.orderSubjects(), self._subjects)
- def reset(self):
+ def reset(self) -> None:
super(TrigSerializer, self).reset()
- self._contexts = {}
+ self._contexts: Dict[
+ _ContextType,
+ Tuple[List[_SubjectType], Dict[_SubjectType, bool]],
+ ] = {}
def serialize(
self,
@@ -93,7 +99,8 @@ class TrigSerializer(TurtleSerializer):
else:
iri = self.getQName(store.identifier)
if iri is None:
- iri = store.identifier.n3()
+ # type error: "IdentifiedNode" has no attribute "n3"
+ iri = store.identifier.n3() # type: ignore[attr-defined]
self.write(self.indent() + "\n%s {" % iri)
self.depth += 1
diff --git a/rdflib/serializer.py b/rdflib/serializer.py
index 064232b5..95f910a2 100644
--- a/rdflib/serializer.py
+++ b/rdflib/serializer.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
"""
Serializer plugin interface.
@@ -10,7 +12,7 @@ See also rdflib.plugin
"""
-from typing import IO, TYPE_CHECKING, Optional
+from typing import IO, TYPE_CHECKING, Any, Optional, TypeVar, Union
from rdflib.term import URIRef
@@ -19,6 +21,8 @@ if TYPE_CHECKING:
__all__ = ["Serializer"]
+_StrT = TypeVar("_StrT", bound=str)
+
class Serializer:
def __init__(self, store: "Graph"):
@@ -31,12 +35,13 @@ class Serializer:
stream: IO[bytes],
base: Optional[str] = None,
encoding: Optional[str] = None,
- **args,
+ **args: Any,
) -> None:
"""Abstract method"""
- def relativize(self, uri: str):
+ def relativize(self, uri: _StrT) -> Union[_StrT, URIRef]:
base = self.base
if base is not None and uri.startswith(base):
- uri = URIRef(uri.replace(base, "", 1))
+ # type error: Incompatible types in assignment (expression has type "str", variable has type "Node")
+ uri = URIRef(uri.replace(base, "", 1)) # type: ignore[assignment]
return uri
diff --git a/rdflib/tools/chunk_serializer.py b/rdflib/tools/chunk_serializer.py
index cb18d399..9f9f133a 100644
--- a/rdflib/tools/chunk_serializer.py
+++ b/rdflib/tools/chunk_serializer.py
@@ -98,7 +98,8 @@ def serialize_in_chunks(
row_bytes = _nt_row(t).encode("utf-8")
if len(row_bytes) > max_file_size:
raise ValueError(
- f"cannot write triple {t!r} as it's serialized size of {row_bytes / 1000} exceeds max_file_size_kb = {max_file_size_kb}"
+ # type error: Unsupported operand types for / ("bytes" and "int")
+ f"cannot write triple {t!r} as it's serialized size of {row_bytes / 1000} exceeds max_file_size_kb = {max_file_size_kb}" # type: ignore[operator]
)
if i == 0:
fp, fhb = xstack.enter_context(_start_new_file(file_no))
diff --git a/rdflib/tools/defined_namespace_creator.py b/rdflib/tools/defined_namespace_creator.py
index 2cfe99f2..0c93ea75 100644
--- a/rdflib/tools/defined_namespace_creator.py
+++ b/rdflib/tools/defined_namespace_creator.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
"""
This rdflib Python script creates a DefinedNamespace Python file from a given RDF file
@@ -14,20 +16,24 @@ import argparse
import datetime
import sys
from pathlib import Path
+from typing import TYPE_CHECKING, Iterable, List, Tuple
sys.path.append(str(Path(__file__).parent.absolute().parent.parent))
-from rdflib import Graph
-from rdflib.namespace import DCTERMS, OWL, RDFS, SKOS
-from rdflib.util import guess_format
+from rdflib.graph import Graph # noqa: E402
+from rdflib.namespace import DCTERMS, OWL, RDFS, SKOS # noqa: E402
+from rdflib.util import guess_format # noqa: E402
+
+if TYPE_CHECKING:
+ from rdflib.query import ResultRow
-def validate_namespace(namespace):
+def validate_namespace(namespace: str) -> None:
if not namespace.endswith(("/", "#")):
raise ValueError("The supplied namespace must end with '/' or '#'")
-def validate_object_id(object_id):
+def validate_object_id(object_id: str) -> None:
for c in object_id:
if not c.isupper():
raise ValueError("The supplied object_id must be an all-capitals string")
@@ -66,7 +72,9 @@ def validate_object_id(object_id):
# return classes
-def get_target_namespace_elements(g, target_namespace):
+def get_target_namespace_elements(
+ g: Graph, target_namespace: str
+) -> Tuple[List[Tuple[str, str]], List[str]]:
namespaces = {"dcterms": DCTERMS, "owl": OWL, "rdfs": RDFS, "skos": SKOS}
q = """
SELECT DISTINCT ?s ?def
@@ -85,13 +93,15 @@ def get_target_namespace_elements(g, target_namespace):
""".replace(
"xxx", target_namespace
)
- elements = []
+ elements: List[Tuple[str, str]] = []
for r in g.query(q, initNs=namespaces):
+ if TYPE_CHECKING:
+ assert isinstance(r, ResultRow)
elements.append((str(r[0]), str(r[1])))
elements.sort(key=lambda tup: tup[0])
- elements_strs = []
+ elements_strs: List[str] = []
for e in elements:
desc = e[1].replace("\n", " ")
elements_strs.append(
@@ -101,7 +111,13 @@ def get_target_namespace_elements(g, target_namespace):
return elements, elements_strs
-def make_dn_file(output_file_name, target_namespace, elements_strs, object_id, fail):
+def make_dn_file(
+ output_file_name: Path,
+ target_namespace: str,
+ elements_strs: Iterable[str],
+ object_id: str,
+ fail: bool,
+) -> None:
header = f'''from rdflib.term import URIRef
from rdflib.namespace import DefinedNamespace, Namespace
View Lorry Controller Status