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)