commit: fa384111eeb216c24cc84d71364a4387fe2b67ee
Author: Brian Harring <ferringb <AT> gmail <DOT> com>
AuthorDate: Sat Dec 13 07:36:03 2025 +0000
Commit: Brian Harring <ferringb <AT> gmail <DOT> com>
CommitDate: Sat Dec 13 19:53:50 2025 +0000
URL:
https://gitweb.gentoo.org/proj/pkgcore/snakeoil.git/commit/?id=fa384111
chore: rewrite suppress_deprecations to handle generators
Warnings filters are dynamic for the context, and generators resume
into the requesting context- not what they started in. It's a known thing,
it's a gotcha that's a PITA.
Either I try to explain the problem in the docstring, or just write code
that suppresses it via injecting a generator 'shape' in front of the
actual generator to suppress, and flip the supression on/off around entry
and exit of the generator.
Tests were added, including asserting the python implementation behavior
this suppresses.
Signed-off-by: Brian Harring <ferringb <AT> gmail.com>
src/snakeoil/deprecation.py | 128 +++++++++++++++++++++++++++++++++++++++-----
tests/test_deprecation.py | 90 +++++++++++++++++++++++++++++++
2 files changed, 205 insertions(+), 13 deletions(-)
diff --git a/src/snakeoil/deprecation.py b/src/snakeoil/deprecation.py
index 380cbb6..8e8e0d2 100644
--- a/src/snakeoil/deprecation.py
+++ b/src/snakeoil/deprecation.py
@@ -17,8 +17,9 @@ when future conditions are met.
__all__ = ("Registry", "RecordCallable", "suppress_deprecations")
-import contextlib
import dataclasses
+import functools
+import inspect
import sys
import typing
import warnings
@@ -26,6 +27,118 @@ import warnings
T = typing.TypeVar("T")
P = typing.ParamSpec("P")
+
+class suppress_deprecations:
+ """Suppress deprecations within this block. Generators and async.Task
require special care to function.
+
+ This cannot be used to decorate a generator function. Using it within a
generator requires explicit code flow for it to work correctly whilst not
causing suppressions outside of the intended usage.
+
+ The cpython warnings filtering is designed around ContextVar- context
specific
+ to a thread, an async.Task, etc. Warnings filtering modifies a context
var thus
+ suppressions are active only within that context. Generators do *not*
bind to any
+ context they started in- whenever they resume, it's resuming in the
context of the thing
+ that resumed them.
+
+ Do not do this in a generator:
+ >>> def f():
+ ... with suppress_deprecations():
+ ... yield invoke_deprecated() # this will be suppressed, but leaks
suppression to what consumed us.
+ ...
+ ... # in resuming, we have no guarantee we're in the same context as
before the yield, where our
+ ... # suppression was added.
+ ... yield invoke_deprecated() # this may or may not be suppressed.
+
+ You have two options. If you do not need fine grained, wrap the
generator; this class will interpose
+ between the generator and consumer and prevent this issue. For example:
+ >>> @suppress_deprecations()
+ ... def f():
+ ... yield invoke_deprecated()
+ ... yield invoke_deprecated()
+
+ If you need the explicit form, use this:
+ >>> def f():
+ ... with suppress_deprecations():
+ ... value = invoke_deprecated() # this will be suppressed
+ ... yield value # we do not force our suppression on the consumer of the
generator
+ ... with suppress_deprecations():
+ ... another_value = invoke_deprecated()
+ ... yield another_value
+ """
+
+ __slots__ = (
+ "_warning_ctx",
+ "kwargs",
+ )
+ _warnings_ctx: None | warnings.catch_warnings
+
+ def __init__(self, category=DeprecationWarning, **kwargs):
+ kwargs.setdefault("action", "ignore")
+ kwargs.setdefault("category", DeprecationWarning)
+ self.kwargs = kwargs
+ self._warning_ctx = None
+
+ def __enter__(self):
+ if self._warning_ctx is not None:
+ raise RuntimeError("this contextmanager has already been entered")
+ self._warning_ctx = warnings.catch_warnings(**self.kwargs)
+ return self._warning_ctx.__enter__()
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ if (ctx := self._warning_ctx) is None:
+ raise RuntimeError("this contextmanager has already exited")
+ ret = ctx.__exit__(exc_type, exc_value, traceback)
+ self._warning_ctx = None
+ return ret
+
+ def __call__(self, thing: typing.Callable[P, T]) -> typing.Callable[P, T]:
+ # being used as a decorator. We unfortunately need to see the actual
call result
+ # to know if it's a generator requiring wrapping.
+ @functools.wraps(thing)
+ def inner(*args: P.args, **kwargs: P.kwargs) -> T:
+ # instantiate a new instance. The callable may result in
re-entrancy.
+ with (ctx := self.__class__(**self.kwargs)):
+ result = thing(*args, **kwargs)
+ if inspect.isgenerator(result):
+ return _GeneratorProxy(result, ctx) # pyright:
ignore[reportReturnType]
+ return result
+
+ return inner
+
+
+class _GeneratorProxy:
+ """Interposing generator. Unfortunately this is required due to how
coroutines work"""
+
+ __slots__ = (
+ "_gen",
+ "_ctx",
+ )
+
+ def __init__(self, gen: typing.Generator, ctx: suppress_deprecations):
+ self._gen = gen
+ self._ctx = ctx
+
+ def __iter__(self):
+ return self
+
+ def __next__(self):
+ with self._ctx:
+ return next(self._gen)
+
+ def send(self, val):
+ with self._ctx:
+ return self._gen.send(val)
+
+ def throw(self, *args):
+ return self._gen.throw(*args)
+
+ def close(self):
+ with self._ctx:
+ self._gen.close()
+
+ def __getattr__(self, attr):
+ return getattr(self._gen, attr)
+
+
Version: typing.TypeAlias = tuple[int, int, int]
warning_category: typing.TypeAlias = type[Warning]
@@ -194,15 +307,7 @@ class Registry:
str(r), category=DeprecationWarning, stacklevel=self.stacklevel + 2
)
- @staticmethod
- @contextlib.contextmanager
- def suppress_deprecations(
- category: warning_category = DeprecationWarning,
- ):
- """Suppress deprecations within this block. Usable as a
contextmanager or decorator"""
- with warnings.catch_warnings():
- warnings.simplefilter(action="ignore", category=category)
- yield
+ suppress_deprecations = staticmethod(suppress_deprecations)
# TODO: py3.13, change this to T per the cvar comments
def __iter__(self) -> typing.Iterator[Record]:
@@ -230,6 +335,3 @@ class Registry:
and python_version >= deprecation.removal_in_py
):
yield deprecation
-
-
-suppress_deprecations = Registry.suppress_deprecations
diff --git a/tests/test_deprecation.py b/tests/test_deprecation.py
index 33d55f3..1f06055 100644
--- a/tests/test_deprecation.py
+++ b/tests/test_deprecation.py
@@ -1,6 +1,8 @@
import dataclasses
+import inspect
import sys
import warnings
+from logging import warning
from pathlib import Path
from textwrap import dedent
@@ -189,6 +191,94 @@ class TestRegistry:
assert w.filename.endswith("/deprecated_import.py")
assert 1 == w.lineno
+ def test_suppress_warnings_generators(self):
+ # See the docstring of suppress_warnings. This asserts that the python
+ # implementation's generator state mutates the local context, rather
than
+ # carrying it's only 'subcontext'. IE, what it does will bleed out the
+ # warning suppression.
+
+ # assert no warning filters are going to screw up this test
requirements
+ with pytest.warns():
+ warnings.warn("must be caught", DeprecationWarning)
+
+ # warnings.catch_warnings() cannot be used for these tests since they
reset the filters to
+ # what the state of it's __enter__. Meaning any warnings mutations
within that block of the generator
+ # get undone if you do this:
+ # >>> with warnings.catch_warnings():
+ # ... next(f) # it added it's suppressions
+ # ... blah # the suppressions from f() got undone by the __exit__ of
catch_warnings
+ warnings_filters = warnings.filters[:]
+
+ def f():
+ with suppress_deprecations():
+ warnings.warn("will be caught", DeprecationWarning)
+ yield
+ warnings.warn(
+ "may not be caught depending on context of resumption",
+ category=DeprecationWarning,
+ )
+
+ i = f()
+ next(i)
+ assert warnings_filters != warnings.filters, (
+ f"generator context modifications were limited to the generator
frame. Is this pypi? a python version >3.15? See test for assumption notes.
Expected {warnings_filters}, got {warnings.filters}"
+ )
+ # exhaust it to exit the generators suppression block.
+ for _ in i:
+ ...
+ assert warnings.filters == warnings_filters
+
+ # ... cool. Now we've asserted the python behavior which is *why* we
have a generator specific
+ # protection built into it. Now to validate that works.
+
+ # simple case. Just yields, not a coroutine
+ @suppress_deprecations()
+ def iterable():
+ warnings.warn("suppressed #1", DeprecationWarning)
+ yield 1
+ warnings.warn("not suppressed", UserWarning)
+ warnings.warn("suppressed #2", DeprecationWarning)
+ yield 2
+ warnings.warn("suppress #3", DeprecationWarning)
+
+ with warnings.catch_warnings(record=True) as w:
+ i = iterable()
+ assert 1 == next(i)
+
+ assert 0 == len(w)
+ with pytest.warns(UserWarning):
+ assert 2 == next(i)
+ assert 0 == len(w)
+ with pytest.raises(StopIteration):
+ next(i)
+ assert 0 == len(w)
+
+ # test coroutines.
+ @suppress_deprecations()
+ def coro():
+ warnings.warn("not suppressed", UserWarning)
+ warnings.warn("suppress #2", DeprecationWarning)
+
+ received = yield 1
+ assert "a1" == received
+ warnings.warn("suppress #3", DeprecationWarning)
+ received = yield 2
+ assert "a2" == received
+ warnings.warn("suppress #3", DeprecationWarning)
+ warnings.warn("not suppressed", UserWarning)
+
+ with warnings.catch_warnings(record=True) as w:
+ gen = coro()
+ assert 0 == len(w) # shouldn't be started by that action alone
+ assert inspect.GEN_CREATED == inspect.getgeneratorstate(gen)
+ with pytest.warns(UserWarning):
+ assert 1 == next(gen) # start it.
+ assert 2 == gen.send("a1")
+ with pytest.warns(UserWarning):
+ with pytest.raises(StopIteration):
+ gen.send("a2")
+ assert 0 == len(w)
+
def test_RecordModule_str():
assert "module='foon.blah', removal in python=3.0.2, reason: why not" ==
str(
