Source

appspace / appspace / registry.py

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

import re
import uuid
import hashlib
from inspect import isclass
from keyword import iskeyword
from unicodedata import normalize
from importlib import import_module

from stuf.six import u, strings
from stuf.utils import exhaustmap

from appspace.keys import (
    ALazyLoad, AppStore, InterfaceClass, AApp, StrictAppStore, ANamespace,
    AManager, AppLookupError, appifies, keys, ConfigurationError)
from callchain.services.queue import KService


def lazyimport(path, attribute=None):
    '''
    deferred module loader

    @param path: something to load
    @param attribute: attribute on loaded module to return
    '''
    if isinstance(path, strings):
        try:
            dot = path.rindex('.')
            # import module
            path = getattr(import_module(path[:dot]), path[dot + 1:])
        # If nothing but module name, import the module
        except (AttributeError, ValueError):
            path = import_module(path)
        if attribute:
            path = getattr(path, attribute)
    return path


class CheckName(object):

    '''ensures string is legal Python name'''

    # Illegal characters for Python names
    ic = '()[]{}@,:`=;+*/%&|^><\'"#\\$?!~'

    def __call__(self, name):
        '''
        ensures string is legal python name

        @param name: name to check
        '''
        # Remove characters that are illegal in a Python name
        name = name.strip().lower().replace('-', '_').replace(
            '.', '_'
        ).replace(' ', '_')
        name = ''.join(i for i in name if i not in self.ic)
        # Add _ if value is a Python keyword
        return name + '_' if iskeyword(name) else name


checkname = CheckName()


@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
        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, labels):
        '''
        update event with other registries

        @param labels: eventconf
        '''
        exhaustmap(vars(labels), self.pack, lambda x: not x[0].startswith('_'))


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(KService, KService, 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
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.