Last will and testament...

Here is an example code snippet that executes 'last rites'. My current use-case is to handle final tasks on a cluster job management system when the script violates some execution criterion e.g. exceeding memory quota. It's helpful if the job could perform some task(s) that indicate that it had to be terminated.

The key module here is the Python signal package which provides a way to handle various system signals. The example here sends an email whenever the script is terminated. To use signal you need to register one or more signals to have a handle i.e. a callback to be executed if that signal is raised. For example:

signal.signal(signal.SIGINT, handler)

registers the handler function when the script receives an keyboard interrupt (Control-C) event.

You can view a full list of signals here.

# -*- coding: utf-8 -*-
import os
import signal
import smtplib
import socket
import sys
# import time
from email.mime.text import MIMEText

signal_dict = {getattr(signal, s): s for s in dir(signal) if s.startswith('SIG') and not s.startswith('SIG_')}


def send_email(t, s, b):
    """Send email"""
    f = "[email protected]"
    msg = MIMEText(b)
    msg['To'] = t
    msg['From'] = f
    msg['Subject'] = s
    s = smtplib.SMTP(
        "email.hostname",
        "email.port",
    )
    s.sendmail(f, [t], msg.as_string())
    s.quit()


def handler(signum, frame):
    """Handle signal"""
    send_email(
        '[email protected]',
        'received signal {} on {}'.format(signal_dict[signum], socket.gethostname()),
        'terminated',
    )
    raise Exception


def main():
    # register signal handlers
    signal.signal(signal.SIGINT, handler)
    signal.signal(signal.SIGQUIT, handler)
    signal.signal(signal.SIGILL, handler)
    signal.signal(signal.SIGABRT, handler)
    signal.signal(signal.SIGFPE, handler)
    signal.signal(signal.SIGSEGV, handler)
    signal.signal(signal.SIGPIPE, handler)
    signal.signal(signal.SIGALRM, handler)
    signal.signal(signal.SIGTERM, handler)
    # do something
    send_email(
        '[email protected]',
        'starting on host {}'.format(socket.gethostname()),
        'started...',
    )
    # simulate a long running task that gets manually killed
    # time.sleep(100)
    # simulate a task that violates the memory limit
    with open(sys.argv[1], 'rb') as f:
        g = f.readlines()
        h = g[::]
        i = h[::]
    send_email(
        '[email protected]',
        'successfully completed on host {}'.format(socket.gethostname()),
        'completed...',
    )
    return os.EX_OK


if __name__ == "__main__":
    sys.exit(main())