Skip to content

Instantly share code, notes, and snippets.

Show Gist options
  • Save kylemanna/d193aaa6b33a89f649524ad27ce47c4b to your computer and use it in GitHub Desktop.
Save kylemanna/d193aaa6b33a89f649524ad27ce47c4b to your computer and use it in GitHub Desktop.
An example network service with systemd-activated socket in Python. #systemd #python #socket #socket-activation

README

The example below creates a TCP server listening on a stream (i.e. SOCK_STREAM) socket. A similar approach can be followed to create a UDP server on a datagram (i.e. SOCK_DGRAM) socket. See man systemd.socket for details.

An example server

Create an simple echo server at ~/tmp/foo/serve.py.

#!/usr/bin/env python3

from socketserver import TCPServer, StreamRequestHandler
import socket
import logging

class Handler(StreamRequestHandler):
    def handle(self):
        self.data = self.rfile.readline().strip()
        logging.info("From <%s>: %s" % (self.client_address, self.data))
        self.wfile.write(self.data.upper() + "\r\n".encode("utf-8"))

class Server(TCPServer):
    
    # The constant would be better initialized by a systemd module
    SYSTEMD_FIRST_SOCKET_FD = 3

    def __init__(self, server_address, handler_cls):
        # Invoke base but omit bind/listen steps (performed by systemd activation!)
        TCPServer.__init__(
            self, server_address, handler_cls, bind_and_activate=False)
        # Override socket
        self.socket = socket.fromfd(
            self.SYSTEMD_FIRST_SOCKET_FD, self.address_family, self.socket_type)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    HOST, PORT = "localhost", 9999 # not really needed here
    server = Server((HOST, PORT), Handler)
    server.serve_forever()

Define the systemd service unit

Create the service definition unit file at ~/.config/systemd/user/foo.service (note that this unit refers to foo.socket):

[Unit]
Description=Foo Service
After=network.target foo.socket
Requires=foo.socket

[Service]
Type=simple
ExecStart=/usr/bin/python3 %h/tmp/foo/serve.py
TimeoutStopSec=5

[Install]
WantedBy=default.target

Define the systemd socket unit

Create the socket definition unit file at ~/.config/systemd/user/foo.socket:

[Unit]
Description=Foo Socket
PartOf=foo.service

[Socket]
ListenStream=127.0.0.1:9999

[Install]
WantedBy=sockets.target

Start Listening Socket

systemctl --user daemon-reload
systemctl --user start foo.socket

Test it

Send a message to the listening service:

date | netcat 127.0.0.1 9999

Watch the log for the echo:

journalctl -f --user-unit foo.service

Changelog

Forked by @kylemanna:

  • Convert docs to use the systemd user daemon
  • Update to work with python3
  • More logs
@Et7f3
Copy link

Et7f3 commented Oct 29, 2023

See this is you want to avoid hardcoding 3 https://github.com/systemd/python-systemd/blob/3f0801c8ac1008fc4c5e3b6a7b2b68e711ca67b2/systemd/daemon.py#L56 (and should be better in case more fd are added)

@NathanAdhitya
Copy link

Are there any resources on a way to automatically shutdown the Python server when it's not used?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment