"""Scope interface and builtin implementations."""importabcimportcontextvarsas_contextvarsimportthreadingimporttypingast
[docs]classScope(metaclass=abc.ABCMeta):"""Scope is an execution context based storage interface. Execution context is a mechanism of storing and accessing data bound to a logical thread of execution. Thus, one may consider processes, threads, greenlets, coroutines, Flask requests to be examples of a logical thread. The interface provides just two methods: * :meth:`.set` - set execution context item * :meth:`.get` - get execution context item See corresponding methods for details below. """
[docs]@abc.abstractmethoddefset(self,key:t.Hashable,value:t.Any)->None:"""Bind `value` to `key` in current execution context."""
[docs]@abc.abstractmethoddefget(self,key:t.Hashable)->t.Any:"""Get `value` by `key` for current execution context."""
[docs]classsingleton(Scope):"""Share instances across application."""def__init__(self):self._store={}defset(self,key:t.Hashable,value:t.Any)->None:self._store[key]=valuedefget(self,key:t.Hashable)->t.Any:returnself._store[key]
[docs]classthreadlocal(Scope):"""Share instances across the same thread."""def__init__(self):self._local=threading.local()defset(self,key:t.Hashable,value:t.Any)->None:try:store=self._local.storeexceptAttributeError:store=self._local.store={}store[key]=valuedefget(self,key:t.Hashable)->t.Any:try:rv=self._local.store[key]exceptAttributeError:raiseKeyError(key)returnrv
[docs]classcontextvars(Scope):"""Share instances across the same execution context (:pep:`567`). Since `asyncio does support context variables`__, the scope could be used in asynchronous applications to share dependencies between coroutines of the same :class:`asyncio.Task`. .. __: https://docs.python.org/3/library/contextvars.html#asyncio-support .. versionadded:: 2.1 """def__init__(self):self._store={}defset(self,key:t.Hashable,value:t.Any)->None:try:var=self._store[key]exceptKeyError:var=self._store[key]=_contextvars.ContextVar("picobox")var.set(value)defget(self,key:t.Hashable)->t.Any:try:returnself._store[key].get()exceptLookupError:raiseKeyError(key)
[docs]classnoscope(Scope):"""Do not share instances, create them each time on demand."""defset(self,key:t.Hashable,value:t.Any)->None:passdefget(self,key:t.Hashable)->t.Any:raiseKeyError(key)