Source

appspace / appspace / registry.py

Full commit
# -*- coding: utf-8 -*-
'''appspace registries'''

import re
import uuid
import hashlib
from itertools import chain
from inspect import isclass, getmro
from unicodedata import normalize

from twoq import twoq
from stuf.six import u, strings, keys, items
from stuf.utils import exhaustcall, selfname

from appspace.utils import lazyimport, checkname
from appspace.support import appifies, AppStore, StrictAppStore
from appspace.keys import (
    ALazyLoad, InterfaceClass, AApp,  ANamespace, AManager, AppLookupError,
    AService, ConfigurationError)


@appifies(ALazyLoad)
class LazyLoad(object):

    '''lazy import loader'''

    __slots__ = ['_path']

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

        @param path: path to component
        '''
        self._path = path

    def __repr__(self):
        return 'lazy import from "{}"'.format(self.path)

    @property
    def path(self):
        module = self._path
        app = lazyimport(module[-1]) if isinstance(
            module, tuple
        ) else lazyimport(module)
        return app, next(list(keys(app)))


class RegistryMixin(object):

    def __init__(self, label, key=AApp):
        '''
        init

        @param label: label for internal namespace
        @param key: registry key (default: AApp)
        '''
        super(RegistryMixin, self).__init__()
        self._key = key
        # root and current label
        self._root = self._current = label
        # register key under namespace
        self.ez_register(ANamespace, label, key)
        # register manager under label
        self.ez_register(AManager, label, self)

    def _lazy(self, thing):
        return LazyLoad(thing) if isinstance(
            thing, (strings, tuple)
        ) else thing

    @classmethod
    def create(cls):
        '''create new key'''
        return InterfaceClass(cls.uuid())

    @staticmethod
    def ez_id(this):
        '''
        easy unique identifier for an object

        @param this: an object
        '''
        return hashlib.sha1(u(id(this))).digest()

    def ez_lookup(self, key, label):
        '''
        streamlined app lookup

        @param key: key to lookup
        @param label: label to lookup
        '''
        app = self.lookup1(key, key, label)
        return self.load(
            label, key, app.path
        ) if self.keyed(ALazyLoad, app) else app

    def ez_register(self, key=None, label=None, app=None):
        '''
        streamlined app registration

        @param key: key to register
        @param label: label to register
        @param app: app to register
        '''
        self.register([key], key, label, app)

    def ez_subscribe(self, key, label, app):
        '''
        streamlined app subscription

        @param key: key to extend to
        @param label: label to extend to
        '''
        self.subscribe(key, self.key(key, label), app)

    def ez_unregister(self, key, label):
        '''
        streamlined app unregistration

        @param key: key to lookup
        @param label: label to lookup
        '''
        self.unregister([key], key, label, self.ez_lookup(key, label))

    unkey = ez_unregister

    def ez_unsubscribe(self, key, label):
        '''
        streamlined app unsubscription

        @param key: key to lookup
        @param label: label to lookup
        '''
        self.unsubscribe(key, self.ez_lookup(key, label))

    @staticmethod
    def keyed(k=False, v=False):
        '''
        check if item has an app key

        @param key: app key
        @param thing: thing to check
        '''
        try:
            return k.implementedBy(v) if isclass(v) else k.providedBy(v)
        except AttributeError:
            return False

    def key(self, key, label):
        '''
        create or fetch key

        @param key: key to register
        @param label: label to register
        '''
        this = self.lookup1(key, key, label)
        if this is None:
            this = self.create()
            self.register([key], key, label, this)
        return this

    def load(self, label, key, thing):
        '''
        import thing into appspace

        @param label: appspaced thing label
        @param key: appspace key
        @param thing: some thing
        '''
        if not self.keyed(ALazyLoad, thing):
            return thing
        app, appkey = thing.path
        key = key if not appkey else appkey
        if appkey.isOrExtends(ANamespace):
            ez_register = self.ez_register
            ez_register(ANamespace, label, key)
            exhaustcall(
                lambda x: ez_register(AService, x, label),
                iter(key.names(True)),
            )
        self.register([key], key, label, app)
        return app

    safename = staticmethod(checkname)

    @staticmethod
    def uuid():
        '''universal unique identifier'''
        return uuid.uuid4().hex.upper()

    def pack(self, label, call):
        '''
        pack things into registry

        @param label: event label
        @param call: some thing
        '''
        self.ez_register(self._key, label, self._lazy(call))

    def update(self, *appconfs):
        '''
        update registry with appconfs

        @param *appconfs: sequence of appconfs
        '''
        def process(k, appconf):
            for k, v in items((twoq(getmro(appconf))
                .alt(isclass).wrap(dict).extract().value())):
                if isinstance(v, (tuple, strings)):
                    self.pack(k, v)
                elif isinstance(v, dict):
                    process(k, v)
        exhaustcall(
            lambda x: process(selfname(x), x), chain.from_iterable(appconfs),
        )


class ManagerMixin(object):

    '''state manager'''

    _first = staticmethod(re.compile('[^\w\s-]').sub)
    _second = staticmethod(re.compile('[-\s]+').sub)

    def apply(self, label, key=False, *args, **kw):
        '''
        invoke appspaced call

        @param label: appspaced call
        @param key: key label (default: False)
        '''
        return self.get(label, key)(*args, **kw)

    def get(self, label, key=False):
        '''
        get thing from appspace

        @param label: appspaced thing label
        @param key: `appspace` key (default: False)
        '''
        # use internal key if key label == internal key
        key = self._key if key == self._root else self.namespace(key)
        app = self.lookup1(key, key, label)
        if app is None:
            raise AppLookupError(app, label)
        return self.load(label, key, app)

    def namespace(self, label):
        '''
        fetch key

        @param label: `appspace` key label
        '''
        this = self.lookup1(ANamespace, ANamespace, label)
        if this is None:
            raise AppLookupError(this, label)
        return this

    def set(self, thing=False, label=False, key=False):
        '''
        add thing to `appspace`

        @param thing: new `appspace` thing (default: False)
        @param label: new `appspace` thing label (default: False)
        @param key: key label (default: False)
        '''
        thing = self._lazy(thing)
        key = self.namespace(key) if key else self._key
        self.register([key], key, self.safename(label), thing)
        return thing

    @classmethod
    def slugify(cls, value):
        '''
        normalizes string, converts to lowercase, removes non-alpha characters,
        and converts spaces to hyphens
        '''
        return cls._second('-', u(cls._first(
            '', normalize('NFKD', value).encode('ascii', 'ignore')
        ).strip().lower()))

    def service(self, label):
        '''
        fetch service

        @param label: service label
        '''
        service = self.lookup1(AService, AService, label)
        if service is None:
            raise ConfigurationError(label)
        return service


class Registry(RegistryMixin, AppStore):

    '''easy registry'''

    __slots__ = ('_current', '_root', '_key')


class StrictRegistry(RegistryMixin, StrictAppStore):

    '''strict registry'''

    __slots__ = ('_current', '_root', '_key')


@appifies(AManager)
class Manager(ManagerMixin, Registry):

    '''state manager'''

    __slots__ = ('_current', '_root', '_key')


@appifies(AManager)
class StrictManager(ManagerMixin, StrictRegistry):

    '''strict manager'''

    __slots__ = ('_current', '_root', '_key')

keyed = Manager.keyed