Skip to content

Warning filter

The WarningFilter if used to automatically filter out warnings from a Task as it runs.

This wraps your function in context manager warnings.catch_warnings() and applies your arguments to warnings.filterwarnings(), as you would normally filter warnings in Python.

Usage 

import warnings
from amltk.scheduling import Scheduler
from amltk.scheduling.plugins import WarningFilter

def f() -> None:
    warnings.warn("This is a warning")

scheduler = Scheduler.with_processes(1)
task = scheduler.task(f, plugins=WarningFilter("ignore"))


╭─ Task f() -> 'None' ─────────────────────────────────────────────────────────╮
│ ╭───────────────────────── Plugin warning-filter ──────────────────────────╮ │
│ │  Args         Kwargs                                                     │ │
│ │  ('ignore',)  {}                                                         │ │
│ ╰──────────────────────────────────────────────────────────────────────────╯ │
│ @on_submitted                                                                │
│ @on_done                                                                     │
│ @on_result                                                                   │
│ @on_exception                                                                │
│ @on_cancelled                                                                │
╰─────────────────────────────── Ref: Rkp26xTA ────────────────────────────────╯

class WarningFilter(*args, **kwargs) #

Bases: Plugin

A plugin that disables warnings emitted from tasks.

PARAMETER DESCRIPTION
*args

arguments to pass to warnings.filterwarnings.

TYPE: Any DEFAULT: ()

**kwargs

keyword arguments to pass to warnings.filterwarnings.

TYPE: Any DEFAULT: {}

Source code in src/amltk/scheduling/plugins/warning_filter.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
def __init__(self, *args: Any, **kwargs: Any):
    """Initialize the plugin.

    Args:
        *args: arguments to pass to
            [`warnings.filterwarnings`][warnings.filterwarnings].
        **kwargs: keyword arguments to pass to
            [`warnings.filterwarnings`][warnings.filterwarnings].
    """
    super().__init__()
    self.task: Task | None = None
    self.warning_args = args
    self.warning_kwargs = kwargs

name: ClassVar
classvar attr #

The name of the plugin.

def attach_task(task) #

Attach the plugin to a task.

Source code in src/amltk/scheduling/plugins/warning_filter.py 
94
95
96
97
@override
def attach_task(self, task: Task) -> None:
    """Attach the plugin to a task."""
    self.task = task

def pre_submit(fn, *args, **kwargs) #

Pre-submit hook.

Wraps the function to ignore warnings.

Source code in src/amltk/scheduling/plugins/warning_filter.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
@override
def pre_submit(
    self,
    fn: Callable[P, R],
    *args: P.args,
    **kwargs: P.kwargs,
) -> tuple[Callable[P, R], tuple, dict]:
    """Pre-submit hook.

    Wraps the function to ignore warnings.
    """
    wrapped_f = _IgnoreWarningWrapper(fn, *self.warning_args, **self.warning_kwargs)
    return wrapped_f, args, kwargs

def copy() #

Return a copy of the plugin.

Source code in src/amltk/scheduling/plugins/warning_filter.py 
113
114
115
116
@override
def copy(self) -> Self:
    """Return a copy of the plugin."""
    return self.__class__(*self.warning_args, **self.warning_kwargs)