PK};Hqʪimagination/__init__.py__version__ = (1, 5, 6)PKg, .. versionadded:: 1.6 .. warning:: experimental feature """ def __init__(self, loader): self.__loader = loader def __call__(self, *args, **kwargs): return self.__loader.package class Loader(object): """Package loader with lazy loading *path_to_package* is a string representing the package path. For example:: # In this case, we load the default renderer of Tori framework. loader = Loader('tori.renderer.DefaultRenderer') # Then, instantiate the default renderer. renderer = loader.package('app.views') """ def __init__(self, path_to_package): self._path = path_to_package self._access_path = re.split('\.', self._path) self._module_path = '.'.join(self._access_path[:-1]) self._module = None self._package_name = self._access_path[-1] self._package = None self.on_demand_package = OnDemandProxy(self) @property def name(self): ''' Get the name of the package. ''' return self.module.__package__ @property def module(self): ''' Get a reference to the module. ''' if not self._module: self._module = retrieve_module(self._module_path) return self._module @property def package(self): ''' Get a reference to the package. ''' if not self._package: self._package = self._retrieve_package() return self._package @property def filename(self): ''' Get the path to the package. ''' return self.module.__file__ def _retrieve_package(self): ''' Retrieve a package by the module path and the package name. ''' target_module = self.module # Either built-in class or a file. if '.' not in self._path: try: importlib.import_module(self._path) return getattr(target_module, self._path) except ImportError as exception: return eval(self._path) # NOP; assume that the request path is a built-in module. return try: __import__(self._module_path, fromlist=[self._package_name]) except TypeError as exception: raise ImportError('Unable to import {}.{} as {}'.format( self._module_path, self._package_name, exception )) try: return getattr(target_module, self._package_name) except AttributeError as exception: raise ImportError('Module \'{}\' has no reference to \'{}\', except {}.'.format( target_module.__name__, self._package_name, ', '.join(dir(target_module)) )) PK};HIxRRimagination/proxy.py''' :Author: Juti Noppornpitak :Availability: 1.5 The module contains the package entity used to be an intermediate between :class:`imagination.locator.Locator` and :class:`imagination.loader.Loader` and simulate the singleton class on the package in the Loader. .. note:: Copyright (c) 2012 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' class Proxy(object): ''' Proxy to a particular entity known by a certain locator. :param `locator`: the locator :param `id`: entity identifier ''' def __init__(self, locator, id): self.__locator = locator self.__id = id @property def id(self): ''' Get the identifier of the proxy. ''' return self.__id def load(self): ''' Load the entity. ''' return self.__locator.get(self.__id) PK};HeGGimagination/locator.py''' :Author: Juti Noppornpitak The module contains the entity locator used to promote reusability of components. .. note:: Copyright (c) 2012 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' from re import split from kotoba import load_from_file from imagination.factorization import Factorization from imagination.entity import CallbackProxy from imagination.entity import Entity from imagination.exception import * from imagination.proxy import Proxy class Locator(object): ''' Entity locator ''' def __init__(self): self._entities = {} self._tag_to_entity_ids = {} self._in_passive_mode = False @property def in_passive_mode(self): return self._in_passive_mode @in_passive_mode.setter def in_passive_mode(self, value): if not isinstance(value, bool): raise ValueError('Expected a boolean value for in_passive_mode.') self._in_passive_mode = value def fork(self, id): ''' Fork the entity identified by ``id``. :param `id`: entity identifier ''' try: entity = self._entities[id] # Prevent the locator from forking a new instance of Proxy or CallbackProxy. if isinstance(entity, CallbackProxy) or isinstance(entity, Proxy): raise ForbiddenForkError('Unable to fork {}'.format(type(entity).__name__)) return entity.fork() except KeyError: raise UnknownEntityError('The requested entity named "%s" is unknown or not found.' % id) @property def entity_identifiers(self): return self._entities.keys() def get(self, id): ''' Retrieve an instance of the entity :param `id`: entity identifier :returns: an instance of the requested entity ''' try: entity = self.get_wrapper(id) # Retrieve a callback entity (callable). if isinstance(entity, CallbackProxy): #print('GET {} "{}" [Done]'.format(type(entity).__name__, id)) return entity() # Retrieve a proxy object to an entity if isinstance(entity, Proxy): #print('GET {} "{}" [Done]'.format(type(entity).__name__, id)) return entity #return the proxy reference. if isinstance(entity, Factorization): #print('GET {} "{}" [Done]'.format(type(entity).__name__, id)) return entity.fork() #print('GET {} "{}" [Done]'.format(type(entity).__name__, id)) return entity.instance if isinstance(entity, Entity) else entity except UnknownEntityError as e: raise UnknownEntityError( 'The requested entity named "{id}" is unknown or not found. This locator only knows {known_keys}. (Reason: {original})'.format( id = id, known_keys = ', '.join(self._entities.keys()), original = e ) ) def get_wrapper(self, id): ''' Retrieve the entity wrapper identified by ``id``. :param `id`: entity identifier :returns: the requested entity wrapper ''' try: return self._entities[id] except KeyError: if self.in_passive_mode: return Proxy(self, id) raise UnknownEntityError('The requested entity named "%s" is unknown or not found.' % id) def find_by_tag(self, tag_label): ''' Retrieve entities by *tag_label*. :param `tag_label`: tag label ''' # First, get the entity identifiers. if tag_label not in self._tag_to_entity_ids: return [] # Then, get references to entities. return [self.get(entity_id) for entity_id in self._tag_to_entity_ids[tag_label]] def set(self, id, entity): ''' Set the given *entity* by *id*. ''' is_proxy = isinstance(entity, Proxy) is_callback = isinstance(entity, CallbackProxy) if type(entity) not in [Entity, Proxy, CallbackProxy, Factorization]: raise UnknownEntityError('The type of the given entity named "{}" is not excepted. ({})'.format(id, type(entity).__name__)) if isinstance(entity, Entity): entity.lock() if is_proxy and id in self._entities: #print('SET Proxy "{}" [Ignored]'.format(id)) return self._entities[id] = entity if is_proxy: #print('SET Proxy "{}" [Done]'.format(id)) return if is_callback: #print('SET Callback "{}" [Done]'.format(id)) return for tag in entity.tags: if tag not in self._tag_to_entity_ids: self._tag_to_entity_ids[tag] = [] self._tag_to_entity_ids[tag].append(entity.id) #print('SET {} "{}" [Done]'.format(type(entity).__name__, id)) def has(self, id): ''' Check if the entity with *id* is already registered. ''' return id in self._entities def load_xml(self, file_path): ''' Load the entities from a XML configuration file at *file_path*. :Status: Deprecated in 1.5 ''' raise DeprecatedAPI('Use imagination.helper.assembler.Assembler.load instead.') PK};HWWimagination/exception.pyclass DeprecatedAPI(Exception): ''' Exception thrown when a deprecated API is used. ''' class MisplacedValidatorError(Exception): ''' Exception thrown when a validator is used with type (e.g., class, perimitive types etc.) ''' class ForbiddenForkError(Exception): ''' Exception thrown only when a locator try to fork a proxy. ''' class DuplicateKeyWarning(Warning): ''' Warning thrown when an internal dictionary detects an attempt of re-assignment to the direction by the existed key. ''' class MultipleInterceptingEventsWarning(Warning): ''' Warning thrown when more than one intercepting event is detected. ''' class IncompatibleBlockError(Exception): ''' Exception thrown when :meth:`imagination.locator.Locator.load_xml` cannot process the entity block. ''' class UnknownFileError(Exception): ''' Exception thrown when :meth:`imagination.locator.Locator.load_xml` cannot locate the file on the given path. ''' class UnknownEntityError(Exception): ''' Exception thrown when :class:`imagination.locator.Locator` constructor receives an unusable entity. ''' class UnknownProxyError(Exception): ''' Exception thrown when :class:`imagination.helper.assembler.Assembler` could not find the proxy. ''' class NonCallableError(Exception): ''' Exception thrown when :class:`imagination.action.Action` tries to execute a non-callable reference. ''' class LockedEntityException(Exception): ''' Exception thrown when a caller attempts to update any of alterable properties of an instance of :class:`imagination.entity.Entity` when it is in the locked state. '''PK};H7ccimagination/entity.py''' :Author: Juti Noppornpitak The module contains the package entity used to be an intermediate between :class:`imagination.locator.Locator` and :class:`imagination.loader.Loader` and simulate the singleton class on the package in the Loader. .. note:: Copyright (c) 2012 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' from imagination.common import InterceptableObject from imagination.decorator.validator import restrict_type from imagination.loader import Loader from imagination.proxy import Proxy class CallbackProxy(object): """Callback Proxy .. codeauthor:: Juti Noppornpitak .. versionadded:: 1.6 .. warning:: experimental feature """ def __init__(self, callback, *args, **kwargs): if not callable(callback): raise ValueError('The callback object is required for {}.'.format(self.__class__.__name__)) self.__callback = callback self.__args = args self.__kwargs = kwargs def __call__(self): return self.__callback(*self.__args, **self.__kwargs) class Entity(InterceptableObject): ''' Entity represents the package, reference and instance of the reference. :param `id`: the service identifier (string). :param `loader`: a service loader which is an instance of :class:`imagination.loader.Loader`. :param `args`: constructor's parameters :type args: list or tuple :param `kwargs`: constructor's parameters :type args: dict If the loader is not an instance of :class:`imagination.loader.Loader` or any classes based on :class:`imagination.loader.Loader`, the exception :class:`imagination.exception.UnknownLoaderError` will be thrown. .. note:: This class is to similar to the decorator `tori.decorator.common.singleton` and `tori.decorator.common.singleton_with` from Tori Framework, except that it is not a singleton class and so any arbitrary class referenced by the *loader* only lasts as long as the entity lives. .. note:: In version 1.5, the entity has the ability to fork an non-supervised instance of the reference. ''' @restrict_type(None, Loader) def __init__(self, id, loader, *args, **kwargs): super(Entity, self).__init__() self._id = id self._loader = loader self._args = args self._kwargs = kwargs self._instance = None self._tags = [] self._prepared = False @property def id(self): ''' Entity ID :rtype: strint or unicode or integer ''' return self._id @property def loader(self): ''' Package loader :rtype: imagination.loader.Loader ''' return self._loader @property def argument_list(self): ''' Get the argument list. ''' return self._args @property def argument_dictionary(self): ''' Get the argument dictionary. ''' return self._kwargs @property def activated(self): ''' Check if the entity is already activated. This will also inspects if the entity already loads a singleton instance into the memory. ''' return self._instance is not None @property def tags(self): ''' Retrieve the entity tags. :rtype: list ''' return self._tags @tags.setter @restrict_type(list) def tags(self, tags): ''' Define the entire entity tags. :param tags: new tags as replacements :type tags: list or tuple ''' if self.locked: raise LockedEntityException self._tags = tags @property def instance(self): ''' Get the singleton instance of the class defined for the loader. ''' if not self._instance: self._instance = self.fork() return self._instance def fork(self): ''' :Version: 1.5 Fork an instance of the class defined for the loader. ''' self.__prepare() instance = self._loader.package(*self._args, **self._kwargs) # Return the instance if this entity is not interceptable. if not self.interceptable: return instance # For each PUBLIC method, make it interceptable with Action. self._bind_interceptions(instance, self._interceptions) return instance def __prepare(self): if self._prepared or self._instance: return for i in range(len(self._args)): if isinstance(self._args[i], Proxy): self._args[i] = self._args[i].load() for i in self._kwargs: if isinstance(self._kwargs[i], Proxy): self._kwargs[i] = self._kwargs[i].load() self._prepared = True PK};H  imagination/common.pyfrom imagination.action import Action from imagination.decorator.validator import restrict_type class InterceptableObject(object): def __init__(self): self._locked = False self._interceptable = True self._interceptions = [] def lock(self): """ Lock the object """ self._locked = True def unlock(self): """ Unlock the object """ self._locked = True @property def locked(self): """ Flag indicating whether the object is locked :rtype: bool """ return self._locked @property def interceptable(self): ''' Flag if this entity is interceptable :rtype: boolean ''' return self._interceptable @interceptable.setter @restrict_type(bool) def interceptable(self, interceptable): ''' Define if this entity is interceptable. :param interceptable: Flag if this entity is interceptable :type interceptable: boolean ''' if self.locked: raise LockedEntityException self._interceptable = interceptable @property def interceptions(self): ''' Retrieve the list of interceptions. ''' return self._interceptions @interceptions.setter @restrict_type(list) def interceptions(self, interceptions): ''' Define the list of interceptions. :param interceptions: list of interceptions :type interceptions: list ''' self._interceptions = interceptions def register_interception(self, interception): self._interceptions.append(interception) def _bind_interceptions(self, instance, interceptions): for attribute in dir(instance): if attribute[0] == '_': continue ref = instance.__getattribute__(attribute) if not callable(ref): continue new_ref = Action(ref) for interception in interceptions: if interception.actor.method_name != attribute: continue new_ref.register(interception) try: instance.__setattr__(attribute, new_ref) except AttributeError: pass PK};H6timagination/factorization.py''' :Author: Juti Noppornpitak :Availability: 1.9 The module contains the package entity used to be an intermediate between :class:`imagination.locator.Locator` and :class:`imagination.loader.Loader` and simulate the singleton class on the package in the Loader. .. note:: Copyright (c) 2015 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' from imagination.decorator.validator import restrict_type from imagination.common import InterceptableObject from imagination.proxy import Proxy class NonCallableFactoryMethod(RuntimeError): """ The factory method is not callable. """ class NotReadyError(RuntimeError): """ When the factory service is not ready to use. """ class Factorization(InterceptableObject): """ Factorization Entity :param imagination.locator.Locator locator: the locator :param str factory_id: the factory entity identifier :param str factory_method: the factory method .. versionadded:: 1.9 """ def __init__(self, locator, factory_id, factory_method, parameters): super(Factorization, self).__init__() self._locator = locator self._factory_id = factory_id self._factory_method = factory_method self._parameters = parameters self._reference = None self._tags = [] @property def id(self): ''' Get the identifier of the proxy. ''' return self.__id @property def tags(self): ''' Retrieve the entity tags. :rtype: list ''' return self._tags @tags.setter @restrict_type(list) def tags(self, tags): ''' Define the entire entity tags. :param tags: new tags as replacements :type tags: list or tuple ''' if self.locked: raise LockedEntityException self._tags = tags def fork(self): ''' Fork the entity. ''' self._prepare() factory = self._locator.get(self._factory_id) if isinstance(factory, Proxy): raise NotReadyError(self._factory_id) factory_method = factory.__getattribute__(self._factory_method) if not factory_method or not callable(factory_method): raise NonCallableFactoryMethod('{}.{} not available ({})'.format(self._factory_id, self._factory_method, type(factory_method))) self._reference = factory_method(*self._parameters.largs, **self._parameters.kwargs) if self._interceptable and isinstance(self._reference, object): self._bind_interceptions(self._reference, self._interceptions) return self._reference def _prepare(self): p = self._parameters for i in range(len(p.largs)): if isinstance(p.largs[i], Proxy): p.largs[i] = p.largs[i].load() for i in p.kwargs: if isinstance(p.kwargs[i], Proxy): p.kwargs[i] = p.kwargs[i].load() PK};H_imagination/action.py# -*- coding: utf-8 -*- ''' :Author: Juti Noppornpitak :Version: 1.5 :Usage: Internal The module contains the classes used for interceptable actions/methods of any instance of :class:`imagination.entity.Entity`. .. note:: Copyright (c) 2012 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' from imagination.decorator.validator import restrict_type, SpecialType from imagination.meta.interception import Interception from imagination.meta.package import Parameter class EventType(object): ''' Type of intercepting events ''' pre_action = 'before' ''' Event before the execution that doesn't care about the input given to the action. ''' pre_condition = 'pre' ''' Event before the execution that only concerns about the input given to the action. ''' post_condition = 'post' ''' Event after the execution that only concerns about the returned value from the action. ''' post_action = 'after' ''' Event after the execution that doesn't care about the returned value from the action. ''' class Action(object): ''' Method wrapper for intercepting actions ''' @restrict_type(SpecialType.function) def __init__(self, f): self.__doc__ = f.__doc__ self.__name__ = f.__name__ self._reference = f self.__pre_actions = [] self.__post_actions = [] def __call__(self, *args, **kwargs): parameters = Parameter(args, kwargs) self.__run_pre_events(parameters) feedback = self.reference(*parameters.largs, **parameters.kwargs) feedback = self.__run_post_events(feedback) return feedback @property def name(self): ''' Name of the action ''' return self.__name__ @property def reference(self): ''' Reference of the action ''' return self._reference @restrict_type(Interception) def register(self, interception): event = str(interception.event) if event in [EventType.pre_action, EventType.pre_condition]: self.__pre_actions.append(interception) return if event in [EventType.post_action, EventType.post_condition]: self.__post_actions.append(interception) return raise ValueError('The event "%s" is not recognized.' % event) @restrict_type(Interception) def __retrieve_callback(self, interception): callback = interception.handler.interface # cache callback? if not callable(callback): raise NonCallableError('%s.%s is not callable' % ( interception.handler.id, interception.handler.method_name )) return callback @restrict_type(Parameter) def __run_pre_events(self, parameters): for interception in self.__pre_actions: callback = self.__retrieve_callback(interception) if interception.event == EventType.pre_action: interception.handler.engage() continue if interception.event == EventType.pre_condition: callback(*parameters.largs, **parameters.kwargs) continue raise ValueError('The event "%s" is not recognized.' % interception.event) def __run_post_events(self, feedback): for interception in self.__post_actions: callback = self.__retrieve_callback(interception) if interception.event == EventType.post_action: interception.handler.engage() continue if interception.event == EventType.post_condition: feedback = callback(feedback) continue raise ValueError('The event "%s" is not recognized.' % interception.event) return feedback PK};Himagination/meta/__init__.pyPK};H)*imagination/meta/aspect.pyfrom imagination.meta.package import Parameter from imagination.proxy import Proxy class Contact(object): def __init__(self, proxy, method_name, parameters=None): self._proxy = proxy self._method_name = method_name self._parameters = parameters @property def id(self): return self._proxy.id @property def method_name(self): return self._method_name @property def interface(self): return self._proxy.load().__getattribute__(self._method_name) @property def parameters(self): return self._parameters def engage(self): return self.interface(*self._parameters.largs, **self._parameters.kwargs)PK};H}ȿimagination/meta/package.py''' :Author: Juti Noppornpitak :Availability: 1.5 :Usage: Internal The module contains the data structure of method interception to aid the analysis and construction of the loaders and entities based on the configuration. .. note:: Copyright (c) 2012 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' from imagination.decorator.validator import restrict_type class Interceptor(object): ''' The metadata for an interceptor. ''' def __init__(self): self._id = None self._action = None self._parameters = None class Parameter(object): ''' Parameter Package represents the parameter of arguments as a list and a dictionary to any callable objects (e.g., constructor and methods). :param `largs`: a list of arguments :param `kwargs`: a dictionary of arguments ''' @restrict_type(list, dict) def __init__(self, largs=None, kwargs=None): self.largs = largs or [] self.kwargs = kwargs or {}PK};H  imagination/meta/interception.py''' :Author: Juti Noppornpitak :Availability: 1.5 :Usage: Internal The module contains the data structure of method interception to aid the analysis and construction of the loaders and entities based on the configuration. .. note:: Copyright (c) 2012 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' from imagination.decorator.validator import restrict_type from imagination.meta.aspect import Contact class Interception(object): ''' .. note:: Please note that both `Interception.actor` and `Interception.handler` are always instances of :class:`imagination.proxy.Proxy`. This is intentional in order to avoid direct circular dependencies. ''' static_guid = 1 @restrict_type(str, Contact, Contact) def __init__(self, event, actor, handler): self._guid = Interception.static_guid self._actor = actor self._event = event self._handler = handler Interception.static_guid += 1 @property def actor(self): return self._actor @property def event(self): return self._event @property def handler(self): return self._handler @staticmethod def self_reference_keyword(): return 'me'PK};H'~~imagination/helper/data.pyfrom kotoba.kotoba import Kotoba from imagination.decorator.validator import restrict_type from imagination.exception import * from imagination.loader import Loader from imagination.locator import Locator class Transformer(object): """ Data transformer .. versionadded: 1.5 :param imagination.locator.Locator locator: the entity locator """ @restrict_type(Locator) def __init__(self, locator): assert locator and isinstance(locator, Locator), "Expecting an instance of imagination.locator.Locator, one of %s was given instead." % (type(locator).__name__) self.__locator = locator def cast(self, data, kind): ''' Transform the given data to the given kind. :param data: the data to be transform :param str kind: the kind of data of the transformed data :return: the data of the given kind ''' actual_data = data.data() if isinstance(data, Kotoba) else data if kind == 'entity': actual_data = self.__locator.get(actual_data) elif kind == 'class': actual_data = Loader(actual_data).package elif kind == 'int': actual_data = int(actual_data) elif kind == 'float': actual_data = float(actual_data) elif kind == 'bool': actual_data = actual_data.capitalize() assert actual_data == 'True' or actual_data == 'False' actual_data = actual_data == 'True' elif kind in ['list', 'tuple', 'set']: assert isinstance(data, Kotoba), 'Asking for a Kotoba object, getting an instance of type {}.'.format(type(data).__name__) actual_data = [] for item in data.children(): item_type = item.attribute('type') actual_data.append(self.cast(item, item_type)) if kind != 'list': actual_data = eval(kind)(actual_data) elif kind == 'dict': assert isinstance(data, Kotoba), 'Asking for a Kotoba object, getting an instance of type {}.'.format(type(data).__name__) actual_data = {} for item in data.children(): item_name = item.attribute('name') item_type = item.attribute('type') actual_data[item_name] = self.cast(item, item_type) elif kind == 'str': actual_data = str(actual_data) elif kind not in ['str', 'unicode']: raise ValueError('Unknown type: {} (Given data: {})'.format(kind, data)) return actual_data def locator(self): return self.__locator class Interception(object): """ Event Interception :param str event: the event type :param str actor: the ID of the actor :param str intercepted_action: the intercepted method :param str handler: the ID of the handler (interceptor) :param str handling_action: the handing (intercepting) method """ __self_reference_keyword = 'me' def __init__(self, event, actor, intercepted_action, handler, handling_action): self.actor = actor == self.__self_reference_keyword and handler or actor self.event = event self.handler = handler self.intercepted_action = intercepted_action self.handling_action = handling_action def __str__(self): return 'Interception: %s %s.%s, %s.%s' % ( self.event, self.actor, self.intercepted_action, self.handler, self.handling_action ) def __unicode__(self): return u'Interception: %s %s.%s, %s.%s' % ( self.event, self.actor, self.intercepted_action, self.handler, self.handling_action ) class ParameterPackage(object): ''' Parameter Package represents the parameter of arguments as a list and a dictionary to any callable objects (e.g., constructor and methods). :param list largs: a list of arguments :param dict kwargs: a dictionary of arguments ''' def __init__(self, largs=None, kwargs=None): self.largs = largs or [] self.kwargs = kwargs or {} PK};HY  imagination/helper/__init__.pyimport os import sys def retrieve_module(name): if name not in sys.modules: __import__(name) return sys.modules[name] def retrieve_module_path(name): module = retrieve_module(name) return os.path.dirname(module.__file__) if module else NonePK};H˞p6(6(imagination/helper/assembler.pyfrom re import split from kotoba.kotoba import Kotoba from kotoba import load_from_file from imagination.decorator.validator import restrict_type from imagination.entity import Entity from imagination.factorization import Factorization, NotReadyError from imagination.exception import * from imagination.loader import Loader from imagination.helper.data import * from imagination.meta.aspect import Contact from imagination.meta.interception import Interception from imagination.meta.package import Parameter from imagination.proxy import Proxy class Assembler(object): """ The entity assembler via configuration files. :param imagination.helper.data.Transformer transformer: the data transformer .. versionadded:: 1.5 This class is added. .. versionchanged:: 1.9 Added the support for entity factorization. .. versionchanged:: 1.10 Added the support for delayed injections when the factory service is not ready to use during the factorization. """ __known_interceptable_events = ['before', 'pre', 'post', 'after'] __entity_element_required_attributes = { 'entity': { 'id': 'entity identifier', 'class': 'entity class', }, 'factorization': { 'id': 'factorized entity identifier', 'with': 'factory entity identifier', 'call': 'factory method', }, } @restrict_type(Transformer) def __init__(self, transformer): self.__interceptions = [] self.__transformer = transformer self.__known_proxies = {} self.__delay_injection_map = {} def activate_passive_loading(self): """ Activate the passive loading mode. This is to delay the instantiation or forking of an entity to when it is referred/queried but undefined. This assumes that the queried entity will later be defined. If later on it isn't defined, the exception will be thrown. When this method is used, please do not forget to call :meth:`deactivate_passive_loading` to disable the passive mode. .. versionadded:: 1.7 """ self.locator.in_passive_mode = True def deactivate_passive_loading(self): """ Deactivate the passive loading mode. When :meth:`activate_passive_loading` is used, this method must be called to disable the passive mode. .. versionadded:: 1.7 """ if self.__delay_injection_map: injection_order = [] for k in self.__delay_injection_map: di = self.__delay_injection_map[k] n = di['node'] p = di['ping'] r = di['ready'] if not n or r: # Skip if either the node is not defined or the delay injection is done for this entry. continue di['ready'] = True injection_order.append((k, di['node'], di['ping'])) injection_order.sort(key=lambda i: i[2]) for di in injection_order: id, node, ping = di self.__get_interceptions(node) self.__register_entity(node) self.__activate_interceptions() self.locator.in_passive_mode = False def load(self, filepath): """ Load the configuration. :param str filepath: the file path to the configuration. """ xml = load_from_file(filepath) # First, register proxies to entities (for lazy initialization). for node in xml.children(): self.__validate_node(node) self.__register_proxy(node) # Then, register loaders for entities. for node in xml.children(): self.__get_interceptions(node) self.__register_entity(node) self.__activate_interceptions() def __activate_interceptions(self): # Then, declare interceptions to target entities. for interception in self.__interceptions: self.locator\ .get_wrapper(interception.actor.id)\ .register_interception(interception) self.__interceptions = [] @property def locator(self): """ The injected locator via the data transformer. :rtype: imagination.locator.Locator """ return self.__transformer.locator() @restrict_type(Kotoba) def __validate_node(self, node): entity_type = node.name() required_attributes = None if entity_type not in self.__entity_element_required_attributes: raise IncompatibleBlockError('Entity class "{}" is undefined.'.format(entity_type)) required_attributes = self.__entity_element_required_attributes[entity_type] for name in required_attributes: if node.attribute(name): continue raise IncompatibleBlockError('Entity class "{}" needs a valid "{}" attribute.'.format(entity_type, required_attributes[name])) @restrict_type(Kotoba) def __register_proxy(self, node): id = node.attribute('id') proxy = Proxy(self.locator, id) self.locator.set(id, proxy) # this is for interceptors self.__known_proxies[id] = proxy @restrict_type(Kotoba) def __register_entity(self, node): entity_type = node.name() try: if entity_type == 'entity': return self.__register_normal_entity(node) if entity_type == 'factorization': return self.__register_factorized_entity(node) except NotReadyError as e: required = str(e) targeted = node.attribute('id') self.__set_delay_injection(required, None, True) self.__set_delay_injection(targeted, node, False) #print('POSTPONED {} DUE TO {}'.format(targeted, required)) def __set_delay_injection(self, id, node, ping): if id not in self.__delay_injection_map: self.__delay_injection_map[id] = { 'ping': 0, 'node': node, 'ready': False, } if node: self.__delay_injection_map[id]['node'] = node if ping: self.__delay_injection_map[id]['ping'] -= 1 def __register_normal_entity(self, node): entity_id = node.attribute('id') kind = node.attribute('class') params = self.__get_params(node) tags = self.__get_tags(node) loader = Loader(kind) entity = Entity(entity_id, loader, *params.largs, **params.kwargs) entity.interceptable = self.__transformer.cast(node.attribute('interceptable') or 'true', 'bool') entity.tags = tags self.locator.set(entity_id, entity) def __register_factorized_entity(self, node): entity_id = node.attribute('id') factory_id = node.attribute('with') factory_method = node.attribute('call') params = self.__get_params(node) tags = self.__get_tags(node) entity = Factorization(self.locator, factory_id, factory_method, params) entity.interceptable = self.__transformer.cast(node.attribute('interceptable') or 'true', 'bool') entity.tags = tags self.locator.set(entity_id, entity) @restrict_type(Kotoba) def __get_tags(self, node): tags = node.attribute('tags') return tags and split(' ', tags.strip()) or [] @restrict_type(Kotoba) def __get_interceptions(self, node): for sub_node in node.children('interception'): self.__interceptions.append(self.__get_interception(sub_node)) @restrict_type(Kotoba) def __get_interception(self, node): actor = None event = None intercepted_action = None handling_action = None for given_event in self.__known_interceptable_events: given_actor = node.attribute(given_event) # If the actor is not defined, continue or if the event is already # set (in the earlier iteration), raise the exception. if not given_actor: continue elif event: raise MultipleInterceptingEventsWarning(given_event) # Initially get the name of the actor and the handler. actor = given_actor handler = node.parent().attribute('id') if actor == Interception.self_reference_keyword(): actor = handler # If the actor or the handler has no proxies, raise the exception. if actor not in self.__known_proxies: raise UnknownProxyError('The target ({}) of the interception is unknown.'.format(actor)) if handler not in self.__known_proxies: raise UnknownProxyError('The handler ({}) of the interception is unknown.'.format(handler)) actor = Contact(self.__known_proxies[actor], node.attribute('do')) handler = Contact(self.__known_proxies[handler], node.attribute('with'), self.__get_params(node)) event = given_event return Interception(event, actor, handler) @restrict_type(Kotoba) def __get_params(self, node): package = Parameter() index = 0 for param in node.children('param'): try: assert param.attribute('name')\ and param.attribute('type'),\ 'The parameter #{} does not have either name or type.'.format(index) except AssertionError as e: raise IncompatibleBlockError(e.message) index += 1 name = param.attribute('name') if name in package.kwargs: raise DuplicateKeyWarning('There is a parameter name "{}" with that name already registered.'.format(name)) pass package.kwargs[name] = self.__transformer.cast( param, param.attribute('type') ) return package PK};H!imagination/decorator/__init__.pyPK};HQM"imagination/decorator/validator.py''' :Author: Juti Noppornpitak :Availability: 1.5 The module contains reusable input validators. .. note:: Copyright (c) 2012 Juti Noppornpitak Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ''' try: from inspect import getfullargspec except ImportError as e: # Fall back to Python 2.7 mode from inspect import getargspec as getfullargspec import sys from imagination.exception import MisplacedValidatorError _disable_decorator = 'sphinx' in sys.modules class SpecialType(object): function = 'type:function' def restrict_type(*restricted_list, **restricted_map): ''' The method decorator to validate the type of inputs given to the method. :param `restricted_list`: the list of types to restrict the type of each parameter in the same order as the parameter given to the method. :param `restricted_map`: the map of types to restrict the type of each parameter by name. When the input fails the validation, an exception of type :class:`TypeError` is throw. There are a few exceptions: * If the given type is ``None``, there will be no restriction. * If the given type is ``long``, the value of ``int`` and ``float`` are also valid. * If the given type is ``unicode``, the valud of ``str`` is also valid. .. warning:: In Imagination 1.6, types ``unicode`` and ``long`` are no longer have fallback check in order to support Python 3.3. .. code-block:: python from imagination.decorator.validator import restrict_type # Example on a function @restrict_type(unicode) def say(context): print context class Person(object): # Example on a constructor @restrict_type(unicode, int) def __init__(self, name, age): self.name = name self.age = age self.__friends = [] # Example on an instance method @restrict_type(Person) def add_friend(self, person): self.__friends.append(person) ''' def inner_decorator(reference): if _disable_decorator: return reference if isinstance(reference, type) or not callable(reference): raise MisplacedValidatorError( 'Can only be used with callable objects, e.g., functions, class methods, instance methods and static methods.' ) params = getfullargspec(reference).args is_class = params and params[0] == 'self' if is_class: def new_reference(self, *args, **kwargs): __validate_type(restricted_list, restricted_map, args, kwargs) return reference(self, *args, **kwargs) else: def new_reference(*args, **kwargs): __validate_type(restricted_list, restricted_map, args, kwargs) return reference(*args, **kwargs) new_reference.__doc__ == reference.__doc__ return new_reference return inner_decorator def __validate_type(allowed_list, allowed_dictionary, argument_list, argument_dictionary): allowed_list = allowed_list[:len(argument_list)] reference = None try: for index in range(len(allowed_list)): expected_type = allowed_list[index] reference = argument_list[index] if not expected_type: continue assert __assert_type(reference, expected_type),\ expected_type == SpecialType.function\ and 'function'\ or expected_type.__name__ for key in allowed_dictionary: expected_type = allowed_dictionary[key] if key not in argument_dictionary: continue reference = argument_dictionary[key] if not expected_type: continue assert __assert_type(reference, expected_type),\ expected_type.__name__ except AssertionError as e: actual_type = e.message if 'message' in dir(e) else e raise TypeError('Argument #%s was excepting %s but %s has been given.' % (index, actual_type, type(reference).__name__)) def __assert_type(instance, expected_type): list_types = [list, tuple, set] callable_expected = expected_type == SpecialType.function instance_expected = isinstance(expected_type, type) if not callable_expected and not instance_expected: raise TypeError('The expected type must be a type.') fallback_types = [] # If the callable is expected, validate if it is callable. if callable_expected: return callable(instance) # Now, validate the type of INSTANCE. if expected_type in list_types: fallback_types.extend(list_types) if type(instance) == expected_type or isinstance(instance, expected_type): return True for fallback_type in fallback_types: if type(instance) == fallback_type: return True return False PK-imagination/helper/data.pyPK};HY  imagination/helper/__init__.pyPK};H˞p6(6(<imagination/helper/assembler.pyPK};H!imagination/decorator/__init__.pyPK};HQM"imagination/decorator/validator.pyPK-