gh-39615: Add warnings.warn() skip_file_prefixes support (#100840)

`warnings.warn()` gains the ability to skip stack frames based on code
filename prefix rather than only a numeric `stacklevel=` via a new
`skip_file_prefixes=` keyword argument.

Doc/library/warnings.rst

@@ -396,7 +396,7 @@ Available Functions
 -------------------
 
 
-.. function:: warn(message, category=None, stacklevel=1, source=None)
+.. function:: warn(message, category=None, stacklevel=1, source=None, \*, skip_file_prefixes=None)
 
    Issue a warning, or maybe ignore it or raise an exception.  The *category*
    argument, if given, must be a :ref:`warning category class <warning-categories>`; it
@@ -407,12 +407,39 @@ Available Functions
    :ref:`warnings filter <warning-filter>`.  The *stacklevel* argument can be used by wrapper
    functions written in Python, like this::
 
-      def deprecation(message):
+      def deprecated_api(message):
           warnings.warn(message, DeprecationWarning, stacklevel=2)
 
-   This makes the warning refer to :func:`deprecation`'s caller, rather than to the
-   source of :func:`deprecation` itself (since the latter would defeat the purpose
-   of the warning message).
+   This makes the warning refer to ``deprecated_api``'s caller, rather than to
+   the source of ``deprecated_api`` itself (since the latter would defeat the
+   purpose of the warning message).
+
+   The *skip_file_prefixes* keyword argument can be used to indicate which
+   stack frames are ignored when counting stack levels. This can be useful when
+   you want the warning to always appear at call sites outside of a package
+   when a constant *stacklevel* does not fit all call paths or is otherwise
+   challenging to maintain. If supplied, it must be a tuple of strings. When
+   prefixes are supplied, stacklevel is implicitly overridden to be ``max(2,
+   stacklevel)``. To cause a warning to be attributed to the caller from
+   outside of the current package you might write::
+
+      # example/lower.py
+      _warn_skips = (os.path.dirname(__file__),)
+
+      def one_way(r_luxury_yacht=None, t_wobbler_mangrove=None):
+          if r_luxury_yacht:
+              warnings.warn("Please migrate to t_wobbler_mangrove=.",
+                            skip_file_prefixes=_warn_skips)
+
+      # example/higher.py
+      from . import lower
+
+      def another_way(**kw):
+          lower.one_way(**kw)
+
+   This makes the warning refer to both the ``example.lower.one_way()`` and
+   ``package.higher.another_way()`` call sites only from calling code living
+   outside of ``example`` package.
 
    *source*, if supplied, is the destroyed object which emitted a
    :exc:`ResourceWarning`.
@@ -420,6 +447,9 @@ Available Functions
    .. versionchanged:: 3.6
       Added *source* parameter.
 
+   .. versionchanged:: 3.12
+      Added *skip_file_prefixes*.
+
 
 .. function:: warn_explicit(message, category, filename, lineno, module=None, registry=None, module_globals=None, source=None)
 

Include/internal/pycore_global_objects_fini_generated.h

@@ -1151,6 +1151,7 @@ _PyStaticObjects_CheckRefcnt(PyInterpreterState *interp) {
     _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(signed));
     _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(size));
     _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(sizehint));
+    _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(skip_file_prefixes));
     _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(sleep));
     _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(sock));
     _PyStaticObject_CheckRefcnt((PyObject *)&_Py_ID(sort));

Include/internal/pycore_global_strings.h

@@ -637,6 +637,7 @@ struct _Py_global_strings {
         STRUCT_FOR_ID(signed)
         STRUCT_FOR_ID(size)
         STRUCT_FOR_ID(sizehint)
+        STRUCT_FOR_ID(skip_file_prefixes)
         STRUCT_FOR_ID(sleep)
         STRUCT_FOR_ID(sock)
         STRUCT_FOR_ID(sort)

Include/internal/pycore_runtime_init_generated.h

@@ -1143,6 +1143,7 @@ extern "C" {
     INIT_ID(signed), \
     INIT_ID(size), \
     INIT_ID(sizehint), \
+    INIT_ID(skip_file_prefixes), \
     INIT_ID(sleep), \
     INIT_ID(sock), \
     INIT_ID(sort), \

Include/internal/pycore_unicodeobject_generated.h

@@ -1180,6 +1180,8 @@ _PyUnicode_InitStaticStrings(void) {
     PyUnicode_InternInPlace(&string);
     string = &_Py_ID(sizehint);
     PyUnicode_InternInPlace(&string);
+    string = &_Py_ID(skip_file_prefixes);
+    PyUnicode_InternInPlace(&string);
     string = &_Py_ID(sleep);
     PyUnicode_InternInPlace(&string);
     string = &_Py_ID(sock);

Lib/test/test_warnings/__init__.py

@@ -12,6 +12,7 @@
 from test.support import warnings_helper
 from test.support.script_helper import assert_python_ok, assert_python_failure
 
+from test.test_warnings.data import package_helper
 from test.test_warnings.data import stacklevel as warning_tests
 
 import warnings as original_warnings
@@ -472,6 +473,42 @@ def test_stacklevel_import(self):
                 self.assertEqual(len(w), 1)
                 self.assertEqual(w[0].filename, __file__)
 
+    def test_skip_file_prefixes(self):
+        with warnings_state(self.module):
+            with original_warnings.catch_warnings(record=True,
+                    module=self.module) as w:
+                self.module.simplefilter('always')
+
+                # Warning never attributed to the data/ package.
+                package_helper.inner_api(
+                        "inner_api", stacklevel=2,
+                        warnings_module=warning_tests.warnings)
+                self.assertEqual(w[-1].filename, __file__)
+                warning_tests.package("package api", stacklevel=2)
+                self.assertEqual(w[-1].filename, __file__)
+                self.assertEqual(w[-2].filename, w[-1].filename)
+                # Low stacklevels are overridden to 2 behavior.
+                warning_tests.package("package api 1", stacklevel=1)
+                self.assertEqual(w[-1].filename, __file__)
+                warning_tests.package("package api 0", stacklevel=0)
+                self.assertEqual(w[-1].filename, __file__)
+                warning_tests.package("package api -99", stacklevel=-99)
+                self.assertEqual(w[-1].filename, __file__)
+
+                # The stacklevel still goes up out of the package.
+                warning_tests.package("prefix02", stacklevel=3)
+                self.assertIn("unittest", w[-1].filename)
+
+    def test_skip_file_prefixes_type_errors(self):
+        with warnings_state(self.module):
+            warn = warning_tests.warnings.warn
+            with self.assertRaises(TypeError):
+                warn("msg", skip_file_prefixes=[])
+            with self.assertRaises(TypeError):
+                warn("msg", skip_file_prefixes=(b"bytes",))
+            with self.assertRaises(TypeError):
+                warn("msg", skip_file_prefixes="a sequence of strs")
+
     def test_exec_filename(self):
         filename = "<warnings-test>"
         codeobj = compile(("import warnings\n"
@@ -895,7 +932,7 @@ def test_formatwarning(self):
         message = "msg"
         category = Warning
         file_name = os.path.splitext(warning_tests.__file__)[0] + '.py'
-        line_num = 3
+        line_num = 5
         file_line = linecache.getline(file_name, line_num).strip()
         format = "%s:%s: %s: %s\n  %s\n"
         expect = format % (file_name, line_num, category.__name__, message,

Lib/test/test_warnings/data/package_helper.py

@@ -0,0 +1,10 @@
+# helper to the helper for testing skip_file_prefixes.
+
+import os
+
+package_path = os.path.dirname(__file__)
+
+def inner_api(message, *, stacklevel, warnings_module):
+    warnings_module.warn(
+            message, stacklevel=stacklevel,
+            skip_file_prefixes=(package_path,))

Lib/test/test_warnings/data/stacklevel.py

@@ -1,9 +1,15 @@
-# Helper module for testing the skipmodules argument of warnings.warn()
+# Helper module for testing stacklevel and skip_file_prefixes arguments
+# of warnings.warn()
 
 import warnings
+from test.test_warnings.data import package_helper
 
 def outer(message, stacklevel=1):
     inner(message, stacklevel)
 
 def inner(message, stacklevel=1):
     warnings.warn(message, stacklevel=stacklevel)
+
+def package(message, *, stacklevel):
+    package_helper.inner_api(message, stacklevel=stacklevel,
+                             warnings_module=warnings)

Lib/warnings.py

@@ -269,22 +269,32 @@ def _getcategory(category):
     return cat
 
 
-def _is_internal_frame(frame):
-    """Signal whether the frame is an internal CPython implementation detail."""
-    filename = frame.f_code.co_filename
+def _is_internal_filename(filename):
     return 'importlib' in filename and '_bootstrap' in filename
 
 
-def _next_external_frame(frame):
-    """Find the next frame that doesn't involve CPython internals."""
+def _is_filename_to_skip(filename, skip_file_prefixes):
+    return any(filename.startswith(prefix) for prefix in skip_file_prefixes)
+
+
+def _is_internal_frame(frame):
+    """Signal whether the frame is an internal CPython implementation detail."""
+    return _is_internal_filename(frame.f_code.co_filename)
+
+
+def _next_external_frame(frame, skip_file_prefixes):
+    """Find the next frame that doesn't involve Python or user internals."""
     frame = frame.f_back
-    while frame is not None and _is_internal_frame(frame):
+    while frame is not None and (
+            _is_internal_filename(filename := frame.f_code.co_filename) or
+            _is_filename_to_skip(filename, skip_file_prefixes)):
         frame = frame.f_back
     return frame
 
 
 # Code typically replaced by _warnings
-def warn(message, category=None, stacklevel=1, source=None):
+def warn(message, category=None, stacklevel=1, source=None,
+         *, skip_file_prefixes=()):
     """Issue a warning, or maybe ignore it or raise an exception."""
     # Check if message is already a Warning object
     if isinstance(message, Warning):
@@ -295,6 +305,11 @@ def warn(message, category=None, stacklevel=1, source=None):
     if not (isinstance(category, type) and issubclass(category, Warning)):
         raise TypeError("category must be a Warning subclass, "
                         "not '{:s}'".format(type(category).__name__))
+    if not isinstance(skip_file_prefixes, tuple):
+        # The C version demands a tuple for implementation performance.
+        raise TypeError('skip_file_prefixes must be a tuple of strs.')
+    if skip_file_prefixes:
+        stacklevel = max(2, stacklevel)
     # Get context information
     try:
         if stacklevel <= 1 or _is_internal_frame(sys._getframe(1)):
@@ -305,7 +320,7 @@ def warn(message, category=None, stacklevel=1, source=None):
             frame = sys._getframe(1)
             # Look for one frame less since the above line starts us off.
             for x in range(stacklevel-1):
-                frame = _next_external_frame(frame)
+                frame = _next_external_frame(frame, skip_file_prefixes)
                 if frame is None:
                     raise ValueError
     except ValueError:

Misc/NEWS.d/next/Library/2023-01-08-00-12-44.gh-issue-39615.gn4PhB.rst

@@ -0,0 +1,3 @@
+:func:`warnings.warn` now has the ability to skip stack frames based on code
+filename prefix rather than only a numeric ``stacklevel`` via the new
+``skip_file_prefixes`` keyword argument.

Python/_warnings.c

@@ -761,56 +761,99 @@ warn_explicit(PyThreadState *tstate, PyObject *category, PyObject *message,
     return result;  /* Py_None or NULL. */
 }
 
-static int
-is_internal_frame(PyFrameObject *frame)
+static PyObject *
+get_frame_filename(PyFrameObject *frame)
 {
-    if (frame == NULL) {
-        return 0;
-    }
-
     PyCodeObject *code = PyFrame_GetCode(frame);
     PyObject *filename = code->co_filename;
     Py_DECREF(code);
+    return filename;
+}
 
-    if (filename == NULL) {
-        return 0;
-    }
+static bool
+is_internal_filename(PyObject *filename)
+{
     if (!PyUnicode_Check(filename)) {
-        return 0;
+        return false;
     }
 
     int contains = PyUnicode_Contains(filename, &_Py_ID(importlib));
     if (contains < 0) {
-        return 0;
+        return false;
     }
     else if (contains > 0) {
         contains = PyUnicode_Contains(filename, &_Py_ID(_bootstrap));
         if (contains < 0) {
-            return 0;
+            return false;
         }
         else if (contains > 0) {
-            return 1;
+            return true;
         }
     }
 
-    return 0;
+    return false;
+}
+
+static bool
+is_filename_to_skip(PyObject *filename, PyTupleObject *skip_file_prefixes)
+{
+    if (skip_file_prefixes) {
+        if (!PyUnicode_Check(filename)) {
+            return false;
+        }
+
+        Py_ssize_t prefixes = PyTuple_GET_SIZE(skip_file_prefixes);
+        for (Py_ssize_t idx = 0; idx < prefixes; ++idx)
+        {
+            PyObject *prefix = PyTuple_GET_ITEM(skip_file_prefixes, idx);
+            int found = PyUnicode_Tailmatch(filename, prefix, 0, -1, -1);
+            if (found == 1) {
+                return true;
+            }
+            if (found < 0) {
+                return false;
+            }
+        }
+    }
+    return false;
+}
+
+static bool
+is_internal_frame(PyFrameObject *frame)
+{
+    if (frame == NULL) {
+        return false;
+    }
+
+    PyObject *filename = get_frame_filename(frame);
+    if (filename == NULL) {
+        return false;
+    }
+
+    return is_internal_filename(filename);
 }
 
 static PyFrameObject *
-next_external_frame(PyFrameObject *frame)
+next_external_frame(PyFrameObject *frame, PyTupleObject *skip_file_prefixes)
 {
+    PyObject *frame_filename;
     do {
         PyFrameObject *back = PyFrame_GetBack(frame);
         Py_SETREF(frame, back);
-    } while (frame != NULL && is_internal_frame(frame));
+    } while (frame != NULL && (frame_filename = get_frame_filename(frame)) &&
+             (is_internal_filename(frame_filename) ||
+              is_filename_to_skip(frame_filename, skip_file_prefixes)));
 
     return frame;
 }
 
 /* filename, module, and registry are new refs, globals is borrowed */
+/* skip_file_prefixes is either NULL or a tuple of strs. */
 /* Returns 0 on error (no new refs), 1 on success */
 static int
-setup_context(Py_ssize_t stack_level, PyObject **filename, int *lineno,
+setup_context(Py_ssize_t stack_level,
+              PyTupleObject *skip_file_prefixes,
+              PyObject **filename, int *lineno,
               PyObject **module, PyObject **registry)
 {
     PyObject *globals;
@@ -820,6 +863,21 @@ setup_context(Py_ssize_t stack_level, PyObject **filename, int *lineno,
     if (tstate == NULL) {
         return 0;
     }
+    if (skip_file_prefixes) {
+        /* Type check our data structure up front. Later code that uses it
+         * isn't structured to report errors. */
+        Py_ssize_t prefixes = PyTuple_GET_SIZE(skip_file_prefixes);
+        for (Py_ssize_t idx = 0; idx < prefixes; ++idx)
+        {
+            PyObject *prefix = PyTuple_GET_ITEM(skip_file_prefixes, idx);
+            if (!PyUnicode_Check(prefix)) {
+                PyErr_Format(PyExc_TypeError,
+                             "Found non-str '%s' in skip_file_prefixes.",
+                             Py_TYPE(prefix)->tp_name);
+                return 0;
+            }
+        }
+    }
     PyInterpreterState *interp = tstate->interp;
     PyFrameObject *f = PyThreadState_GetFrame(tstate);
     // Stack level comparisons to Python code is off by one as there is no
@@ -832,7 +890,7 @@ setup_context(Py_ssize_t stack_level, PyObject **filename, int *lineno,
     }
     else {
         while (--stack_level > 0 && f != NULL) {
-            f = next_external_frame(f);
+            f = next_external_frame(f, skip_file_prefixes);
         }
     }
 
@@ -925,7 +983,7 @@ get_category(PyObject *message, PyObject *category)
 
 static PyObject *
 do_warn(PyObject *message, PyObject *category, Py_ssize_t stack_level,
-        PyObject *source)
+        PyObject *source, PyTupleObject *skip_file_prefixes)
 {
     PyObject *filename, *module, *registry, *res;
     int lineno;
@@ -935,7 +993,8 @@ do_warn(PyObject *message, PyObject *category, Py_ssize_t stack_level,
         return NULL;
     }
 
-    if (!setup_context(stack_level, &filename, &lineno, &module, &registry))
+    if (!setup_context(stack_level, skip_file_prefixes,
+                       &filename, &lineno, &module, &registry))
         return NULL;
 
     res = warn_explicit(tstate, category, message, filename, lineno, module, registry,
@@ -950,22 +1009,42 @@ do_warn(PyObject *message, PyObject *category, Py_ssize_t stack_level,
 warn as warnings_warn
 
     message: object
+      Text of the warning message.
     category: object = None
+      The Warning category subclass. Defaults to UserWarning.
     stacklevel: Py_ssize_t = 1
+      How far up the call stack to make this warning appear. A value of 2 for
+      example attributes the warning to the caller of the code calling warn().
     source: object = None
+      If supplied, the destroyed object which emitted a ResourceWarning
+    *
+    skip_file_prefixes: object(type='PyTupleObject *', subclass_of='&PyTuple_Type') = NULL
+      An optional tuple of module filename prefixes indicating frames to skip
+      during stacklevel computations for stack frame attribution.
 
 Issue a warning, or maybe ignore it or raise an exception.
 [clinic start generated code]*/
 
 static PyObject *
 warnings_warn_impl(PyObject *module, PyObject *message, PyObject *category,
-                   Py_ssize_t stacklevel, PyObject *source)
-/*[clinic end generated code: output=31ed5ab7d8d760b2 input=bfdf5cf99f6c4edd]*/
+                   Py_ssize_t stacklevel, PyObject *source,
+                   PyTupleObject *skip_file_prefixes)
+/*[clinic end generated code: output=a68e0f6906c65f80 input=eb37c6a18bec4ea1]*/
 {
     category = get_category(message, category);
     if (category == NULL)
         return NULL;
-    return do_warn(message, category, stacklevel, source);
+    if (skip_file_prefixes) {
+        if (PyTuple_GET_SIZE(skip_file_prefixes) > 0) {
+            if (stacklevel < 2) {
+                stacklevel = 2;
+            }
+        } else {
+            Py_DECREF((PyObject *)skip_file_prefixes);
+            skip_file_prefixes = NULL;
+        }
+    }
+    return do_warn(message, category, stacklevel, source, skip_file_prefixes);
 }
 
 static PyObject *
@@ -1113,7 +1192,7 @@ warn_unicode(PyObject *category, PyObject *message,
     if (category == NULL)
         category = PyExc_RuntimeWarning;
 
-    res = do_warn(message, category, stack_level, source);
+    res = do_warn(message, category, stack_level, source, NULL);
     if (res == NULL)
         return -1;
     Py_DECREF(res);

Python/clinic/_warnings.c.h

@@ -9,17 +9,32 @@ preserve
 
 
 PyDoc_STRVAR(warnings_warn__doc__,
-"warn($module, /, message, category=None, stacklevel=1, source=None)\n"
+"warn($module, /, message, category=None, stacklevel=1, source=None, *,\n"
+"     skip_file_prefixes=<unrepresentable>)\n"
 "--\n"
 "\n"
-"Issue a warning, or maybe ignore it or raise an exception.");
+"Issue a warning, or maybe ignore it or raise an exception.\n"
+"\n"
+"  message\n"
+"    Text of the warning message.\n"
+"  category\n"
+"    The Warning category subclass. Defaults to UserWarning.\n"
+"  stacklevel\n"
+"    How far up the call stack to make this warning appear. A value of 2 for\n"
+"    example attributes the warning to the caller of the code calling warn().\n"
+"  source\n"
+"    If supplied, the destroyed object which emitted a ResourceWarning\n"
+"  skip_file_prefixes\n"
+"    An optional tuple of module filename prefixes indicating frames to skip\n"
+"    during stacklevel computations for stack frame attribution.");
 
 #define WARNINGS_WARN_METHODDEF    \
     {"warn", _PyCFunction_CAST(warnings_warn), METH_FASTCALL|METH_KEYWORDS, warnings_warn__doc__},
 
 static PyObject *
 warnings_warn_impl(PyObject *module, PyObject *message, PyObject *category,
-                   Py_ssize_t stacklevel, PyObject *source);
+                   Py_ssize_t stacklevel, PyObject *source,
+                   PyTupleObject *skip_file_prefixes);
 
 static PyObject *
 warnings_warn(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
@@ -27,14 +42,14 @@ warnings_warn(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObjec
     PyObject *return_value = NULL;
     #if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
 
-    #define NUM_KEYWORDS 4
+    #define NUM_KEYWORDS 5
     static struct {
         PyGC_Head _this_is_not_used;
         PyObject_VAR_HEAD
         PyObject *ob_item[NUM_KEYWORDS];
     } _kwtuple = {
         .ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
-        .ob_item = { &_Py_ID(message), &_Py_ID(category), &_Py_ID(stacklevel), &_Py_ID(source), },
+        .ob_item = { &_Py_ID(message), &_Py_ID(category), &_Py_ID(stacklevel), &_Py_ID(source), &_Py_ID(skip_file_prefixes), },
     };
     #undef NUM_KEYWORDS
     #define KWTUPLE (&_kwtuple.ob_base.ob_base)
@@ -43,19 +58,20 @@ warnings_warn(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObjec
     #  define KWTUPLE NULL
     #endif  // !Py_BUILD_CORE
 
-    static const char * const _keywords[] = {"message", "category", "stacklevel", "source", NULL};
+    static const char * const _keywords[] = {"message", "category", "stacklevel", "source", "skip_file_prefixes", NULL};
     static _PyArg_Parser _parser = {
         .keywords = _keywords,
         .fname = "warn",
         .kwtuple = KWTUPLE,
     };
     #undef KWTUPLE
-    PyObject *argsbuf[4];
+    PyObject *argsbuf[5];
     Py_ssize_t noptargs = nargs + (kwnames ? PyTuple_GET_SIZE(kwnames) : 0) - 1;
     PyObject *message;
     PyObject *category = Py_None;
     Py_ssize_t stacklevel = 1;
     PyObject *source = Py_None;
+    PyTupleObject *skip_file_prefixes = NULL;
 
     args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 4, 0, argsbuf);
     if (!args) {
@@ -88,9 +104,23 @@ warnings_warn(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObjec
             goto skip_optional_pos;
         }
     }
-    source = args[3];
+    if (args[3]) {
+        source = args[3];
+        if (!--noptargs) {
+            goto skip_optional_pos;
+        }
+    }
 skip_optional_pos:
-    return_value = warnings_warn_impl(module, message, category, stacklevel, source);
+    if (!noptargs) {
+        goto skip_optional_kwonly;
+    }
+    if (!PyTuple_Check(args[4])) {
+        _PyArg_BadArgument("warn", "argument 'skip_file_prefixes'", "tuple", args[4]);
+        goto exit;
+    }
+    skip_file_prefixes = (PyTupleObject *)args[4];
+skip_optional_kwonly:
+    return_value = warnings_warn_impl(module, message, category, stacklevel, source, skip_file_prefixes);
 
 exit:
     return return_value;
@@ -216,4 +246,4 @@ warnings_filters_mutated(PyObject *module, PyObject *Py_UNUSED(ignored))
 {
     return warnings_filters_mutated_impl(module);
 }
-/*[clinic end generated code: output=0d264d1ddfc37100 input=a9049054013a1b77]*/
+/*[clinic end generated code: output=20429719d7223bdc input=a9049054013a1b77]*/