diff options
Diffstat (limited to 'bsfs/schema/serialize.py')
| -rw-r--r-- | bsfs/schema/serialize.py | 17 |
1 files changed, 9 insertions, 8 deletions
diff --git a/bsfs/schema/serialize.py b/bsfs/schema/serialize.py index c1ac9a9..0eb6628 100644 --- a/bsfs/schema/serialize.py +++ b/bsfs/schema/serialize.py @@ -5,7 +5,6 @@ A copy of the license is provided with the project. Author: Matthias Baumgartner, 2022 """ # standard imports -from collections import abc import itertools import typing @@ -42,7 +41,7 @@ def from_string(schema_str: str) -> schema.Schema: return value.value if isinstance(value, rdflib.URIRef): return URI(value) - raise errors.BackendError(f'expected Literal or URIRef, found {typename(value)}') + raise errors.UnreachableError(f'expected Literal or URIRef, found {typename(value)}') def _fetch_hierarchically(factory, curr): """Walk through a rdfs:subClassOf hierarchy, creating symbols along the way.""" @@ -80,14 +79,16 @@ def from_string(schema_str: str) -> schema.Schema: # fetch predicates # FIXME: type annotation def _fetch_value(subject: URI, predicate: rdflib.URIRef, value_factory) -> typing.Optional[typing.Any]: - """Fetch the object of a given subject and predicate. Raises a `errors.ConsistencyError` if multiple objects match.""" + """Fetch the object of a given subject and predicate. + Raises a `errors.ConsistencyError` if multiple objects match. + """ values = list(graph.objects(rdflib.URIRef(subject), predicate)) if len(values) == 0: return None - elif len(values) == 1: + if len(values) == 1: return value_factory(values[0]) - else: - raise errors.ConsistencyError(f'{subject} has multiple values for predicate {str(predicate)}, expected zero or one') + raise errors.ConsistencyError( + f'{subject} has multiple values for predicate {str(predicate)}, expected zero or one') def _build_predicate(uri, parent, **annotations): """Predicate factory.""" @@ -102,13 +103,13 @@ def from_string(schema_str: str) -> schema.Schema: dom = _fetch_value(uri, rdflib.RDFS.domain, URI) if dom is not None and dom not in nodes_lut: raise errors.ConsistencyError(f'predicate {uri} has undefined domain {dom}') - elif dom is not None: + if dom is not None: dom = nodes_lut[dom] # get range rng = _fetch_value(uri, rdflib.RDFS.range, URI) if rng is not None and rng not in nodes_lut and rng not in literals_lut: raise errors.ConsistencyError(f'predicate {uri} has undefined range {rng}') - elif rng is not None: + if rng is not None: rng = nodes_lut.get(rng, literals_lut.get(rng)) # get unique unique = _fetch_value(uri, rdflib.URIRef(ns.bsfs.unique), bool) |