daemon.py

#!/usr/bin/env python
# -*- coding: utf8 -*-

import atexit
import os
import sys
import time
from signal import SIGINT, SIGTERM


class Daemon(object):
    """
    A generic daemon class.

    Usage: subclass the Daemon class and override the run() method
    """

    startmsg = "started with pid %s"

    def __init__(
        self, pidfile, stdin="/dev/null", stdout="/dev/null", stderr="/dev/null"
    ):
        self.stdin = stdin
        self.stdout = stdout
        self.stderr = stderr
        self.pidfile = pidfile

    def daemonize(self):
        """
        do the UNIX double-fork magic, see Stevens' "Advanced
        Programming in the UNIX Environment" for details (ISBN 0201563177)
        http://www.erlenstar.demon.co.uk/unix/faq_2.html#SEC16
        """
        try:
            pid = os.fork()
            if pid > 0:
                # exit first parent
                sys.exit(0)
        except OSError as e:
            sys.stderr.write(f"fork #1 failed: {e.errno} ({e.strerror})\n")
            sys.exit(1)

        # decouple from parent environment
        os.chdir("/")
        os.setsid()
        os.umask(0)

        # do second fork
        try:
            pid = os.fork()
            if pid > 0:
                # exit from second parent
                sys.exit(0)
        except OSError as e:
            sys.stderr.write(f"fork #2 failed: {e.errno} ({e.strerror})\n")
            sys.exit(1)

        # redirect standard file descriptors
        si = open(self.stdin, "r", encoding="utf-8")
        so = open(self.stdout, "a+")
        se = open(self.stderr, "ab+", 0)

        pid = str(os.getpid())

        sys.stderr.write("\n%s\n" % self.startmsg % pid)
        sys.stderr.flush()

        if self.pidfile:
            open(self.pidfile, "w+").write(f"{pid}\n")

        atexit.register(self.delpid)
        os.dup2(si.fileno(), sys.stdin.fileno())
        os.dup2(so.fileno(), sys.stdout.fileno())
        os.dup2(se.fileno(), sys.stderr.fileno())

    def delpid(self):
        try:
            os.remove(self.pidfile)
        except OSError:
            pass

    def start(self):
        """
        Start the daemon
        """
        # Check for a pidfile to see if the daemon already runs
        try:
            pf = open(self.pidfile, "r")
            pid = int(pf.read().strip())
            pf.close()
        except IOError:
            pid = None
        except SystemExit:
            pid = None

        if pid:
            message = f"pidfile {self.pidfile} already exist. Daemon already running?\n"
            sys.stderr.write(message)
            sys.exit(1)

        # Start the daemon
        self.daemonize()
        self.run()

    def get_pid(self):
        # Get the pid from the pidfile
        try:
            pf = open(self.pidfile, "r")
            pid = int(pf.read().strip())
            pf.close()
        except IOError:
            pid = None
        except SystemExit:
            pid = None
        return pid

    def stop(self):
        """
        Stop the daemon
        """
        pid = self.get_pid()

        if not pid:
            message = f"pidfile {self.pidfile} does not exist. Daemon not running?\n"
            sys.stderr.write(message)
            return  # not an error in a restart

        # Try killing the daemon process
        try:
            while 1:
                os.kill(pid, SIGTERM)
                time.sleep(0.1)
        except OSError as err:
            err = str(err)
            if err.find("No such process") > 0:
                if os.path.exists(self.pidfile):
                    self.delpid()
                else:
                    print(str(err))
                    sys.exit(1)

    def restart(self):
        """
        Restart the daemon
        """
        self.stop()
        self.start()

    def run(self):
        """
        You should override this method when you subclass Daemon. It will be called after the process has been
        daemonized by start() or restart().
        """
        pass

这个项目是来源于 daemon 这个仓库，作者是在Python 2.x版本的时候写的，我做了一些 Python 3.x 版的移植，并且修复了一些bug，这个项目可以用于编写常见的轻量web框架的后台服务管理脚本，比安装一个supervisor更易于维护。

以下是使用fastapi框架跑后台服务的代码示例

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


import logging
import sys

import uvicorn

import app

PIDFILE = "/tmp/test.pid"
LOGFILE = "/tmp/test.log"

logging.basicConfig(filename=LOGFILE, level=logging.DEBUG)


class WebDaemon(app.Daemon):
    def run(self):
        config = uvicorn.Config("main:app", port=5000, log_level="debug")
        server = uvicorn.Server(config)
        server.run()


if __name__ == "__main__":
    daemon = WebDaemon(PIDFILE)

    if len(sys.argv) == 2:
        if sys.argv[1] == "start":
            try:
                daemon.start()
            except SystemExit:
                pass
            except Exception:
                logging.exception("")
        elif sys.argv[1] == "stop":
            print("Stopping ...")
            daemon.stop()
        elif sys.argv[1] == "restart":
            print("Restaring ...")
            daemon.restart()
        elif sys.argv[1] == "status":
            try:
                pf = open(PIDFILE, "r")
                pid = int(pf.read().strip())
                pf.close()
            except IOError:
                pid = None
            except SystemExit:
                pid = None

            if pid:
                print(f"web app is running as pid {pid}")
            else:
                print("web app is not running.")
        else:
            print("Unknown command")
            sys.exit(2)
            sys.exit(0)
    else:
        print("Usage: python {} start|stop|restart|status".format(sys.argv[0]))
        sys.exit(2)