Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/local.py: 72%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import copy
4import math
5import operator
6import typing as t
7from contextvars import ContextVar
8from functools import partial
9from functools import update_wrapper
10from operator import attrgetter
12from .wsgi import ClosingIterator
14if t.TYPE_CHECKING:
15 from _typeshed.wsgi import StartResponse
16 from _typeshed.wsgi import WSGIApplication
17 from _typeshed.wsgi import WSGIEnvironment
19T = t.TypeVar("T")
20F = t.TypeVar("F", bound=t.Callable[..., t.Any])
23def release_local(local: Local | LocalStack[t.Any]) -> None:
24 """Release the data for the current context in a :class:`Local` or
25 :class:`LocalStack` without using a :class:`LocalManager`.
27 This should not be needed for modern use cases, and may be removed
28 in the future.
30 .. versionadded:: 0.6.1
31 """
32 local.__release_local__()
35class Local:
36 """Create a namespace of context-local data. This wraps a
37 :class:`ContextVar` containing a :class:`dict` value.
39 This may incur a performance penalty compared to using individual
40 context vars, as it has to copy data to avoid mutating the dict
41 between nested contexts.
43 :param context_var: The :class:`~contextvars.ContextVar` to use as
44 storage for this local. If not given, one will be created.
45 Context vars not created at the global scope may interfere with
46 garbage collection.
48 .. versionchanged:: 2.0
49 Uses ``ContextVar`` instead of a custom storage implementation.
50 """
52 __slots__ = ("__storage",)
54 def __init__(self, context_var: ContextVar[dict[str, t.Any]] | None = None) -> None:
55 if context_var is None:
56 # A ContextVar not created at global scope interferes with
57 # Python's garbage collection. However, a local only makes
58 # sense defined at the global scope as well, in which case
59 # the GC issue doesn't seem relevant.
60 context_var = ContextVar(f"werkzeug.Local<{id(self)}>.storage")
62 object.__setattr__(self, "_Local__storage", context_var)
64 def __iter__(self) -> t.Iterator[tuple[str, t.Any]]:
65 return iter(self.__storage.get({}).items())
67 def __call__(
68 self, name: str, *, unbound_message: str | None = None
69 ) -> LocalProxy[t.Any]:
70 """Create a :class:`LocalProxy` that access an attribute on this
71 local namespace.
73 :param name: Proxy this attribute.
74 :param unbound_message: The error message that the proxy will
75 show if the attribute isn't set.
76 """
77 return LocalProxy(self, name, unbound_message=unbound_message)
79 def __release_local__(self) -> None:
80 self.__storage.set({})
82 def __getattr__(self, name: str) -> t.Any:
83 values = self.__storage.get({})
85 if name in values:
86 return values[name]
88 raise AttributeError(name)
90 def __setattr__(self, name: str, value: t.Any) -> None:
91 values = self.__storage.get({}).copy()
92 values[name] = value
93 self.__storage.set(values)
95 def __delattr__(self, name: str) -> None:
96 values = self.__storage.get({})
98 if name in values:
99 values = values.copy()
100 del values[name]
101 self.__storage.set(values)
102 else:
103 raise AttributeError(name)
106class LocalStack(t.Generic[T]):
107 """Create a stack of context-local data. This wraps a
108 :class:`ContextVar` containing a :class:`list` value.
110 This may incur a performance penalty compared to using individual
111 context vars, as it has to copy data to avoid mutating the list
112 between nested contexts.
114 :param context_var: The :class:`~contextvars.ContextVar` to use as
115 storage for this local. If not given, one will be created.
116 Context vars not created at the global scope may interfere with
117 garbage collection.
119 .. versionchanged:: 2.0
120 Uses ``ContextVar`` instead of a custom storage implementation.
122 .. versionadded:: 0.6.1
123 """
125 __slots__ = ("_storage",)
127 def __init__(self, context_var: ContextVar[list[T]] | None = None) -> None:
128 if context_var is None:
129 # A ContextVar not created at global scope interferes with
130 # Python's garbage collection. However, a local only makes
131 # sense defined at the global scope as well, in which case
132 # the GC issue doesn't seem relevant.
133 context_var = ContextVar(f"werkzeug.LocalStack<{id(self)}>.storage")
135 self._storage = context_var
137 def __release_local__(self) -> None:
138 self._storage.set([])
140 def push(self, obj: T) -> list[T]:
141 """Add a new item to the top of the stack."""
142 stack = self._storage.get([]).copy()
143 stack.append(obj)
144 self._storage.set(stack)
145 return stack
147 def pop(self) -> T | None:
148 """Remove the top item from the stack and return it. If the
149 stack is empty, return ``None``.
150 """
151 stack = self._storage.get([])
153 if len(stack) == 0:
154 return None
156 rv = stack[-1]
157 self._storage.set(stack[:-1])
158 return rv
160 @property
161 def top(self) -> T | None:
162 """The topmost item on the stack. If the stack is empty,
163 `None` is returned.
164 """
165 stack = self._storage.get([])
167 if len(stack) == 0:
168 return None
170 return stack[-1]
172 def __call__(
173 self, name: str | None = None, *, unbound_message: str | None = None
174 ) -> LocalProxy[t.Any]:
175 """Create a :class:`LocalProxy` that accesses the top of this
176 local stack.
178 :param name: If given, the proxy access this attribute of the
179 top item, rather than the item itself.
180 :param unbound_message: The error message that the proxy will
181 show if the stack is empty.
182 """
183 return LocalProxy(self, name, unbound_message=unbound_message)
186class LocalManager:
187 """Manage releasing the data for the current context in one or more
188 :class:`Local` and :class:`LocalStack` objects.
190 This should not be needed for modern use cases, and may be removed
191 in the future.
193 :param locals: A local or list of locals to manage.
195 .. versionchanged:: 2.1
196 The ``ident_func`` was removed.
198 .. versionchanged:: 0.7
199 The ``ident_func`` parameter was added.
201 .. versionchanged:: 0.6.1
202 The :func:`release_local` function can be used instead of a
203 manager.
204 """
206 __slots__ = ("locals",)
208 def __init__(
209 self,
210 locals: None
211 | (Local | LocalStack[t.Any] | t.Iterable[Local | LocalStack[t.Any]]) = None,
212 ) -> None:
213 if locals is None:
214 self.locals = []
215 elif isinstance(locals, Local):
216 self.locals = [locals]
217 else:
218 self.locals = list(locals) # type: ignore[arg-type]
220 def cleanup(self) -> None:
221 """Release the data in the locals for this context. Call this at
222 the end of each request or use :meth:`make_middleware`.
223 """
224 for local in self.locals:
225 release_local(local)
227 def make_middleware(self, app: WSGIApplication) -> WSGIApplication:
228 """Wrap a WSGI application so that local data is released
229 automatically after the response has been sent for a request.
230 """
232 def application(
233 environ: WSGIEnvironment, start_response: StartResponse
234 ) -> t.Iterable[bytes]:
235 return ClosingIterator(app(environ, start_response), self.cleanup)
237 return application
239 def middleware(self, func: WSGIApplication) -> WSGIApplication:
240 """Like :meth:`make_middleware` but used as a decorator on the
241 WSGI application function.
243 .. code-block:: python
245 @manager.middleware
246 def application(environ, start_response):
247 ...
248 """
249 return update_wrapper(self.make_middleware(func), func)
251 def __repr__(self) -> str:
252 return f"<{type(self).__name__} storages: {len(self.locals)}>"
255class _ProxyLookup:
256 """Descriptor that handles proxied attribute lookup for
257 :class:`LocalProxy`.
259 :param f: The built-in function this attribute is accessed through.
260 Instead of looking up the special method, the function call
261 is redone on the object.
262 :param fallback: Return this function if the proxy is unbound
263 instead of raising a :exc:`RuntimeError`.
264 :param is_attr: This proxied name is an attribute, not a function.
265 Call the fallback immediately to get the value.
266 :param class_value: Value to return when accessed from the
267 ``LocalProxy`` class directly. Used for ``__doc__`` so building
268 docs still works.
269 """
271 __slots__ = ("bind_f", "fallback", "is_attr", "class_value", "name")
273 def __init__(
274 self,
275 f: t.Callable[..., t.Any] | None = None,
276 fallback: t.Callable[[LocalProxy[t.Any]], t.Any] | None = None,
277 class_value: t.Any | None = None,
278 is_attr: bool = False,
279 ) -> None:
280 bind_f: t.Callable[[LocalProxy[t.Any], t.Any], t.Callable[..., t.Any]] | None
282 if hasattr(f, "__get__"):
283 # A Python function, can be turned into a bound method.
285 def bind_f(
286 instance: LocalProxy[t.Any], obj: t.Any
287 ) -> t.Callable[..., t.Any]:
288 return f.__get__(obj, type(obj)) # type: ignore
290 elif f is not None:
291 # A C function, use partial to bind the first argument.
293 def bind_f(
294 instance: LocalProxy[t.Any], obj: t.Any
295 ) -> t.Callable[..., t.Any]:
296 return partial(f, obj)
298 else:
299 # Use getattr, which will produce a bound method.
300 bind_f = None
302 self.bind_f = bind_f
303 self.fallback = fallback
304 self.class_value = class_value
305 self.is_attr = is_attr
307 def __set_name__(self, owner: LocalProxy[t.Any], name: str) -> None:
308 self.name = name
310 def __get__(self, instance: LocalProxy[t.Any], owner: type | None = None) -> t.Any:
311 if instance is None:
312 if self.class_value is not None:
313 return self.class_value
315 return self
317 try:
318 obj = instance._get_current_object()
319 except RuntimeError:
320 if self.fallback is None:
321 raise
323 fallback = self.fallback.__get__(instance, owner)
325 if self.is_attr:
326 # __class__ and __doc__ are attributes, not methods.
327 # Call the fallback to get the value.
328 return fallback()
330 return fallback
332 if self.bind_f is not None:
333 return self.bind_f(instance, obj)
335 return getattr(obj, self.name)
337 def __repr__(self) -> str:
338 return f"proxy {self.name}"
340 def __call__(
341 self, instance: LocalProxy[t.Any], *args: t.Any, **kwargs: t.Any
342 ) -> t.Any:
343 """Support calling unbound methods from the class. For example,
344 this happens with ``copy.copy``, which does
345 ``type(x).__copy__(x)``. ``type(x)`` can't be proxied, so it
346 returns the proxy type and descriptor.
347 """
348 return self.__get__(instance, type(instance))(*args, **kwargs)
351class _ProxyIOp(_ProxyLookup):
352 """Look up an augmented assignment method on a proxied object. The
353 method is wrapped to return the proxy instead of the object.
354 """
356 __slots__ = ()
358 def __init__(
359 self,
360 f: t.Callable[..., t.Any] | None = None,
361 fallback: t.Callable[[LocalProxy[t.Any]], t.Any] | None = None,
362 ) -> None:
363 super().__init__(f, fallback)
365 def bind_f(instance: LocalProxy[t.Any], obj: t.Any) -> t.Callable[..., t.Any]:
366 def i_op(self: t.Any, other: t.Any) -> LocalProxy[t.Any]:
367 f(self, other) # type: ignore
368 return instance
370 return i_op.__get__(obj, type(obj)) # type: ignore
372 self.bind_f = bind_f
375def _l_to_r_op(op: F) -> F:
376 """Swap the argument order to turn an l-op into an r-op."""
378 def r_op(obj: t.Any, other: t.Any) -> t.Any:
379 return op(other, obj)
381 return t.cast(F, r_op)
384def _identity(o: T) -> T:
385 return o
388class LocalProxy(t.Generic[T]):
389 """A proxy to the object bound to a context-local object. All
390 operations on the proxy are forwarded to the bound object. If no
391 object is bound, a ``RuntimeError`` is raised.
393 :param local: The context-local object that provides the proxied
394 object.
395 :param name: Proxy this attribute from the proxied object.
396 :param unbound_message: The error message to show if the
397 context-local object is unbound.
399 Proxy a :class:`~contextvars.ContextVar` to make it easier to
400 access. Pass a name to proxy that attribute.
402 .. code-block:: python
404 _request_var = ContextVar("request")
405 request = LocalProxy(_request_var)
406 session = LocalProxy(_request_var, "session")
408 Proxy an attribute on a :class:`Local` namespace by calling the
409 local with the attribute name:
411 .. code-block:: python
413 data = Local()
414 user = data("user")
416 Proxy the top item on a :class:`LocalStack` by calling the local.
417 Pass a name to proxy that attribute.
419 .. code-block::
421 app_stack = LocalStack()
422 current_app = app_stack()
423 g = app_stack("g")
425 Pass a function to proxy the return value from that function. This
426 was previously used to access attributes of local objects before
427 that was supported directly.
429 .. code-block:: python
431 session = LocalProxy(lambda: request.session)
433 ``__repr__`` and ``__class__`` are proxied, so ``repr(x)`` and
434 ``isinstance(x, cls)`` will look like the proxied object. Use
435 ``issubclass(type(x), LocalProxy)`` to check if an object is a
436 proxy.
438 .. code-block:: python
440 repr(user) # <User admin>
441 isinstance(user, User) # True
442 issubclass(type(user), LocalProxy) # True
444 .. versionchanged:: 2.2.2
445 ``__wrapped__`` is set when wrapping an object, not only when
446 wrapping a function, to prevent doctest from failing.
448 .. versionchanged:: 2.2
449 Can proxy a ``ContextVar`` or ``LocalStack`` directly.
451 .. versionchanged:: 2.2
452 The ``name`` parameter can be used with any proxied object, not
453 only ``Local``.
455 .. versionchanged:: 2.2
456 Added the ``unbound_message`` parameter.
458 .. versionchanged:: 2.0
459 Updated proxied attributes and methods to reflect the current
460 data model.
462 .. versionchanged:: 0.6.1
463 The class can be instantiated with a callable.
464 """
466 __slots__ = ("__wrapped", "_get_current_object")
468 _get_current_object: t.Callable[[], T]
469 """Return the current object this proxy is bound to. If the proxy is
470 unbound, this raises a ``RuntimeError``.
472 This should be used if you need to pass the object to something that
473 doesn't understand the proxy. It can also be useful for performance
474 if you are accessing the object multiple times in a function, rather
475 than going through the proxy multiple times.
476 """
478 def __init__(
479 self,
480 local: ContextVar[T] | Local | LocalStack[T] | t.Callable[[], T],
481 name: str | None = None,
482 *,
483 unbound_message: str | None = None,
484 ) -> None:
485 if name is None:
486 get_name = _identity
487 else:
488 get_name = attrgetter(name) # type: ignore[assignment]
490 if unbound_message is None:
491 unbound_message = "object is not bound"
493 if isinstance(local, Local):
494 if name is None:
495 raise TypeError("'name' is required when proxying a 'Local' object.")
497 def _get_current_object() -> T:
498 try:
499 return get_name(local) # type: ignore[return-value]
500 except AttributeError:
501 raise RuntimeError(unbound_message) from None
503 elif isinstance(local, LocalStack):
505 def _get_current_object() -> T:
506 obj = local.top
508 if obj is None:
509 raise RuntimeError(unbound_message)
511 return get_name(obj)
513 elif isinstance(local, ContextVar):
515 def _get_current_object() -> T:
516 try:
517 obj = local.get()
518 except LookupError:
519 raise RuntimeError(unbound_message) from None
521 return get_name(obj)
523 elif callable(local):
525 def _get_current_object() -> T:
526 return get_name(local())
528 else:
529 raise TypeError(f"Don't know how to proxy '{type(local)}'.")
531 object.__setattr__(self, "_LocalProxy__wrapped", local)
532 object.__setattr__(self, "_get_current_object", _get_current_object)
534 __doc__ = _ProxyLookup( # type: ignore[assignment]
535 class_value=__doc__, fallback=lambda self: type(self).__doc__, is_attr=True
536 )
537 __wrapped__ = _ProxyLookup(
538 fallback=lambda self: self._LocalProxy__wrapped, # type: ignore[attr-defined]
539 is_attr=True,
540 )
541 # __del__ should only delete the proxy
542 __repr__ = _ProxyLookup( # type: ignore[assignment]
543 repr, fallback=lambda self: f"<{type(self).__name__} unbound>"
544 )
545 __str__ = _ProxyLookup(str) # type: ignore[assignment]
546 __bytes__ = _ProxyLookup(bytes)
547 __format__ = _ProxyLookup() # type: ignore[assignment]
548 __lt__ = _ProxyLookup(operator.lt)
549 __le__ = _ProxyLookup(operator.le)
550 __eq__ = _ProxyLookup(operator.eq) # type: ignore[assignment]
551 __ne__ = _ProxyLookup(operator.ne) # type: ignore[assignment]
552 __gt__ = _ProxyLookup(operator.gt)
553 __ge__ = _ProxyLookup(operator.ge)
554 __hash__ = _ProxyLookup(hash) # type: ignore[assignment]
555 __bool__ = _ProxyLookup(bool, fallback=lambda self: False)
556 __getattr__ = _ProxyLookup(getattr)
557 # __getattribute__ triggered through __getattr__
558 __setattr__ = _ProxyLookup(setattr) # type: ignore[assignment]
559 __delattr__ = _ProxyLookup(delattr) # type: ignore[assignment]
560 __dir__ = _ProxyLookup(dir, fallback=lambda self: []) # type: ignore[assignment]
561 # __get__ (proxying descriptor not supported)
562 # __set__ (descriptor)
563 # __delete__ (descriptor)
564 # __set_name__ (descriptor)
565 # __objclass__ (descriptor)
566 # __slots__ used by proxy itself
567 # __dict__ (__getattr__)
568 # __weakref__ (__getattr__)
569 # __init_subclass__ (proxying metaclass not supported)
570 # __prepare__ (metaclass)
571 __class__ = _ProxyLookup(fallback=lambda self: type(self), is_attr=True) # type: ignore[assignment]
572 __instancecheck__ = _ProxyLookup(lambda self, other: isinstance(other, self))
573 __subclasscheck__ = _ProxyLookup(lambda self, other: issubclass(other, self))
574 # __class_getitem__ triggered through __getitem__
575 __call__ = _ProxyLookup(lambda self, *args, **kwargs: self(*args, **kwargs))
576 __len__ = _ProxyLookup(len)
577 __length_hint__ = _ProxyLookup(operator.length_hint)
578 __getitem__ = _ProxyLookup(operator.getitem)
579 __setitem__ = _ProxyLookup(operator.setitem)
580 __delitem__ = _ProxyLookup(operator.delitem)
581 # __missing__ triggered through __getitem__
582 __iter__ = _ProxyLookup(iter)
583 __next__ = _ProxyLookup(next)
584 __reversed__ = _ProxyLookup(reversed)
585 __contains__ = _ProxyLookup(operator.contains)
586 __add__ = _ProxyLookup(operator.add)
587 __sub__ = _ProxyLookup(operator.sub)
588 __mul__ = _ProxyLookup(operator.mul)
589 __matmul__ = _ProxyLookup(operator.matmul)
590 __truediv__ = _ProxyLookup(operator.truediv)
591 __floordiv__ = _ProxyLookup(operator.floordiv)
592 __mod__ = _ProxyLookup(operator.mod)
593 __divmod__ = _ProxyLookup(divmod)
594 __pow__ = _ProxyLookup(pow)
595 __lshift__ = _ProxyLookup(operator.lshift)
596 __rshift__ = _ProxyLookup(operator.rshift)
597 __and__ = _ProxyLookup(operator.and_)
598 __xor__ = _ProxyLookup(operator.xor)
599 __or__ = _ProxyLookup(operator.or_)
600 __radd__ = _ProxyLookup(_l_to_r_op(operator.add))
601 __rsub__ = _ProxyLookup(_l_to_r_op(operator.sub))
602 __rmul__ = _ProxyLookup(_l_to_r_op(operator.mul))
603 __rmatmul__ = _ProxyLookup(_l_to_r_op(operator.matmul))
604 __rtruediv__ = _ProxyLookup(_l_to_r_op(operator.truediv))
605 __rfloordiv__ = _ProxyLookup(_l_to_r_op(operator.floordiv))
606 __rmod__ = _ProxyLookup(_l_to_r_op(operator.mod))
607 __rdivmod__ = _ProxyLookup(_l_to_r_op(divmod))
608 __rpow__ = _ProxyLookup(_l_to_r_op(pow))
609 __rlshift__ = _ProxyLookup(_l_to_r_op(operator.lshift))
610 __rrshift__ = _ProxyLookup(_l_to_r_op(operator.rshift))
611 __rand__ = _ProxyLookup(_l_to_r_op(operator.and_))
612 __rxor__ = _ProxyLookup(_l_to_r_op(operator.xor))
613 __ror__ = _ProxyLookup(_l_to_r_op(operator.or_))
614 __iadd__ = _ProxyIOp(operator.iadd)
615 __isub__ = _ProxyIOp(operator.isub)
616 __imul__ = _ProxyIOp(operator.imul)
617 __imatmul__ = _ProxyIOp(operator.imatmul)
618 __itruediv__ = _ProxyIOp(operator.itruediv)
619 __ifloordiv__ = _ProxyIOp(operator.ifloordiv)
620 __imod__ = _ProxyIOp(operator.imod)
621 __ipow__ = _ProxyIOp(operator.ipow)
622 __ilshift__ = _ProxyIOp(operator.ilshift)
623 __irshift__ = _ProxyIOp(operator.irshift)
624 __iand__ = _ProxyIOp(operator.iand)
625 __ixor__ = _ProxyIOp(operator.ixor)
626 __ior__ = _ProxyIOp(operator.ior)
627 __neg__ = _ProxyLookup(operator.neg)
628 __pos__ = _ProxyLookup(operator.pos)
629 __abs__ = _ProxyLookup(abs)
630 __invert__ = _ProxyLookup(operator.invert)
631 __complex__ = _ProxyLookup(complex)
632 __int__ = _ProxyLookup(int)
633 __float__ = _ProxyLookup(float)
634 __index__ = _ProxyLookup(operator.index)
635 __round__ = _ProxyLookup(round)
636 __trunc__ = _ProxyLookup(math.trunc)
637 __floor__ = _ProxyLookup(math.floor)
638 __ceil__ = _ProxyLookup(math.ceil)
639 __enter__ = _ProxyLookup()
640 __exit__ = _ProxyLookup()
641 __await__ = _ProxyLookup()
642 __aiter__ = _ProxyLookup()
643 __anext__ = _ProxyLookup()
644 __aenter__ = _ProxyLookup()
645 __aexit__ = _ProxyLookup()
646 __copy__ = _ProxyLookup(copy.copy)
647 __deepcopy__ = _ProxyLookup(copy.deepcopy)
648 # __getnewargs_ex__ (pickle through proxy not supported)
649 # __getnewargs__ (pickle)
650 # __getstate__ (pickle)
651 # __setstate__ (pickle)
652 # __reduce__ (pickle)
653 # __reduce_ex__ (pickle)