Source code for anyblok.environment

# This file is a part of the AnyBlok project
#
#    Copyright (C) 2014 Jean-Sebastien SUZANNE <[email protected]>
#
# This Source Code Form is subject to the terms of the Mozilla Public License,
# v. 2.0. If a copy of the MPL was not distributed with this file,You can
# obtain one at http://mozilla.org/MPL/2.0/.
import threading
from inspect import ismethod


[docs]class EnvironmentException(AttributeError): """ Exception for the Environment """
[docs]class EnvironmentManager: """ Manage the Environment for an application """ environment = None
[docs] @classmethod def define_environment_cls(cls, Environment): """ Define the class used for the environment :param Environment: class of environment :exception: EnvironmentException """ def check_classmethod(method, acceptNone=False): if not hasattr(Environment, method): raise EnvironmentException("No %r found" % method) m = getattr(Environment, method) if m is None: if acceptNone: return if ismethod(m): return raise EnvironmentException("%r must be a class method" % m) check_classmethod('scoped_function_for_session', acceptNone=True) check_classmethod('setter') check_classmethod('getter') cls.environment = Environment
[docs] @classmethod def set(cls, key, value): """ Save the value of the key in the environment :param key: the key of the value to save :param value: the value to save :exception: EnvironmentException """ if cls.environment is None: raise EnvironmentException("No environments defined") cls.environment.setter(key, value)
[docs] @classmethod def get(cls, key, default=None): """ Load the value of the key in the environment :param key: the key of the value to load :param default: return this value if not value loaded for the key :rtype: the value of the key :exception: EnvironmentException """ if cls.environment is None: raise EnvironmentException("No environments defined") return cls.environment.getter(key, default)
[docs] @classmethod def scoped_function_for_session(cls): """ Save the value of the key in the environment """ return cls.environment.scoped_function_for_session
[docs]class ThreadEnvironment: """ Use the thread, to get the environment """ scoped_function_for_session = None """ No scoped function here because for none value sqlalchemy already uses a thread to save the session """ values = {}
[docs] @classmethod def setter(cls, key, value): """ Save the value of the key in the environment :param key: the key of the value to save :param value: the value to save """ if str(threading.current_thread()) not in cls.values: cls.values[str(threading.current_thread())] = {} cls.values[str(threading.current_thread())][key] = value
[docs] @classmethod def getter(cls, key, default): """ Get the value of the key in the environment :param key: the key of the value to retrieve :param default: return this value if no value loaded for the key :rtype: the value of the key """ if str(threading.current_thread()) not in cls.values: return default return cls.values[str(threading.current_thread())].get(key, default)
EnvironmentManager.define_environment_cls(ThreadEnvironment)