Files @ f13124540331
Branch filter:

Location: light9/light9/typedgraph.py

drewp@bigasterisk.com
factor out typedValue, add many tests, and fail to get it to work
import decimal
from types import UnionType
from typing import Type, TypeVar, cast, get_args

from rdfdb.syncedgraph.syncedgraph import SyncedGraph
from rdflib import Graph
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 _typeIncludes(t1: Type, t2: Type) -> bool:
    """same as issubclass but t1 can be a NewType"""
    # if hasattr(t1, '__supertype__'):
    #     t1 = t1.__supertype__
    print(f'{isinstance(t1,  UnionType)=}')
    if isinstance(t1, UnionType):
        print(f" i see {t1} is union")
        return any(_typeIncludes(t, t2) for t in get_args(t1))
    # print('iss', t1, t2,     isinstance(t1,t2))
    # if t1 is float:
    #     return float in get_args(t2)
    return issubclass(t1, t2)


def typedValue(objType: Type[_ObjType], graph: EitherGraph, subj: Node, pred: Node) -> _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.
    """
    obj = graph.value(subj, pred)
    if obj is None:
        if type(None) in get_args(objType):
            return None
        raise ValueError(f'No obj for {subj=} {pred=}')
    conv = obj  #if _typeIncludes(objType, Node) else obj.toPython()
    # if objType is float and isinstance(conv, decimal.Decimal):
    #     conv = float(conv)
    return cast(objType, conv)