# imports import typing # bsfs imports from bsfs.graph import Graph, ac from bsfs.triple_store import TripleStoreBase, SparqlStore from bsfs.utils import URI, errors # exports __all__: typing.Sequence[str] = ( 'build_graph', ) # constants _graph_classes = { 'Graph': Graph, } _backend_classes = { 'SparqlStore': SparqlStore, } ## code ## def build_backend(cfg: typing.Any) -> TripleStoreBase: """Build and return a backend from user-provided config.""" # essential checks if not isinstance(cfg, dict): raise TypeError(cfg) if len(cfg) != 1: raise errors.ConfigError(f'expected a single key that identifies the backend class, found {list(cfg)}') # unpack from config name = next(iter(cfg)) args = cfg[name] # check name if name not in _backend_classes: raise errors.ConfigError(f'{name} is not a valid triple store class name') # build and return backend cls = _backend_classes[name] return cls.Open(**args) def build_graph(cfg: typing.Any) -> Graph: """Build and return a Graph from user-provided config.""" # essential checks if not isinstance(cfg, dict): raise TypeError(cfg) if len(cfg) != 1: raise errors.ConfigError(f'expected a single key that identifies the graph class, found {list(cfg)}') # unpack from config name = next(iter(cfg)) args = cfg[name] # check name if name not in _graph_classes: raise errors.ConfigError(f'{name} is not a valid graph class name') # check user argument if 'user' not in args: raise errors.ConfigError('required argument "user" is not provided') user = URI(args['user']) # check backend argument if 'backend' not in args: raise errors.ConfigError('required argument "backend" is not provided') backend = build_backend(args['backend']) # build access controls access_controls = ac.NullAC(backend, user) # build and return graph cls = _graph_classes[name] return cls(backend, access_controls) ## EOF ##