Files @ ddff5ce676eb
Branch filter:

Location: light9/light9/typedgraph.py

drewp@bigasterisk.com
reorder
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)