atexit：如何手动触发它？答案

【问题标题】：atexit: How does one trigger it manually?atexit：如何手动触发它？
【发布时间】：2019-08-23 22:54:44
【问题描述】：

当我运行 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 不会使这个特定的替代方案变得容易） .

【问题讨论】：

标签： python airflow atexit

【解决方案1】：

你可以给 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

【讨论】：

太棒了

【解决方案2】：

如果我没有误解你的问题，你可以通过atexit._run_exitfuncs()触发atexit注册的所有功能：

import atexit


def do_something():
    print('I am doing some work')


def run_on_exit():
    print('I run on exit')


def do_something_else():
    print('I am doing more work')


if __name__ == '__main__':
    atexit.register(run_on_exit)
    do_something()
    atexit._run_exitfuncs()
    do_something_else()

输出显示触发注册的退出函数不会停止程序流，但允许运行更多函数。

【讨论】：