Source

graphalchemy / graphalchemy / models / readers.py

Full commit
# -*- coding: utf-8 -*-
'''graph model readers'''

from appspace import AppLookupError

from graphalchemy.mixins.workers import ReaderMixin
from graphalchemy.core import octopus, join, factory

__all__ = ['Nodes']
# settings
conf = octopus.G
backends = conf.backends
db = conf.key.db


class Nodes(ReaderMixin, octopus.context.Thing):

    '''graph node model reader'''

    # model finder
    _finder = join(conf.model.finder.node, conf.models)
    # direct node reader
    _r = factory(conf.read.node, backends, db)
    # direct node writer
    _w = factory(conf.write.node, backends, db)

    def __init__(self, model):
        '''
        init

        @param model: graph model
        '''
        super(Nodes, self).__init__()
        self.model = model

    def anchor(self):
        '''anchor node for a graph model'''
        try:
            L = self.model.L
            anchor_link = L.anchor_link
            G = self.G
            userspace = G.userspace
            return self.Q.app(anchor_link, userspace)
        except AppLookupError:
            label = G.anchor.label
            anchor = self.filter_by(L.index, label, label)
            if anchor:
                self.Q.add(anchor, anchor_link, userspace)
                return anchor

    def reference(self):
        '''reference node for a model'''
        try:
            G = self.G
            L = self.model.L
            return self.Q.app(L.reference_link, G.userspace)
        except AppLookupError:
            label = G.reference.label
            ref = self.filter_by(L.index, label, label)
            if ref:
                self.Q.add(ref, L.reference_link, G.userspace)
                return ref

    def filter(self, index, queries):
        '''
        full text search using index

        @param index: name of index
        @param queries: queries built with query builder
        '''
        return self._r.filter(index, queries, self.model)

    def filter_by(self, index, key, value):
        '''
        filter by keywords

        @param index: index name
        @param key: keyword in index
        @param value: value in index (or second key)
        '''
        return self._r.filter_by(
            index,
            key,
            value,
            lambda x: self._finder(element=x)(x),
        )

    def get(self, element):
        '''
        get database element by id

        @param element: element id
        '''
        return self.model(self._r.get(element))

    def traverse(self, this, tester=None, links=None, unique=None):
        '''
        traverse graph, yielding graph elements

        @param this: graph object
        @param tester: filter for traversed elements (default: None)
        @param links: links to traverse (default: None)
        @param unique: how often to traverse the same element (default: None)
        '''
        finder = self._finder
        for element in self._r.traverser(this, tester, links, unique):
            yield finder(element=element)(element)

    def walk(self, this, direction=None, label=None, keys=None, end='end'):
        '''
        iterate links and yield node from link

        @param this: graph node instance
        @param direction: direction of links (default: None)
        @param label: label for links (default: None)
        @param keys: keys to find on elements (default: None)
        @param end: which end of link to return (default: 'end')
        '''
        finder = self._finder
        for element in self._r.walker(this, direction, label, keys, end):
            yield finder(element=element)(element)