python - atexit:如何手动触发它？

当我运行python脚本时，我可以退出解释器，并且atexit将执行我已注册的所有功能。

现在，我正在使用气流，并想触发atexit任务on_kill()（即当我清除或杀死气流中的dag节点时）。

例如，在伪代码中，我需要能够：

class Foo(PythonOperator):
  ...
  def on_kill():
     # somehow, trigger atexit tasks without exiting the
     # process entirely

atexit也不是必需的-我可以做其他事情。最主要的一点是，需要以程序方式杀死在python上下文之外执行的某些事情，并且理想情况下，通过参考外壳脚本传递kill函数将是最后的选择（python不会使这种特殊选择变得容易） 。

最佳答案

您可以猴子补丁atexit模块，如下所示：

import atexit
from queue import LifoQueue


save_register = atexit.register
atexit_queue = LifoQueue()

def my_register(func, *args, **kwargs):
    save_register(func, *args, **kwargs)
    atexit_queue.put((func, args, kwargs))

atexit.register = my_register


if __name__ == '__main__':

    def func1():
        print('func1() called')

    def func2(arg):
        print(f'func2({arg}) called')

    def func3(arg, kwarg1=1, kwarg2='foo'):
        print(f'func3({arg}, kwarg1={kwarg1}, kwarg2={kwarg2!r}) called')

    atexit.register(func1)
    atexit.register(func2, 1)
    atexit.register(func3, 2, kwarg1=42, kwarg2='bar')

    print('Calling queued atexit functions:\n')
    while atexit_queue.qsize():
        func, args, kwargs = atexit_queue.get()
        atexit.unregister(func)  # Prevent it from being called again.
        func(*args, **kwargs)

输出：

Calling queued atexit functions:

func3(2, kwarg1=42, kwarg2='bar') called
func2(1) called
func1() called

关于python - atexit:如何手动触发它？，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/57633815/