Source

graphalchemy / graphalchemy / models / workers.py

# -*- coding: utf-8 -*-
'''graph model manager'''

from itertools import count
from functools import partial

from markupsafe import escape

from graphalchemy.mixins.workers import WriterMixin
from graphalchemy.core import octopus, app, factory, defer

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


class Nodes(WriterMixin, octopus.workflow.Worker):

    '''graph node worker'''

    # graph source
    _db = app(conf.key.backend, conf.userspace)
    # model finder
    _finder = app(conf.model.finder.node, conf.appspace)
    # model query
    _query = app(conf.model.node.reader, conf.appspace)
    # model user
    _using = app(conf.model.node.writer, conf.appspace)
    # node reader
    _r = factory(conf.read.node, backends, db)
    # node writer
    _w = factory(conf.write.node, backends, db)
    # direct
    direct = factory(conf.direct.worker.node, conf.direct, 'session')

    def _autoslug(self, properties):
        '''
        auto slug an element

        @param properties: graph element properties
        '''
        L = self.model.L
        slug_from = L.slug_from
        if not slug_from:
            return
        slug = tmpslug = self.Q.slugify(properties[slug_from])
        slug_field = L.slug_field
        # loop until unique slug is located
        filter_by = self._r.filter_by
        index = L.index
        for cnt in count(1):
            prev = filter_by(index, slug_field, tmpslug)
            if not prev:
                break
            tmpslug = '{slug}-{count}'.format(slug=slug, count=cnt)
        # add slug
        properties[slug_field] = tmpslug

    def _prepare(self, this):
        # snapshot if flag set
        if this.L.versioned:
            self.query(self.model).snapshot(this)
        super(Nodes, self)._prepare(this)

    @defer
    def _create(self, data):
        '''
        create a new graph node

        @param data: data
        '''
        model = self.model
        L = model.L
        data.update(_created=self.now(), _model=L.name, _uuid=self.Q.uuid())
        # number of versions to maintain
        if L.versioned:
            data['_versions'] = 0
        self._preprocess(data)
        thing = self._w.create(
            data,
            L.reference_link,
            self.query(model).reference(),
            partial(self._postprocess, model=model),
        )
        thing._refresh()

    def _preprocess(self, this):
        '''
        pre-process graph element

        @param this: graph object
        '''
        esc = escape
        for k in self.model.L.escaped:
            this[k] = esc(this[k])
        this['_modified'] = self.now()
        # slug field
        self._autoslug(this)

    def _postprocess(self, this):
        '''
        postprocess graph node

        @param this: graph object
        '''
        this = self.model(this)
        L = self.model.L
        index = L.index
        # index created date
        index_one = self.index_one
        if index:
            index_one(index, 'created', self.now(), this)
        # index any listed properties
        indexed = L.indexed
        if indexed:
            self.index_many(index, this, indexed)
        # add a slug if present
        slug_from = L.slug_from
        if slug_from:
            index_one(index, slug_from, getattr(this, slug_from), this)
        # index properties that support full text search
        fts_indexed = L.fts_indexed
        if fts_indexed:
            self.index_many(L.fts_index, this, fts_indexed)
        return this

    @defer
    def _update(self, this, data):
        '''
        update a graph element's properties

        @param this: graph object
        @param data: data
        '''
        self._preprocess(data)
        self._w.update(
            this, data, partial(self._postprocess, model=self.model)
        )
        this._refresh()

    def query(self, model):
        '''
        query using a graph model

        @param model: graph model
        '''
        return self._query(model)

    def using(self, model):
        '''
        use graph model

        @param model: graph model
        '''
        return self._using(model)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.