from typing import List, Type, TypeVar, cast, get_args from rdfdb.syncedgraph.syncedgraph import SyncedGraph from rdflib import XSD, Graph, Literal, URIRef from rdflib.term import Node # todo: this ought to just require a suitable graph.value method EitherGraph = Graph | SyncedGraph _ObjType = TypeVar('_ObjType') class ConversionError(ValueError): """graph had a value, but it does not safely convert to any of the requested types""" def _expandUnion(t: Type) -> List[Type]: if hasattr(t, '__args__'): return list(get_args(t)) return [t] def _typeIncludes(t1: Type, t2: Type) -> bool: """same as issubclass but t1 can be a NewType""" if t2 is None: t2 = type(None) if t1 == t2: return True if getattr(t1, '__supertype__', None) == t2: return True ts = _expandUnion(t1) if len(ts) > 1: return any(_typeIncludes(t, t2) for t in ts) # if t1 is float: # return float in get_args(t2) print(f'down to {t1} {t2}') return False def _convLiteral(objType: Type[_ObjType], x: Literal) -> _ObjType: if _typeIncludes(objType, Literal): return cast(objType, x) for outType, dtypes in [ (float, (XSD['integer'], XSD['double'], XSD['decimal'])), (int, (XSD['integer'],)), (str, ()), ]: for t in _expandUnion(objType): if _typeIncludes(t, outType) and (not dtypes or x.datatype in dtypes): # e.g. user wants float and we have xsd:double return cast(objType, outType(x.toPython())) raise ConversionError def typedValue(objType: Type[_ObjType], graph: EitherGraph, subj: Node, pred: URIRef) -> _ObjType: """graph.value(subj, pred) with a given return type. If objType is not an rdflib.Node, we toPython() the value. Allow objType to include None if you want a None return for not-found. """ if objType is None: raise TypeError('must allow non-None result type') obj = graph.value(subj, pred) if obj is None: if _typeIncludes(objType, None): return cast(objType, None) raise ValueError(f'No obj for {subj=} {pred=}') ConvFrom: Type[Node] = type(obj) ConvTo = objType try: if ConvFrom == URIRef and _typeIncludes(ConvTo, URIRef): conv = obj elif ConvFrom == Literal: conv = _convLiteral(objType, cast(Literal, obj)) else: # e.g. BNode is not handled yet raise ConversionError except ConversionError: raise ConversionError(f'graph contains {type(obj)}, caller requesting {objType}') # if objType is float and isinstance(conv, decimal.Decimal): # conv = float(conv) return cast(objType, conv)