mode.locals

Implements thread-local stack using ContextVar (PEP 567).

This is a reimplementation of the local stack as used by Flask, Werkzeug, Celery, and other libraries to keep a thread-local stack of objects.

• Supports typing:

  request_stack: LocalStack[Request] = LocalStack()
  

Proxies

Proxy objects are lazy and pass all method calls and attribute accesses to an underlying object.

There are mixins/roles for many of the generic classes, and these can be combined to create proxies.

For example to create a proxy to a class that both implements the mutable mapping interface and is an async context manager:

def create_real():
    print('CREATING X')
    return X()

class XProxy(MutableMappingRole, AsyncContextManagerRole):
    ...

x = XProxy(create_real)

Evaluation

By default the callable passed to Proxy will be evaluated every time it is needed, so in the example above a new X will be created every time you access the underlying object:

>>> x['foo'] = 'value'
CREATING X

>>> x['foo']
CREATING X
'value'

>>> X['foo']
CREATING X
'value'

>>> # evaluates twice, once for async with then for __getitem__:
>>> async with x:
...    x['foo']
CREATING X
CREATING X
'value'

If you want the creation of the object to be lazy (created when first needed), you can pass the cache=True argument to Proxy:

>>> x = XProxy(create_real, cache=True)

>>> # Now only evaluates the first time it is needed.
>>> x['foo'] = 'value'
CREATING X

>>> x['foo']
'value'

>>> X['foo']
'value'

>>> async with x:
...    x['foo']
'value'

class mode.locals.LocalStack

LocalStack.

Manage state per coroutine (also thread safe).

Most famously used probably in Flask to keep track of the current request object.

push(obj: T) → Generator[[None, None], None]

Push a new item to the stack.

push_without_automatic_cleanup(obj: T) → None

pop() → Optional[T]

Remove the topmost item from the stack.

Note

Will return the old value or None if the stack was already empty.

property stack

property top

Return the topmost item on the stack.

Does not remove it from the stack.

Note

If the stack is empty, None is returned.

class mode.locals.Proxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to another object.

class mode.locals.AwaitableRole

Role/Mixin for typing.Awaitable proxy methods.

class mode.locals.AwaitableProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.Awaitable object.

class mode.locals.CoroutineRole

Role/Mixin for typing.Coroutine proxy methods.

send(value: T_contra) → T_co

Send a value into the coroutine. Return next yielded value or raise StopIteration.

throw(typ: Type[BaseException], val: Optional[BaseException] = None, tb: traceback = None) → T_co

Raise an exception in the coroutine. Return next yielded value or raise StopIteration.

close() → None

Raise GeneratorExit inside coroutine.

class mode.locals.CoroutineProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.Coroutine object.

class mode.locals.AsyncIterableRole

Role/Mixin for typing.AsyncIterable proxy methods.

class mode.locals.AsyncIterableProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.AsyncIterable object.

class mode.locals.AsyncIteratorRole

Role/Mixin for typing.AsyncIterator proxy methods.

class mode.locals.AsyncIteratorProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.AsyncIterator object.

class mode.locals.AsyncGeneratorRole

Role/Mixin for typing.AsyncGenerator proxy methods.

asend(value: T_contra) → Awaitable[T_co]

Send a value into the asynchronous generator. Return next yielded value or raise StopAsyncIteration.

athrow(typ: Type[BaseException], val: Optional[BaseException] = None, tb: traceback = None) → Awaitable[T_co]

Raise an exception in the asynchronous generator. Return next yielded value or raise StopAsyncIteration.

aclose() → Awaitable[None]

Raise GeneratorExit inside coroutine.

class mode.locals.AsyncGeneratorProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.AsyncGenerator object.

class mode.locals.SequenceRole

Role/Mixin for typing.Sequence proxy methods.

index(value[, start[, stop]]) → integer -- return first index of value.

Raises ValueError if the value is not present.

Supporting start and stop arguments is optional, but recommended.

count(value) → integer -- return number of occurrences of value

class mode.locals.SequenceProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.Sequence object.

class mode.locals.MutableSequenceRole

Role/Mixin for typing.MutableSequence proxy methods.

insert(index: int, object: T) → None

S.insert(index, value) – insert value before index

append(obj: T) → None

S.append(value) – append value to the end of the sequence

extend(iterable: Iterable[T]) → None

S.extend(iterable) – extend sequence by appending elements from the iterable

reverse() → None

S.reverse() – reverse IN PLACE

pop([index]) → item -- remove and return item at index (default last).

Raise IndexError if list is empty or index is out of range.

remove(object: T) → None

S.remove(value) – remove first occurrence of value. Raise ValueError if the value is not present.

class mode.locals.MutableSequenceProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.MutableSequence object.

class mode.locals.SetRole

Role/Mixin for typing.AbstractSet proxy methods.

isdisjoint(s: Iterable[Any]) → bool

Return True if two sets have a null intersection.

class mode.locals.SetProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.AbstractSet object.

class mode.locals.MutableSetRole

Role/Mixin for typing.MutableSet proxy methods.

add(x: T) → None

Add an element.

discard(x: T) → None

Remove an element. Do not raise an exception if absent.

clear() → None

This is slow (creates N new iterators!) but effective.

pop() → T

Return the popped value. Raise KeyError if empty.

remove(element: T) → None

Remove an element. If not a member, raise a KeyError.

class mode.locals.MutableSetProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.MutableSet object.

class mode.locals.ContextManagerRole

Role/Mixin for typing.ContextManager proxy methods.

class mode.locals.ContextManagerProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.ContextManager object.

class mode.locals.AsyncContextManagerRole

Role/Mixin for typing.AsyncContextManager proxy methods.

class mode.locals.AsyncContextManagerProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.AsyncContextManager object.

class mode.locals.MappingRole

Role/Mixin for typing.Mapping proxy methods.

get(k[, d]) → D[k] if k in D, else d. d defaults to None.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

values() → an object providing a view on D's values

class mode.locals.MappingProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.Mapping object.

class mode.locals.MutableMappingRole

Role/Mixin for typing.MutableMapping proxy methods.

clear() → None. Remove all items from D.

pop(k[, d]) → v, remove specified key and return the corresponding value.

If key is not found, d is returned if given, otherwise KeyError is raised.

popitem() → (k, v), remove and return some (key, value) pair

as a 2-tuple; but raise KeyError if D is empty.

setdefault(k[, d]) → D.get(k,d), also set D[k]=d if k not in D

update([E, ]**F) → None. Update D from mapping/iterable E and F.

If E present and has a .keys() method, does: for k in E: D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

class mode.locals.MutableMappingProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.MutableMapping object.

class mode.locals.CallableRole

Role/Mixin for typing.Callable proxy methods.

class mode.locals.CallableProxy(local: Callable[..., T], args: Tuple = None, kwargs: Dict = None, name: str = None, cache: bool = False, __doc__: str = None)

Proxy to typing.Callable object.

mode.locals.maybe_evaluate(obj: Any) → Any

Attempt to evaluate promise, even if obj is not a promise.