import threading from contextlib import contextmanager from typing import Any, Optional, Mapping class Context: ''' Context instances can be used to hold state and values during the execution life cycle of a python application without needing to change the existing API of various interacting classes. State/values can be updated in one part of the application and be visible in another part of the application without requiring to pass this state thru the various methods of the call stack. Context also supports maintenance of global and thread specific state which can be very useful in a multi-threaded applications like Flask/Falcon based web applications or the NES Journal reader. State stored at the global level is visible in every concurrently executed thread, while thread specific state is isolated to the corresponding thread of execution. This can useful again in a multi-threaded application like a web-app where each incoming request is processed by a separate thread and things like request headers, authentication user context is thread specific and isolated to the thread handling a particular request. Example usage: -- In the main thread of an application's start up code we might want to inject some global state like so. # In some start-up file called app.py from primordial.context import Context MY_APP_CTX = Context() # Instantiate some shared object jwt_fetcher = JWTFetcher(some_config) MY_APP_CTX.set_global('jwt_fetcher', jwt_fetcher) # In a thread that's handling a particular HTTP request # handle_req.py from app import MY_APP_CTX MY_APP_CTX.user = User() MY_APP_CTX.token = copy_token_from_header() # In a third file somewhere down the line of request processing # some_file_called_by_controller.py from app import MY_APP_CTX def some_func(): # All of these are magically available. # some_func's function signature didn't require to be changed MY_APP_CTX.jwt_fetcher.get() MY_APP_CTX.user.name == 'jDoe' MY_APP_CTX.token.is_valid() ''' def __init__(self): self._global = {} self._local = threading.local() def __getattr__(self, name: str) -> Any: try: return getattr(self._local, name) except AttributeError: if name in self._global: return self._global[name] raise def __setattr__(self, name: str, value: Any): if name in ('_global', '_local'): return super().__setattr__(name, value) setattr(self._local, name, value) def __delattr__(self, name: str): try: delattr(self._local, name) except AttributeError: if name not in self._global: raise del self._global[name] def set_global(self, name: str, value: Any): self._global[name] = value def unset_global(self, name: str): self._global.pop(name) CTX = Context() @contextmanager def make_context(local_vals: Optional[Mapping[str, Any]] = None, global_vals: Optional[Mapping[str, Any]] = None, ctx: Optional[Context] = None): ''' Python context-manager for managing the life-cycle of state stored in a context. This context manager allows for state to be both stored in a context and also cleans up this state when the context-manager is exited. Usage: # In some python module file1.py from primordial import Context ctx = Context() # In some python module file2.py from threading import Thread from primordial import make_context from file1 import ctx from file3 import fn3 def fn2(): global_vals = {'v1': 'abc', v2: 'def'} # Set some global values with make_context(global_vals=global_value, ctx): # Kick of fn3 in a new thread t1 = Thread(target=fn3, args=[]) t1.start() t1.join() fn2() # In some python module file3.py from primordial import make_context from file1 import ctx from file4 import fn4 def fn3(): # v2 will shadow the value that was set globally local_vals = {'v3': 'ghi', v2: 'jkl'} # Set some thread specific values # Once this function returns, ctx.v3 and ctx.v2 are not available for access with make_context(local_vals=local_value, ctx): fn4() # We can still access the globally set state here even after the above context manager # has exited. ctx.v1 ctx.v2 # The globally set v2 # In some python module file3.py from file1 import ctx def fn4(): # All of the accesses are valid ctx.v1 ctx.v2 # This accesses the local thread specific v2 ctx.v3 ''' ctx = ctx if ctx else CTX local_vals = local_vals if local_vals else {} global_vals = global_vals if global_vals else {} for k, v in local_vals.items(): setattr(ctx, k, v) for k, v in global_vals.items(): ctx.set_global(k, v) try: yield ctx finally: for k in local_vals: delattr(ctx, k) for k in global_vals: ctx.unset_global(k)