+++ /dev/null
-from __future__ import annotations
-
-import copy
-import math
-import operator
-import typing as t
-from contextvars import ContextVar
-from functools import partial
-from functools import update_wrapper
-from operator import attrgetter
-
-from .wsgi import ClosingIterator
-
-if t.TYPE_CHECKING:
- from _typeshed.wsgi import StartResponse
- from _typeshed.wsgi import WSGIApplication
- from _typeshed.wsgi import WSGIEnvironment
-
-T = t.TypeVar("T")
-F = t.TypeVar("F", bound=t.Callable[..., t.Any])
-
-
-def release_local(local: Local | LocalStack) -> None:
- """Release the data for the current context in a :class:`Local` or
- :class:`LocalStack` without using a :class:`LocalManager`.
-
- This should not be needed for modern use cases, and may be removed
- in the future.
-
- .. versionadded:: 0.6.1
- """
- local.__release_local__()
-
-
-class Local:
- """Create a namespace of context-local data. This wraps a
- :class:`ContextVar` containing a :class:`dict` value.
-
- This may incur a performance penalty compared to using individual
- context vars, as it has to copy data to avoid mutating the dict
- between nested contexts.
-
- :param context_var: The :class:`~contextvars.ContextVar` to use as
- storage for this local. If not given, one will be created.
- Context vars not created at the global scope may interfere with
- garbage collection.
-
- .. versionchanged:: 2.0
- Uses ``ContextVar`` instead of a custom storage implementation.
- """
-
- __slots__ = ("__storage",)
-
- def __init__(self, context_var: ContextVar[dict[str, t.Any]] | None = None) -> None:
- if context_var is None:
- # A ContextVar not created at global scope interferes with
- # Python's garbage collection. However, a local only makes
- # sense defined at the global scope as well, in which case
- # the GC issue doesn't seem relevant.
- context_var = ContextVar(f"werkzeug.Local<{id(self)}>.storage")
-
- object.__setattr__(self, "_Local__storage", context_var)
-
- def __iter__(self) -> t.Iterator[tuple[str, t.Any]]:
- return iter(self.__storage.get({}).items())
-
- def __call__(self, name: str, *, unbound_message: str | None = None) -> LocalProxy:
- """Create a :class:`LocalProxy` that access an attribute on this
- local namespace.
-
- :param name: Proxy this attribute.
- :param unbound_message: The error message that the proxy will
- show if the attribute isn't set.
- """
- return LocalProxy(self, name, unbound_message=unbound_message)
-
- def __release_local__(self) -> None:
- self.__storage.set({})
-
- def __getattr__(self, name: str) -> t.Any:
- values = self.__storage.get({})
-
- if name in values:
- return values[name]
-
- raise AttributeError(name)
-
- def __setattr__(self, name: str, value: t.Any) -> None:
- values = self.__storage.get({}).copy()
- values[name] = value
- self.__storage.set(values)
-
- def __delattr__(self, name: str) -> None:
- values = self.__storage.get({})
-
- if name in values:
- values = values.copy()
- del values[name]
- self.__storage.set(values)
- else:
- raise AttributeError(name)
-
-
-class LocalStack(t.Generic[T]):
- """Create a stack of context-local data. This wraps a
- :class:`ContextVar` containing a :class:`list` value.
-
- This may incur a performance penalty compared to using individual
- context vars, as it has to copy data to avoid mutating the list
- between nested contexts.
-
- :param context_var: The :class:`~contextvars.ContextVar` to use as
- storage for this local. If not given, one will be created.
- Context vars not created at the global scope may interfere with
- garbage collection.
-
- .. versionchanged:: 2.0
- Uses ``ContextVar`` instead of a custom storage implementation.
-
- .. versionadded:: 0.6.1
- """
-
- __slots__ = ("_storage",)
-
- def __init__(self, context_var: ContextVar[list[T]] | None = None) -> None:
- if context_var is None:
- # A ContextVar not created at global scope interferes with
- # Python's garbage collection. However, a local only makes
- # sense defined at the global scope as well, in which case
- # the GC issue doesn't seem relevant.
- context_var = ContextVar(f"werkzeug.LocalStack<{id(self)}>.storage")
-
- self._storage = context_var
-
- def __release_local__(self) -> None:
- self._storage.set([])
-
- def push(self, obj: T) -> list[T]:
- """Add a new item to the top of the stack."""
- stack = self._storage.get([]).copy()
- stack.append(obj)
- self._storage.set(stack)
- return stack
-
- def pop(self) -> T | None:
- """Remove the top item from the stack and return it. If the
- stack is empty, return ``None``.
- """
- stack = self._storage.get([])
-
- if len(stack) == 0:
- return None
-
- rv = stack[-1]
- self._storage.set(stack[:-1])
- return rv
-
- @property
- def top(self) -> T | None:
- """The topmost item on the stack. If the stack is empty,
- `None` is returned.
- """
- stack = self._storage.get([])
-
- if len(stack) == 0:
- return None
-
- return stack[-1]
-
- def __call__(
- self, name: str | None = None, *, unbound_message: str | None = None
- ) -> LocalProxy:
- """Create a :class:`LocalProxy` that accesses the top of this
- local stack.
-
- :param name: If given, the proxy access this attribute of the
- top item, rather than the item itself.
- :param unbound_message: The error message that the proxy will
- show if the stack is empty.
- """
- return LocalProxy(self, name, unbound_message=unbound_message)
-
-
-class LocalManager:
- """Manage releasing the data for the current context in one or more
- :class:`Local` and :class:`LocalStack` objects.
-
- This should not be needed for modern use cases, and may be removed
- in the future.
-
- :param locals: A local or list of locals to manage.
-
- .. versionchanged:: 2.1
- The ``ident_func`` was removed.
-
- .. versionchanged:: 0.7
- The ``ident_func`` parameter was added.
-
- .. versionchanged:: 0.6.1
- The :func:`release_local` function can be used instead of a
- manager.
- """
-
- __slots__ = ("locals",)
-
- def __init__(
- self,
- locals: None | (Local | LocalStack | t.Iterable[Local | LocalStack]) = None,
- ) -> None:
- if locals is None:
- self.locals = []
- elif isinstance(locals, Local):
- self.locals = [locals]
- else:
- self.locals = list(locals) # type: ignore[arg-type]
-
- def cleanup(self) -> None:
- """Release the data in the locals for this context. Call this at
- the end of each request or use :meth:`make_middleware`.
- """
- for local in self.locals:
- release_local(local)
-
- def make_middleware(self, app: WSGIApplication) -> WSGIApplication:
- """Wrap a WSGI application so that local data is released
- automatically after the response has been sent for a request.
- """
-
- def application(
- environ: WSGIEnvironment, start_response: StartResponse
- ) -> t.Iterable[bytes]:
- return ClosingIterator(app(environ, start_response), self.cleanup)
-
- return application
-
- def middleware(self, func: WSGIApplication) -> WSGIApplication:
- """Like :meth:`make_middleware` but used as a decorator on the
- WSGI application function.
-
- .. code-block:: python
-
- @manager.middleware
- def application(environ, start_response):
- ...
- """
- return update_wrapper(self.make_middleware(func), func)
-
- def __repr__(self) -> str:
- return f"<{type(self).__name__} storages: {len(self.locals)}>"
-
-
-class _ProxyLookup:
- """Descriptor that handles proxied attribute lookup for
- :class:`LocalProxy`.
-
- :param f: The built-in function this attribute is accessed through.
- Instead of looking up the special method, the function call
- is redone on the object.
- :param fallback: Return this function if the proxy is unbound
- instead of raising a :exc:`RuntimeError`.
- :param is_attr: This proxied name is an attribute, not a function.
- Call the fallback immediately to get the value.
- :param class_value: Value to return when accessed from the
- ``LocalProxy`` class directly. Used for ``__doc__`` so building
- docs still works.
- """
-
- __slots__ = ("bind_f", "fallback", "is_attr", "class_value", "name")
-
- def __init__(
- self,
- f: t.Callable | None = None,
- fallback: t.Callable | None = None,
- class_value: t.Any | None = None,
- is_attr: bool = False,
- ) -> None:
- bind_f: t.Callable[[LocalProxy, t.Any], t.Callable] | None
-
- if hasattr(f, "__get__"):
- # A Python function, can be turned into a bound method.
-
- def bind_f(instance: LocalProxy, obj: t.Any) -> t.Callable:
- return f.__get__(obj, type(obj)) # type: ignore
-
- elif f is not None:
- # A C function, use partial to bind the first argument.
-
- def bind_f(instance: LocalProxy, obj: t.Any) -> t.Callable:
- return partial(f, obj)
-
- else:
- # Use getattr, which will produce a bound method.
- bind_f = None
-
- self.bind_f = bind_f
- self.fallback = fallback
- self.class_value = class_value
- self.is_attr = is_attr
-
- def __set_name__(self, owner: LocalProxy, name: str) -> None:
- self.name = name
-
- def __get__(self, instance: LocalProxy, owner: type | None = None) -> t.Any:
- if instance is None:
- if self.class_value is not None:
- return self.class_value
-
- return self
-
- try:
- obj = instance._get_current_object()
- except RuntimeError:
- if self.fallback is None:
- raise
-
- fallback = self.fallback.__get__(instance, owner)
-
- if self.is_attr:
- # __class__ and __doc__ are attributes, not methods.
- # Call the fallback to get the value.
- return fallback()
-
- return fallback
-
- if self.bind_f is not None:
- return self.bind_f(instance, obj)
-
- return getattr(obj, self.name)
-
- def __repr__(self) -> str:
- return f"proxy {self.name}"
-
- def __call__(self, instance: LocalProxy, *args: t.Any, **kwargs: t.Any) -> t.Any:
- """Support calling unbound methods from the class. For example,
- this happens with ``copy.copy``, which does
- ``type(x).__copy__(x)``. ``type(x)`` can't be proxied, so it
- returns the proxy type and descriptor.
- """
- return self.__get__(instance, type(instance))(*args, **kwargs)
-
-
-class _ProxyIOp(_ProxyLookup):
- """Look up an augmented assignment method on a proxied object. The
- method is wrapped to return the proxy instead of the object.
- """
-
- __slots__ = ()
-
- def __init__(
- self, f: t.Callable | None = None, fallback: t.Callable | None = None
- ) -> None:
- super().__init__(f, fallback)
-
- def bind_f(instance: LocalProxy, obj: t.Any) -> t.Callable:
- def i_op(self: t.Any, other: t.Any) -> LocalProxy:
- f(self, other) # type: ignore
- return instance
-
- return i_op.__get__(obj, type(obj)) # type: ignore
-
- self.bind_f = bind_f
-
-
-def _l_to_r_op(op: F) -> F:
- """Swap the argument order to turn an l-op into an r-op."""
-
- def r_op(obj: t.Any, other: t.Any) -> t.Any:
- return op(other, obj)
-
- return t.cast(F, r_op)
-
-
-def _identity(o: T) -> T:
- return o
-
-
-class LocalProxy(t.Generic[T]):
- """A proxy to the object bound to a context-local object. All
- operations on the proxy are forwarded to the bound object. If no
- object is bound, a ``RuntimeError`` is raised.
-
- :param local: The context-local object that provides the proxied
- object.
- :param name: Proxy this attribute from the proxied object.
- :param unbound_message: The error message to show if the
- context-local object is unbound.
-
- Proxy a :class:`~contextvars.ContextVar` to make it easier to
- access. Pass a name to proxy that attribute.
-
- .. code-block:: python
-
- _request_var = ContextVar("request")
- request = LocalProxy(_request_var)
- session = LocalProxy(_request_var, "session")
-
- Proxy an attribute on a :class:`Local` namespace by calling the
- local with the attribute name:
-
- .. code-block:: python
-
- data = Local()
- user = data("user")
-
- Proxy the top item on a :class:`LocalStack` by calling the local.
- Pass a name to proxy that attribute.
-
- .. code-block::
-
- app_stack = LocalStack()
- current_app = app_stack()
- g = app_stack("g")
-
- Pass a function to proxy the return value from that function. This
- was previously used to access attributes of local objects before
- that was supported directly.
-
- .. code-block:: python
-
- session = LocalProxy(lambda: request.session)
-
- ``__repr__`` and ``__class__`` are proxied, so ``repr(x)`` and
- ``isinstance(x, cls)`` will look like the proxied object. Use
- ``issubclass(type(x), LocalProxy)`` to check if an object is a
- proxy.
-
- .. code-block:: python
-
- repr(user) # <User admin>
- isinstance(user, User) # True
- issubclass(type(user), LocalProxy) # True
-
- .. versionchanged:: 2.2.2
- ``__wrapped__`` is set when wrapping an object, not only when
- wrapping a function, to prevent doctest from failing.
-
- .. versionchanged:: 2.2
- Can proxy a ``ContextVar`` or ``LocalStack`` directly.
-
- .. versionchanged:: 2.2
- The ``name`` parameter can be used with any proxied object, not
- only ``Local``.
-
- .. versionchanged:: 2.2
- Added the ``unbound_message`` parameter.
-
- .. versionchanged:: 2.0
- Updated proxied attributes and methods to reflect the current
- data model.
-
- .. versionchanged:: 0.6.1
- The class can be instantiated with a callable.
- """
-
- __slots__ = ("__wrapped", "_get_current_object")
-
- _get_current_object: t.Callable[[], T]
- """Return the current object this proxy is bound to. If the proxy is
- unbound, this raises a ``RuntimeError``.
-
- This should be used if you need to pass the object to something that
- doesn't understand the proxy. It can also be useful for performance
- if you are accessing the object multiple times in a function, rather
- than going through the proxy multiple times.
- """
-
- def __init__(
- self,
- local: ContextVar[T] | Local | LocalStack[T] | t.Callable[[], T],
- name: str | None = None,
- *,
- unbound_message: str | None = None,
- ) -> None:
- if name is None:
- get_name = _identity
- else:
- get_name = attrgetter(name) # type: ignore[assignment]
-
- if unbound_message is None:
- unbound_message = "object is not bound"
-
- if isinstance(local, Local):
- if name is None:
- raise TypeError("'name' is required when proxying a 'Local' object.")
-
- def _get_current_object() -> T:
- try:
- return get_name(local) # type: ignore[return-value]
- except AttributeError:
- raise RuntimeError(unbound_message) from None
-
- elif isinstance(local, LocalStack):
-
- def _get_current_object() -> T:
- obj = local.top
-
- if obj is None:
- raise RuntimeError(unbound_message)
-
- return get_name(obj)
-
- elif isinstance(local, ContextVar):
-
- def _get_current_object() -> T:
- try:
- obj = local.get()
- except LookupError:
- raise RuntimeError(unbound_message) from None
-
- return get_name(obj)
-
- elif callable(local):
-
- def _get_current_object() -> T:
- return get_name(local())
-
- else:
- raise TypeError(f"Don't know how to proxy '{type(local)}'.")
-
- object.__setattr__(self, "_LocalProxy__wrapped", local)
- object.__setattr__(self, "_get_current_object", _get_current_object)
-
- __doc__ = _ProxyLookup( # type: ignore
- class_value=__doc__, fallback=lambda self: type(self).__doc__, is_attr=True
- )
- __wrapped__ = _ProxyLookup(
- fallback=lambda self: self._LocalProxy__wrapped, is_attr=True
- )
- # __del__ should only delete the proxy
- __repr__ = _ProxyLookup( # type: ignore
- repr, fallback=lambda self: f"<{type(self).__name__} unbound>"
- )
- __str__ = _ProxyLookup(str) # type: ignore
- __bytes__ = _ProxyLookup(bytes)
- __format__ = _ProxyLookup() # type: ignore
- __lt__ = _ProxyLookup(operator.lt)
- __le__ = _ProxyLookup(operator.le)
- __eq__ = _ProxyLookup(operator.eq) # type: ignore
- __ne__ = _ProxyLookup(operator.ne) # type: ignore
- __gt__ = _ProxyLookup(operator.gt)
- __ge__ = _ProxyLookup(operator.ge)
- __hash__ = _ProxyLookup(hash) # type: ignore
- __bool__ = _ProxyLookup(bool, fallback=lambda self: False)
- __getattr__ = _ProxyLookup(getattr)
- # __getattribute__ triggered through __getattr__
- __setattr__ = _ProxyLookup(setattr) # type: ignore
- __delattr__ = _ProxyLookup(delattr) # type: ignore
- __dir__ = _ProxyLookup(dir, fallback=lambda self: []) # type: ignore
- # __get__ (proxying descriptor not supported)
- # __set__ (descriptor)
- # __delete__ (descriptor)
- # __set_name__ (descriptor)
- # __objclass__ (descriptor)
- # __slots__ used by proxy itself
- # __dict__ (__getattr__)
- # __weakref__ (__getattr__)
- # __init_subclass__ (proxying metaclass not supported)
- # __prepare__ (metaclass)
- __class__ = _ProxyLookup(
- fallback=lambda self: type(self), is_attr=True
- ) # type: ignore
- __instancecheck__ = _ProxyLookup(lambda self, other: isinstance(other, self))
- __subclasscheck__ = _ProxyLookup(lambda self, other: issubclass(other, self))
- # __class_getitem__ triggered through __getitem__
- __call__ = _ProxyLookup(lambda self, *args, **kwargs: self(*args, **kwargs))
- __len__ = _ProxyLookup(len)
- __length_hint__ = _ProxyLookup(operator.length_hint)
- __getitem__ = _ProxyLookup(operator.getitem)
- __setitem__ = _ProxyLookup(operator.setitem)
- __delitem__ = _ProxyLookup(operator.delitem)
- # __missing__ triggered through __getitem__
- __iter__ = _ProxyLookup(iter)
- __next__ = _ProxyLookup(next)
- __reversed__ = _ProxyLookup(reversed)
- __contains__ = _ProxyLookup(operator.contains)
- __add__ = _ProxyLookup(operator.add)
- __sub__ = _ProxyLookup(operator.sub)
- __mul__ = _ProxyLookup(operator.mul)
- __matmul__ = _ProxyLookup(operator.matmul)
- __truediv__ = _ProxyLookup(operator.truediv)
- __floordiv__ = _ProxyLookup(operator.floordiv)
- __mod__ = _ProxyLookup(operator.mod)
- __divmod__ = _ProxyLookup(divmod)
- __pow__ = _ProxyLookup(pow)
- __lshift__ = _ProxyLookup(operator.lshift)
- __rshift__ = _ProxyLookup(operator.rshift)
- __and__ = _ProxyLookup(operator.and_)
- __xor__ = _ProxyLookup(operator.xor)
- __or__ = _ProxyLookup(operator.or_)
- __radd__ = _ProxyLookup(_l_to_r_op(operator.add))
- __rsub__ = _ProxyLookup(_l_to_r_op(operator.sub))
- __rmul__ = _ProxyLookup(_l_to_r_op(operator.mul))
- __rmatmul__ = _ProxyLookup(_l_to_r_op(operator.matmul))
- __rtruediv__ = _ProxyLookup(_l_to_r_op(operator.truediv))
- __rfloordiv__ = _ProxyLookup(_l_to_r_op(operator.floordiv))
- __rmod__ = _ProxyLookup(_l_to_r_op(operator.mod))
- __rdivmod__ = _ProxyLookup(_l_to_r_op(divmod))
- __rpow__ = _ProxyLookup(_l_to_r_op(pow))
- __rlshift__ = _ProxyLookup(_l_to_r_op(operator.lshift))
- __rrshift__ = _ProxyLookup(_l_to_r_op(operator.rshift))
- __rand__ = _ProxyLookup(_l_to_r_op(operator.and_))
- __rxor__ = _ProxyLookup(_l_to_r_op(operator.xor))
- __ror__ = _ProxyLookup(_l_to_r_op(operator.or_))
- __iadd__ = _ProxyIOp(operator.iadd)
- __isub__ = _ProxyIOp(operator.isub)
- __imul__ = _ProxyIOp(operator.imul)
- __imatmul__ = _ProxyIOp(operator.imatmul)
- __itruediv__ = _ProxyIOp(operator.itruediv)
- __ifloordiv__ = _ProxyIOp(operator.ifloordiv)
- __imod__ = _ProxyIOp(operator.imod)
- __ipow__ = _ProxyIOp(operator.ipow)
- __ilshift__ = _ProxyIOp(operator.ilshift)
- __irshift__ = _ProxyIOp(operator.irshift)
- __iand__ = _ProxyIOp(operator.iand)
- __ixor__ = _ProxyIOp(operator.ixor)
- __ior__ = _ProxyIOp(operator.ior)
- __neg__ = _ProxyLookup(operator.neg)
- __pos__ = _ProxyLookup(operator.pos)
- __abs__ = _ProxyLookup(abs)
- __invert__ = _ProxyLookup(operator.invert)
- __complex__ = _ProxyLookup(complex)
- __int__ = _ProxyLookup(int)
- __float__ = _ProxyLookup(float)
- __index__ = _ProxyLookup(operator.index)
- __round__ = _ProxyLookup(round)
- __trunc__ = _ProxyLookup(math.trunc)
- __floor__ = _ProxyLookup(math.floor)
- __ceil__ = _ProxyLookup(math.ceil)
- __enter__ = _ProxyLookup()
- __exit__ = _ProxyLookup()
- __await__ = _ProxyLookup()
- __aiter__ = _ProxyLookup()
- __anext__ = _ProxyLookup()
- __aenter__ = _ProxyLookup()
- __aexit__ = _ProxyLookup()
- __copy__ = _ProxyLookup(copy.copy)
- __deepcopy__ = _ProxyLookup(copy.deepcopy)
- # __getnewargs_ex__ (pickle through proxy not supported)
- # __getnewargs__ (pickle)
- # __getstate__ (pickle)
- # __setstate__ (pickle)
- # __reduce__ (pickle)
- # __reduce_ex__ (pickle)
projects / dlqueue.git / blobdiff