### Build and Install Package for Development Source: https://github.com/systemd/python-systemd/blob/main/README.md Build the Python package using 'build' and install it locally using pip. This allows for testing the package in a development environment. ```bash python -m build python -m pip install . ``` -------------------------------- ### Monitor Session Events with Context Manager Source: https://context7.com/systemd/python-systemd/llms.txt Utilize login.Monitor as a context manager to automatically handle setup and cleanup for monitoring session events. The poll timeout is set to 2000ms. ```python import select from systemd import login # Context manager — automatically closes on exit with login.Monitor("session") as mon: p2 = select.poll() p2.register(mon, mon.get_events()) p2.poll(2000) mon.flush() print("Sessions now:", login.sessions()) ``` -------------------------------- ### Poll for systemd login events Source: https://github.com/systemd/python-systemd/blob/main/docs/login.md Integrate systemd login event monitoring into an external event loop using `select.poll`. This example demonstrates polling for machine events. ```python import select from systemd import login m = login.Monitor("machine") p = select.poll() p.register(m, m.get_events()) login.machine_names() [] p.poll() [(3, 1)] login.machine_names() ['fedora-25'] ``` -------------------------------- ### Send notification with file descriptors Source: https://context7.com/systemd/python-systemd/llms.txt Sends a notification to systemd along with file descriptors. This example binds a socket and sends its file descriptor, allowing systemd to manage or access it. Requires `FDSTORE=1` and `FDNAME` in the notification message. ```python from systemd.daemon import notify import socket # Send with fds (store fds in systemd's fd store) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.bind(('', 8080)) notify('FDSTORE=1\nFDNAME=http', fds=[sock.fileno()]) ``` -------------------------------- ### daemon.listen_fds() Source: https://context7.com/systemd/python-systemd/llms.txt Retrieve socket-activated file descriptors passed to the process by systemd. File descriptors start at 3 (LISTEN_FDS_START). The LISTEN_FDS and LISTEN_PID environment variables are cleared after the first call by default. ```APIDOC ## daemon.listen_fds() ### Description Retrieve socket-activated file descriptors passed to the process by systemd. ### Parameters None ### Returns - list: A list of integer file descriptors. ### Example ```python from systemd.daemon import listen_fds import socket # Basic socket activation fds = listen_fds() if fds: # Wrap the first fd in a socket object server = socket.fromfd(fds[0], socket.AF_INET6, socket.SOCK_STREAM) server.setblocking(False) print(f"Accepted {len(fds)} socket(s) from systemd") else: # Fall back to manual bind for development server = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) server.bind(('::', 8080)) server.listen(128) ``` ``` -------------------------------- ### Send a Test Message to Journald Source: https://github.com/systemd/python-systemd/blob/main/README.md Send a simple test message 'Test' to the systemd journal from a Python interactive session. This requires the package to be installed. ```python from systemd import journal journal.send("Test") ``` -------------------------------- ### Retrieve and Use Socket-Activated File Descriptors with systemd Source: https://context7.com/systemd/python-systemd/llms.txt Use `listen_fds()` to get file descriptors passed by systemd during socket activation. The environment variables `LISTEN_FDS` and `LISTEN_PID` are cleared after the first call by default. `listen_fds_with_names()` retrieves named file descriptors. ```python from systemd.daemon import listen_fds, listen_fds_with_names, booted import socket, os # Basic socket activation fds = listen_fds() if fds: # Wrap the first fd in a socket object server = socket.fromfd(fds[0], socket.AF_INET6, socket.SOCK_STREAM) server.setblocking(False) print(f"Accepted {len(fds)} socket(s) from systemd") else: # Fall back to manual bind for development server = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) server.bind(('::', 8080)) server.listen(128) # Named socket activation (requires FileDescriptorName= in .socket unit) named = listen_fds_with_names() # e.g. {3: 'http', 4: 'https'} for fd, name in named.items(): print(f"fd {fd} → {name}") if name == 'http': http_sock = socket.fromfd(fd, socket.AF_INET6, socket.SOCK_STREAM) elif name == 'https': https_sock = socket.fromfd(fd, socket.AF_INET6, socket.SOCK_STREAM) ``` -------------------------------- ### Read Journal Entries Since a Specific Time Source: https://github.com/systemd/python-systemd/blob/main/README.md Read journal entries from a `journal.Reader` instance starting from a specified real-time. Requires importing `datetime` and `timedelta`. ```python from systemd import journal from datetime import datetime, timedelta j = journal.Reader() j.seek_realtime(datetime.now() - timedelta(minutes=20)) for entry in j: print(entry['MESSAGE']) ``` -------------------------------- ### Custom journal field converters Source: https://context7.com/systemd/python-systemd/llms.txt Initializes a journal reader with custom converters to automatically parse specific fields. This example shows converting a hexadecimal string in 'MY_FIELD' to an integer. ```python from systemd import journal # --- Custom field converters --- j = journal.Reader(converters={'MY_FIELD': lambda b: int(b.decode(), 16)}) ``` -------------------------------- ### Build Documentation Locally Source: https://github.com/systemd/python-systemd/blob/main/README.md Build the project's documentation locally using Ninja. The compiled HTML documentation will be available in the 'build/html' directory. ```bash ninja -C build html ``` -------------------------------- ### daemon.booted() Source: https://context7.com/systemd/python-systemd/llms.txt Checks if the system was booted under systemd by verifying the presence of '/run/systemd/private'. ```APIDOC ## daemon.booted() ### Description Returns `True` if the system was booted under systemd (by checking for `/run/systemd/private`), wrapping `sd_booted(3)`. ### Usage Example **Conditional notification:** ```python from systemd.daemon import booted, notify if booted(): notify('READY=1') else: print("Not running under systemd, skipping sd_notify") ``` ``` -------------------------------- ### Query Active Logins with systemd.login Source: https://context7.com/systemd/python-systemd/llms.txt The `systemd.login` module provides simple functions like `seats()`, `sessions()`, `uids()`, and `machine_names()` to query active system resources. These functions wrap `sd_get_*` calls and return lists of available seats, sessions, user IDs, and machine names. ```python from systemd import login seats = login.seats() # e.g. ['seat0'] sessions = login.sessions() # e.g. ['1', '2', 'c1'] uids = login.uids() # e.g. [1000, 1001] machines = login.machine_names() # e.g. ['fedora-25', 'debian-testing'] print(f"Active seats: {seats}") print(f"Active sessions: {sessions}") print(f"Logged-in UIDs: {uids}") print(f"Running VMs: {machines}") ``` -------------------------------- ### Monitor Machine Events with python-systemd Source: https://context7.com/systemd/python-systemd/llms.txt Use login.Monitor to track machine start/stop events. Requires registering the monitor with a select.poll() object and flushing events. ```python import select from systemd import login # Monitor machine start/stop events m = login.Monitor("machine") p = select.poll() p.register(m, m.get_events()) print("Waiting for VM/container changes...") while True: events = p.poll(m.get_timeout_ms()) if events: m.flush() # reset wakeup state print("Machines now:", login.machine_names()) ``` -------------------------------- ### login.machine_names() Source: https://context7.com/systemd/python-systemd/llms.txt Query the list of names of currently running virtual machines or containers. ```APIDOC ## login.machine_names() ### Description Get a list of running VM or container names. ### Parameters None ### Returns - list: A list of strings, where each string is a machine name. ### Example ```python from systemd import login machines = login.machine_names() print(f"Running VMs: {machines}") ``` ``` -------------------------------- ### Send READY=1 notification Source: https://context7.com/systemd/python-systemd/llms.txt Signals to systemd that the service is ready after initialization. This is a common use case for `daemon.notify()` and requires the service to be running under systemd. ```python from systemd.daemon import notify # Signal readiness after full initialisation notify('READY=1') ``` -------------------------------- ### View Journal Output in Real-time Source: https://github.com/systemd/python-systemd/blob/main/README.md Use journalctl to follow and view systemd journal entries in JSON format. This command requires root privileges. ```bash sudo journalctl -f --output=json ``` -------------------------------- ### Check if systemd is booted Source: https://context7.com/systemd/python-systemd/llms.txt Checks if the system is running under systemd by verifying the existence of '/run/systemd/private'. This is a prerequisite for using `sd_notify` functions. ```python from systemd.daemon import booted if booted(): from systemd.daemon import notify notify('READY=1') else: print("Not running under systemd, skipping sd_notify") ``` -------------------------------- ### Send Simple Message to Journal Source: https://github.com/systemd/python-systemd/blob/main/README.md Use `journal.send` to log a simple string message to the systemd journal. The message is the first argument. ```python from systemd import journal journal.send('Hello world') ``` -------------------------------- ### daemon.listen_fds_with_names() Source: https://context7.com/systemd/python-systemd/llms.txt Retrieve socket-activated file descriptors along with their associated names, as defined by FileDescriptorName in the systemd socket unit. ```APIDOC ## daemon.listen_fds_with_names() ### Description Retrieve socket-activated file descriptors with their names. ### Parameters None ### Returns - dict: A dictionary mapping file descriptor integers to their string names. ### Example ```python from systemd.daemon import listen_fds_with_names import socket # Named socket activation (requires FileDescriptorName= in .socket unit) named = listen_fds_with_names() # e.g. {3: 'http', 4: 'https'} for fd, name in named.items(): print(f"fd {fd} → {name}") if name == 'http': http_sock = socket.fromfd(fd, socket.AF_INET6, socket.SOCK_STREAM) elif name == 'https': https_sock = socket.fromfd(fd, socket.AF_INET6, socket.SOCK_STREAM) ``` ``` -------------------------------- ### login.sessions() Source: https://context7.com/systemd/python-systemd/llms.txt Query the list of currently active login sessions. ```APIDOC ## login.sessions() ### Description Get a list of active login sessions. ### Parameters None ### Returns - list: A list of strings, where each string is a session identifier (e.g., '1', 'c1'). ### Example ```python from systemd import login sessions = login.sessions() print(f"Active sessions: {sessions}") ``` ``` -------------------------------- ### login.Monitor Source: https://context7.com/systemd/python-systemd/llms.txt A file-descriptor-backed monitor for login state changes (session, uid, seat, machine events). It integrates into select/poll/epoll event loops. ```APIDOC ## login.Monitor ### Description Monitor for changes in login sessions, UIDs, seats, or machines. ### Parameters - `categories` (list, optional): A list of categories to monitor (e.g., `"seat"`, `"session"`, `"uid"`, `"machine"`). If `None`, all categories are monitored. ### Returns - login.Monitor: An instance of the Monitor class. ### Example ```python import select from systemd import login # Monitor all login events m = login.Monitor() # Add the monitor's file descriptor to a poller inputs = [m.fileno()] while True: readable, _, _ = select.select(inputs, [], []) for fd in readable: if fd == m.fileno(): for event in m: print(f"Login event: {event}") ``` ``` -------------------------------- ### login.seats() Source: https://context7.com/systemd/python-systemd/llms.txt Query the list of currently active seats. ```APIDOC ## login.seats() ### Description Get a list of active seats. ### Parameters None ### Returns - list: A list of strings, where each string is a seat identifier (e.g., 'seat0'). ### Example ```python from systemd import login seats = login.seats() print(f"Active seats: {seats}") ``` ``` -------------------------------- ### Monitor Login State Changes with systemd.login.Monitor Source: https://context7.com/systemd/python-systemd/llms.txt The `login.Monitor` class provides a file-descriptor-based mechanism to watch for changes in login sessions, UIDs, seats, and machines. It can be integrated into `select`/`poll`/`epoll` event loops. Supported categories include 'seat', 'session', 'uid', 'machine', or `None` for all events. ```python import select from systemd import login ``` -------------------------------- ### daemon.is_socket() / is_socket_inet() / is_socket_unix() / is_socket_sockaddr() / is_fifo() Source: https://context7.com/systemd/python-systemd/llms.txt A family of functions to verify the type and properties of inherited file descriptors. They accept raw integer file descriptors or objects with a `.fileno()` method. ```APIDOC ## daemon.is_socket() / is_socket_inet() / is_socket_unix() / is_socket_sockaddr() / is_fifo() ### Description Verify the type and properties of file descriptors received via socket activation. ### Parameters - `fd`: An integer file descriptor or an object with a `.fileno()` method. - Additional parameters vary per function (e.g., address family, socket type, listening status, port). ### Returns - bool: True if the file descriptor matches the specified criteria, False otherwise. ### Example ```python from systemd.daemon import listen_fds, is_socket, is_socket_inet, is_socket_unix, is_fifo import socket fds = listen_fds(unset_environment=False) for fd in fds: if is_socket_inet(fd, socket.AF_INET6, socket.SOCK_STREAM, listening=1, port=443): print(f"fd {fd}: HTTPS listener") elif is_socket_inet(fd, port=80): print(f"fd {fd}: HTTP listener") elif is_socket_unix(fd, socket.SOCK_STREAM): print(f"fd {fd}: Unix stream socket") elif is_fifo(fd): print(f"fd {fd}: named pipe / FIFO") else: print(f"fd {fd}: unknown type") # Check a socket object directly srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.bind(('127.0.0.1', 9000)) assert is_socket(srv, socket.AF_INET, socket.SOCK_STREAM) ``` ``` -------------------------------- ### id128.get_boot() Source: https://context7.com/systemd/python-systemd/llms.txt Retrieve the 128-bit boot identifier, which changes on every reboot. ```APIDOC ## id128.get_boot() ### Description Get the current boot ID as a UUID object. ### Parameters None ### Returns - uuid.UUID: The boot ID. ### Example ```python from systemd import id128 boot_id = id128.get_boot() print(f"Boot ID: {boot_id}") ``` ``` -------------------------------- ### Send Structured Journal Message with systemd.journal.send() Source: https://context7.com/systemd/python-systemd/llms.txt Use journal.send() to log messages to the systemd journal. Arbitrary fields can be added as keyword arguments. Source location is captured automatically unless overridden. ```python from systemd import journal import uuid # Simple message journal.send('Application started') # Message with custom fields journal.send( 'User login failed', SYSLOG_IDENTIFIER='myapp', PRIORITY=str(journal.LOG_WARNING), USERNAME='alice', REMOTE_IP='192.168.1.42', ) # Message with a stable MESSAGE_ID for catalog lookups mid = uuid.UUID('0123456789abcdef0123456789abcdef') journal.send( 'Database connection lost', MESSAGE_ID=mid, DB_HOST='db.example.com', ERRNO='111', ) # Binary field (sent as-is, not UTF-8 encoded) journal.send('Core dump captured', COREDUMP=b'\xde\xad\xbe\xef') ``` -------------------------------- ### Read Journal Entries with Catalog Explanations Source: https://github.com/systemd/python-systemd/blob/main/README.md Display log messages along with their catalog explanations if available. Uses a try-except block to handle entries without catalog information. ```python from systemd import journal j = journal.Reader() for entry in j: print("MESSAGE: ", entry['MESSAGE']) try: print("CATALOG: ", j.get_catalog()) except: pass ``` -------------------------------- ### Poll for Journal Events with Python Source: https://github.com/systemd/python-systemd/blob/main/docs/journal.md Demonstrates how to poll for journal events using the systemd.journal module and the select.poll() method. This is useful for integrating journal reading into an external event loop. Ensure you have events to read by sending a message or by having other system activity. ```python import select from systemd import journal j = journal.Reader() j.seek_tail() journal.send('testing 1,2,3') # make sure we have something to read j.add_match('MESSAGE=testing 1,2,3') p = select.poll() p.register(j, j.get_events()) p.poll() j.get_next() ``` -------------------------------- ### Validate Inherited File Descriptors with systemd Source: https://context7.com/systemd/python-systemd/llms.txt Use `is_socket()`, `is_socket_inet()`, `is_socket_unix()`, `is_socket_sockaddr()`, `is_fifo()`, and `is_mq()` to verify the type and properties of file descriptors obtained via socket activation. These functions accept raw file descriptor integers or objects with a `.fileno()` method. `listen_fds(unset_environment=False)` can be used to prevent clearing environment variables. ```python from systemd.daemon import ( listen_fds, is_socket, is_socket_inet, is_socket_unix, is_socket_sockaddr, is_fifo ) import socket fds = listen_fds(unset_environment=False) for fd in fds: if is_socket_inet(fd, socket.AF_INET6, socket.SOCK_STREAM, listening=1, port=443): print(f"fd {fd}: HTTPS listener") elif is_socket_inet(fd, port=80): print(f"fd {fd}: HTTP listener") elif is_socket_unix(fd, socket.SOCK_STREAM): print(f"fd {fd}: Unix stream socket") elif is_socket_unix(fd, socket.SOCK_DGRAM): print(f"fd {fd}: Unix datagram socket") elif is_fifo(fd): print(f"fd {fd}: named pipe / FIFO") else: print(f"fd {fd}: unknown type") # Check a socket object directly srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv.bind(('127.0.0.1', 9000)) assert is_socket(srv, socket.AF_INET, socket.SOCK_STREAM) assert is_socket_sockaddr(srv, '127.0.0.1:9000', socket.SOCK_STREAM) ``` -------------------------------- ### journal.sendv() Source: https://context7.com/systemd/python-systemd/llms.txt A lower-level call that sends pre-formatted `FIELD=value` strings directly to `sd_journal_sendv`. Strings are sent as UTF-8; bytes objects are sent as binary data. Raises `ValueError` if called with no arguments or an invalid field. ```APIDOC ## journal.sendv() ### Description Lower-level call that sends pre-formatted `FIELD=value` strings directly to `sd_journal_sendv`. Strings are sent as UTF-8; bytes objects are sent as binary data. Raises `ValueError` if called with no arguments or an invalid field. ### Parameters - **fields** (string or bytes) - Required - One or more `FIELD=value` strings or bytes objects. ### Request Example ```python from systemd import journal # Equivalent to journal.send('Hello world') journal.sendv('MESSAGE=Hello world') # Multiple fields journal.sendv( 'MESSAGE=Packet received', 'PRIORITY=6', 'SYSLOG_IDENTIFIER=netmon', 'SRC_IP=10.0.0.1', ) # Binary field using bytes journal.sendv('MESSAGE=Binary payload', b'PAYLOAD=\xde\xad\xbe\xef') ``` ``` -------------------------------- ### login.uids() Source: https://context7.com/systemd/python-systemd/llms.txt Query the list of user IDs (UIDs) that are currently logged in. ```APIDOC ## login.uids() ### Description Get a list of logged-in user IDs. ### Parameters None ### Returns - list: A list of integers, where each integer is a user ID. ### Example ```python from systemd import login uids = login.uids() print(f"Logged-in UIDs: {uids}") ``` ``` -------------------------------- ### Send Raw Journal Fields with systemd.journal.sendv() Source: https://context7.com/systemd/python-systemd/llms.txt journal.sendv() sends pre-formatted 'FIELD=value' strings directly to sd_journal_sendv. It supports UTF-8 strings and binary data via bytes objects. Raises ValueError if called with no arguments or an invalid field. ```python from systemd import journal # Equivalent to journal.send('Hello world') journal.sendv('MESSAGE=Hello world') # Multiple fields journal.sendv( 'MESSAGE=Packet received', 'PRIORITY=6', 'SYSLOG_IDENTIFIER=netmon', 'SRC_IP=10.0.0.1', ) # Binary field using bytes journal.sendv('MESSAGE=Binary payload', b'PAYLOAD=\xde\xad\xbe\xef') ``` -------------------------------- ### Monitor All Login Event Categories Source: https://context7.com/systemd/python-systemd/llms.txt Monitor all categories of login events using login.Monitor without specifying a category. The file descriptor for polling is printed. ```python from systemd import login # Monitor all categories with login.Monitor() as mon: fd = mon.fileno() print(f"Polling fd {fd} for all login events") ``` -------------------------------- ### Send Message Using sendv Source: https://github.com/systemd/python-systemd/blob/main/README.md Utilize the `journal.sendv` function for sending messages, which accepts arguments in the format 'KEY=VALUE'. This provides a more direct interface to the underlying journald API. ```python from systemd import journal journal.sendv('MESSAGE=Hello world') ``` ```python from systemd import journal journal.sendv('MESSAGE=Hello, again, world', 'FIELD2=Greetings!', 'FIELD3=Guten tag') ``` ```python from systemd import journal journal.sendv('MESSAGE=Binary message', b'BINARY=\xde\xad\xbe\xef') ``` -------------------------------- ### Stream to Journal with systemd.journal.stream() Source: https://context7.com/systemd/python-systemd/llms.txt journal.stream() returns a file-like object for writing to the journal. Each newline-terminated line becomes a journal entry. It can be used as a replacement for sys.stdout or with custom identifiers and priorities. ```python from systemd import journal from syslog import LOG_DEBUG, LOG_ERR import sys # Default: identifier taken from sys.argv[0], priority INFO stream = journal.stream() print('Server initialised', file=stream) # Custom identifier and priority err_stream = journal.stream('myapp-errors', priority=LOG_ERR) err_stream.write('Failed to bind port 8080\n') # Redirect stdout so all print() calls go to journal sys.stdout = journal.stream('myapp', priority=LOG_DEBUG) print('This goes to the journal') ``` -------------------------------- ### Send Binary Message to Journal Source: https://github.com/systemd/python-systemd/blob/main/README.md Log a message containing binary data to the journal. The binary data should be provided as a bytes object. ```python from systemd import journal journal.send('Binary message', BINARY=b'\xde\xad\xbe\xef') ``` -------------------------------- ### daemon.notify() Source: https://context7.com/systemd/python-systemd/llms.txt Sends service status notifications to systemd via the NOTIFY_SOCKET. Used to signal readiness, status updates, or watchdog events. ```APIDOC ## daemon.notify() ### Description Sends a notification message to systemd via the `NOTIFY_SOCKET` environment variable (`sd_notify(3)`). Returns `True` if the message was sent, `False` if there is no notification socket (i.e. not running under systemd). Common messages: `READY=1`, `STOPPING=1`, `RELOADING=1`, `STATUS=`, `WATCHDOG=1`. ### Usage Examples **Signal readiness after full initialization:** ```python from systemd.daemon import notify notify('READY=1') ``` **Update the service status text:** ```python from systemd.daemon import notify notify('STATUS=Processing 42 queued jobs') ``` **Watchdog keep-alive (requires WatchdogSec= in unit file):** ```python from systemd.daemon import notify import time while True: notify('WATCHDOG=1') time.sleep(10) ``` **Signal clean shutdown:** ```python from systemd.daemon import notify notify('STOPPING=1') ``` **Send with file descriptors (FDs):** ```python from systemd.daemon import notify import socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.bind(('', 8080)) notify('FDSTORE=1\nFDNAME=http', fds=[sock.fileno()]) ``` ``` -------------------------------- ### Python Logging with systemd.journal.JournalHandler Source: https://context7.com/systemd/python-systemd/llms.txt JournalHandler integrates with Python's logging framework to send log records to the systemd journal. It maps Python log levels to syslog priorities and supports per-handler fields and per-message MESSAGE_ID. ```python import logging import uuid from systemd import journal # Dedicated logger, messages go only to journal log = logging.getLogger('myapp') log.propagate = False log.setLevel(logging.DEBUG) log.addHandler(journal.JournalHandler(SYSLOG_IDENTIFIER='myapp')) log.info('Service ready on port %d', 8080) log.warning('Disk usage above 90%%') try: 1 / 0 except ZeroDivisionError: log.exception('Unexpected error in request handler') # Attach a stable MESSAGE_ID to a specific message mid = uuid.UUID('deadbeefdeadbeefdeadbeefdeadbeef') log.error('License validation failed', extra={'MESSAGE_ID': mid}) # Configure via logging.config.dictConfig import logging.config logging.config.dictConfig({ 'version': 1, 'handlers': { 'journal': { 'class': 'systemd.journal.JournalHandler', 'SYSLOG_IDENTIFIER': 'myapp', } }, 'root': {'handlers': ['journal'], 'level': 'DEBUG'}, }) ``` -------------------------------- ### journal.stream() Source: https://context7.com/systemd/python-systemd/llms.txt Returns a line-buffered file object backed by a journal stream socket. Every newline-terminated line written to it becomes a journal entry. Suitable for use with `print()` or as a drop-in replacement for `sys.stdout`. ```APIDOC ## journal.stream() ### Description Returns a line-buffered file object backed by a journal stream socket (`sd_journal_stream_fd`). Every `\n`-terminated line written to it becomes a journal entry. Suitable for use with `print()` or as a drop-in replacement for `sys.stdout`. ### Parameters - **identifier** (string) - Optional - The `SYSLOG_IDENTIFIER` for the journal entries. - **priority** (int) - Optional - The syslog priority for the journal entries. Defaults to `syslog.LOG_INFO`. ### Request Example ```python from systemd import journal from syslog import LOG_DEBUG, LOG_ERR import sys # Default: identifier taken from sys.argv[0], priority INFO stream = journal.stream() print('Server initialised', file=stream) # Custom identifier and priority err_stream = journal.stream('myapp-errors', priority=LOG_ERR) err_stream.write('Failed to bind port 8080\n') # Redirect stdout so all print() calls go to journal sys.stdout = journal.stream('myapp', priority=LOG_DEBUG) print('This goes to the journal') ``` ``` -------------------------------- ### journal.send() Source: https://context7.com/systemd/python-systemd/llms.txt Sends a structured log message to the systemd journal. The MESSAGE argument is required; arbitrary additional fields can be passed as keyword arguments. Source location is captured automatically unless overridden. ```APIDOC ## journal.send() ### Description Sends a structured log message to the systemd journal. The `MESSAGE` argument is required; arbitrary additional fields (always uppercase) can be passed as keyword arguments with string or bytes values. Source location (`CODE_FILE`, `CODE_LINE`, `CODE_FUNC`) is captured automatically from the call stack unless explicitly overridden. ### Parameters - **MESSAGE** (string) - Required - The main log message. - **keyword arguments** (string or bytes) - Optional - Arbitrary additional fields (uppercase). ### Request Example ```python from systemd import journal import uuid # Simple message journal.send('Application started') # Message with custom fields journal.send( 'User login failed', SYSLOG_IDENTIFIER='myapp', PRIORITY=str(journal.LOG_WARNING), USERNAME='alice', REMOTE_IP='192.168.1.42', ) # Message with a stable MESSAGE_ID for catalog lookups mid = uuid.UUID('0123456789abcdef0123456789abcdef') journal.send( 'Database connection lost', MESSAGE_ID=mid, DB_HOST='db.example.com', ERRNO='111', ) # Binary field (sent as-is, not UTF-8 encoded) journal.send('Core dump captured', COREDUMP=b'\xde\xad\xbe\xef') ``` ``` -------------------------------- ### Read Journal Entries by Executable Path Source: https://github.com/systemd/python-systemd/blob/main/README.md Filter journal entries to show only those generated by a specific executable. The match is added using `_EXE` field. ```python from systemd import journal j = journal.Reader() j.add_match('_EXE=/usr/bin/vim') for entry in j: print(entry['MESSAGE']) ``` -------------------------------- ### journal.get_catalog() Source: https://context7.com/systemd/python-systemd/llms.txt Retrieves the human-readable catalog description for a given MESSAGE_ID UUID, similar to `journalctl --explain`. ```APIDOC ## journal.get_catalog() ### Description Retrieves the human-readable catalog description for a given `MESSAGE_ID` UUID, equivalent to `journalctl --explain`. Returns a string or raises `OSError` if the ID is not found. ### Usage Examples **Look up a known systemd message ID:** ```python from systemd import journal, id128 try: text = journal.get_catalog(id128.SD_MESSAGE_SESSION_START) print(text) except OSError as e: print(f"No catalog entry: {e}") ``` **Look up by hex string:** ```python from systemd import journal try: text = journal.get_catalog('fc2e22bc6ee647b6b90729ab34a250b1') print(text) except OSError: pass ``` ``` -------------------------------- ### id128.get_machine_app_specific() Source: https://context7.com/systemd/python-systemd/llms.txt Derive a deterministically generated, application-specific 128-bit identifier based on the machine ID. ```APIDOC ## id128.get_machine_app_specific() ### Description Get an application-specific machine ID derived from the stable machine ID. ### Parameters - `machine_id` (uuid.UUID): The stable machine ID. ### Returns - uuid.UUID: The application-specific machine ID. ### Example ```python from systemd import id128 machine_id = id128.get_machine() app_uuid = id128.get_machine_app_specific(machine_id) print(f"App+machine ID: {app_uuid}") ``` ``` -------------------------------- ### Configure Python Logging to Journald Source: https://github.com/systemd/python-systemd/blob/main/README.md Configure the Python logging module to send messages to the systemd journal. Ensure 'custom_logger_name' and 'custom_unit_name' are set appropriately for your application. ```python import logging from systemd import journal logger = logging.getLogger('custom_logger_name') logger.addHandler(journal.JournalHandler(SYSLOG_IDENTIFIER='custom_unit_name')) logger.warning("Some message: %s", 'detail') ``` -------------------------------- ### id128.get_machine() Source: https://context7.com/systemd/python-systemd/llms.txt Retrieve the stable 128-bit machine identifier, read from /etc/machine-id. ```APIDOC ## id128.get_machine() ### Description Get the stable machine ID as a UUID object. ### Parameters None ### Returns - uuid.UUID: The machine ID. ### Example ```python from systemd import id128 machine_id = id128.get_machine() print(f"Machine ID: {machine_id}") ``` ``` -------------------------------- ### Look up message catalog entry by UUID Source: https://context7.com/systemd/python-systemd/llms.txt Retrieves the human-readable description for a systemd message ID using `journal.get_catalog()`. Handles `OSError` if the ID is not found. Requires importing `id128` for known IDs. ```python from systemd import journal, id128 # Look up a known systemd message ID try: text = journal.get_catalog(id128.SD_MESSAGE_SESSION_START) print(text) except OSError as e: print(f"No catalog entry: {e}") ``` -------------------------------- ### Generate and Use 128-bit Unique Identifiers with systemd.id128 Source: https://context7.com/systemd/python-systemd/llms.txt The `systemd.id128` module provides wrappers for `sd_id128_*` functions to generate stable or random 128-bit identifiers as `uuid.UUID` objects. It includes functions for machine ID, boot ID, application-specific machine IDs, and random UUIDs. It also exports `SD_MESSAGE_*` constants for filtering journal entries. ```python from systemd import id128 # Stable machine ID (same across reboots, read from /etc/machine-id) machine_id = id128.get_machine() print(f"Machine ID: {machine_id}") # e.g. UUID('263bb31e-...') # Current boot ID (changes on every reboot) boot_id = id128.get_boot()) print(f"Boot ID: {boot_id}") # Application-specific machine ID (deterministically derived, stable per app+machine) app_uuid = id128.get_machine_app_specific(machine_id) print(f"App+machine ID: {app_uuid}") # Random UUID (wraps sd_id128_randomize, uses kernel entropy) r1 = id128.randomize()) r2 = id128.randomize()) assert r1 != r2 # Use SD_MESSAGE_* constants to filter journal reads from systemd import journal with journal.Reader() as j: j.messageid_match(id128.SD_MESSAGE_SESSION_START) for entry in j: print("Session started:", entry.get('MESSAGE')) ``` -------------------------------- ### Filter journal entries by executable and priority Source: https://context7.com/systemd/python-systemd/llms.txt Filters journal entries to show messages from a specific executable ('/usr/bin/python3') and a priority level of 3. Uses 'AND' logic for multiple matches. ```python from systemd import journal # --- Filter by executable --- with journal.Reader() as j: j.add_match('_EXE=/usr/bin/python3') j.add_match('PRIORITY=3') # AND with different field for entry in j: print(entry.get('MESSAGE')) ``` -------------------------------- ### Query unique journal field values Source: https://context7.com/systemd/python-systemd/llms.txt Queries the journal for unique values of specified fields, such as '_SYSTEMD_UNIT' and '_PID'. This is useful for understanding the distinct entities that have logged messages. ```python from systemd import journal with journal.Reader() as j: units = j.query_unique('_SYSTEMD_UNIT') print("Units that logged today:") for u in sorted(units): print(' ', u) pids = j.query_unique('_PID') print(f"Distinct PIDs seen: {len(pids)}") ``` -------------------------------- ### Read Kernel Ring Buffer Entries Source: https://github.com/systemd/python-systemd/blob/main/README.md Access kernel ring buffer messages from the journal by adding a match for `_TRANSPORT=kernel`. ```python from systemd import journal j = journal.Reader() j.add_match('_TRANSPORT=kernel') for entry in j: print(entry['MESSAGE']) ``` -------------------------------- ### Wait for new journal entries with select.poll Source: https://context7.com/systemd/python-systemd/llms.txt Integrates with a non-blocking event loop to wait for and process new journal entries using `select.poll`. It seeks to the tail of the journal and then polls for events with a 5-second timeout. ```python from systemd import journal import select # --- Wait for new entries (non-blocking event loop integration) --- j = journal.Reader() j.seek_tail() j.get_next() # move past existing tail p = select.poll() p.register(j, j.get_events()) while True: events = p.poll(5000) # 5-second timeout if events: j.process() # update internal state for entry in j: print('NEW:', entry.get('MESSAGE')) ``` -------------------------------- ### Send STOPPING=1 notification Source: https://context7.com/systemd/python-systemd/llms.txt Signals a clean shutdown to systemd. This is typically sent just before the service terminates. ```python from systemd.daemon import notify # Signal clean shutdown notify('STOPPING=1') ``` -------------------------------- ### journal.Reader Source: https://context7.com/systemd/python-systemd/llms.txt Provides an iterator-based interface to read and filter systemd journal entries. Entries are returned as Python dictionaries and can be filtered by various criteria. ```APIDOC ## journal.Reader ### Description An iterator-based interface to the systemd journal that mirrors `journalctl`. Entries are returned as plain Python dictionaries with fields automatically converted to native types (integers, `datetime.datetime`, `uuid.UUID`, etc.) via a configurable `converters` map. Supports filtering by field matches, log level, boot ID, machine ID, and time range. ### Usage Examples **Read recent entries:** ```python from systemd import journal from datetime import datetime, timedelta j = journal.Reader() j.seek_realtime(datetime.now() - timedelta(minutes=30)) for entry in j: ts = entry.get('__REALTIME_TIMESTAMP', '') msg = entry.get('MESSAGE', '') unit = entry.get('_SYSTEMD_UNIT', 'unknown') print(f"[{ts}] {unit}: {msg}") ``` **Filter by unit and log level (INFO and above):** ```python from systemd import journal with journal.Reader() as j: j.this_boot() j.log_level(journal.LOG_INFO) j.add_match(_SYSTEMD_UNIT='sshd.service') for entry in j: print(entry['MESSAGE']) ``` **Filter by executable:** ```python from systemd import journal with journal.Reader() as j: j.add_match('_EXE=/usr/bin/python3') j.add_match('PRIORITY=3') # AND with different field for entry in j: print(entry.get('MESSAGE')) ``` **Kernel messages only (journalctl -k):** ```python from systemd import journal with journal.Reader() as j: j.add_match('_TRANSPORT=kernel') for entry in j: print(entry['MESSAGE']) ``` **Wait for new entries (non-blocking event loop integration):** ```python from systemd import journal import select j = journal.Reader() j.seek_tail() j.get_next() # move past existing tail p = select.poll() p.register(j, j.get_events()) while True: events = p.poll(5000) # 5-second timeout if events: j.process() # update internal state for entry in j: print('NEW:', entry.get('MESSAGE')) ``` **Reverse iteration (newest first):** ```python from systemd import journal class ReverseReader(journal.Reader): def __next__(self): ans = self.get_previous() if ans: return ans raise StopIteration() j = ReverseReader() j.add_match(_SYSTEMD_UNIT='nginx.service') j.seek_tail() for entry in j: print(entry['MESSAGE']) ``` **Custom field converters:** ```python from systemd import journal j = journal.Reader(converters={'MY_FIELD': lambda b: int(b.decode(), 16)}) ``` ``` -------------------------------- ### Filter journal entries by unit and log level Source: https://context7.com/systemd/python-systemd/llms.txt Filters journal entries to show only messages from 'sshd.service' with a log level of INFO or higher. Requires importing the journal module. ```python from systemd import journal # --- Filter by unit and log level (INFO and above) --- with journal.Reader() as j: j.this_boot() j.log_level(journal.LOG_INFO) j.add_match(_SYSTEMD_UNIT='sshd.service') for entry in j: print(entry['MESSAGE']) ``` -------------------------------- ### Watchdog keep-alive notification Source: https://context7.com/systemd/python-systemd/llms.txt Sends periodic watchdog keep-alive signals to systemd. This requires `WatchdogSec=` to be configured in the service's unit file to prevent systemd from restarting the service due to perceived inactivity. ```python from systemd.daemon import notify import time # Watchdog keep-alive (requires WatchdogSec= in unit file) while True: notify('WATCHDOG=1') time.sleep(10) ``` -------------------------------- ### journal.JournalHandler Source: https://context7.com/systemd/python-systemd/llms.txt A `logging.Handler` subclass that routes Python log records into the systemd journal, automatically mapping Python log levels to syslog priorities. Per-handler fields are attached to every emitted record. ```APIDOC ## journal.JournalHandler ### Description A `logging.Handler` subclass that routes Python log records into the systemd journal, automatically mapping Python log levels to syslog priorities. Per-handler fields (e.g. `SYSLOG_IDENTIFIER`) are attached to every emitted record; per-message `MESSAGE_ID` can be set via the `extra` dict. ### Parameters - **SYSLOG_IDENTIFIER** (string) - Optional - Identifier for the journal entries. - **priority** (int) - Optional - Default priority for log records. ### Request Example ```python import logging import uuid from systemd import journal # Dedicated logger, messages go only to journal log = logging.getLogger('myapp') log.propagate = False log.setLevel(logging.DEBUG) log.addHandler(journal.JournalHandler(SYSLOG_IDENTIFIER='myapp')) log.info('Service ready on port %d', 8080) log.warning('Disk usage above 90%%') try: 1 / 0 except ZeroDivisionError: log.exception('Unexpected error in request handler') # Attach a stable MESSAGE_ID to a specific message mid = uuid.UUID('deadbeefdeadbeefdeadbeefdeadbeef') log.error('License validation failed', extra={'MESSAGE_ID': mid}) # Configure via logging.config.dictConfig import logging.config logging.config.dictConfig({ 'version': 1, 'handlers': { 'journal': { 'class': 'systemd.journal.JournalHandler', 'SYSLOG_IDENTIFIER': 'myapp', } }, 'root': {'handlers': ['journal'], 'level': 'DEBUG'}, }) ``` ``` -------------------------------- ### Update service status text Source: https://context7.com/systemd/python-systemd/llms.txt Updates the status text visible in `systemctl status`. This allows dynamic reporting of the service's current state or activity. ```python from systemd.daemon import notify # Update the service status text visible in systemctl status notify('STATUS=Processing 42 queued jobs') ``` -------------------------------- ### Look up message catalog entry by hex string Source: https://context7.com/systemd/python-systemd/llms.txt Retrieves a message catalog entry using a hexadecimal string representation of the message ID. Catches `OSError` if the ID is not found in the catalog. ```python from systemd import journal # Look up by hex string try: text = journal.get_catalog('fc2e22bc6ee647b6b90729ab34a250b1') print(text) except OSError: pass ``` -------------------------------- ### id128.randomize() Source: https://context7.com/systemd/python-systemd/llms.txt Generate a random 128-bit identifier using kernel entropy. ```APIDOC ## id128.randomize() ### Description Generate a random UUID. ### Parameters None ### Returns - uuid.UUID: A random UUID. ### Example ```python from systemd import id128 r1 = id128.randomize() r2 = id128.randomize() assert r1 != r2 ``` ``` -------------------------------- ### Filter for kernel messages Source: https://context7.com/systemd/python-systemd/llms.txt Retrieves only kernel messages from the journal, equivalent to 'journalctl -k'. Uses a transport match for '_transport=kernel'. ```python from systemd import journal # --- Kernel messages only (journalctl -k) --- with journal.Reader() as j: j.add_match('_TRANSPORT=kernel') for entry in j: print(entry['MESSAGE']) ``` -------------------------------- ### Read recent journal entries Source: https://context7.com/systemd/python-systemd/llms.txt Reads and prints recent journal entries from the last 30 minutes. Entries are displayed with their timestamp, unit, and message. ```python from systemd import journal from datetime import datetime, timedelta import select # --- Read recent entries --- j = journal.Reader() j.seek_realtime(datetime.now() - timedelta(minutes=30)) for entry in j: ts = entry.get('__REALTIME_TIMESTAMP', '') msg = entry.get('MESSAGE', '') unit = entry.get('_SYSTEMD_UNIT', 'unknown') print(f"[{ts}] {unit}: {msg}") ``` -------------------------------- ### Reverse iterate journal entries Source: https://context7.com/systemd/python-systemd/llms.txt Iterates through journal entries in reverse order (newest first) using a custom `ReverseReader` class. This is useful for viewing recent logs. Requires `seek_tail()` to position the reader. ```python from systemd import journal # --- Reverse iteration (newest first) --- class ReverseReader(journal.Reader): def __next__(self): ans = self.get_previous() if ans: return ans raise StopIteration() j = ReverseReader() j.add_match(_SYSTEMD_UNIT='nginx.service') j.seek_tail() for entry in j: print(entry['MESSAGE']) ``` -------------------------------- ### Read Journal Entries in Reverse Order Source: https://github.com/systemd/python-systemd/blob/main/README.md Implement a custom reader class `ReverseReader` inheriting from `journal.Reader` to iterate through journal entries in reverse chronological order using `get_previous()`. ```python from systemd import journal class ReverseReader(journal.Reader): def __next__(self): ans = self.get_previous() if ans: return ans raise StopIteration() j = ReverseReader() j.add_match('_EXE=/usr/bin/vim') j.seek_tail() for entry in j: print(entry['MESSAGE']) ``` -------------------------------- ### Send Structured Message to Journal Source: https://github.com/systemd/python-systemd/blob/main/README.md Send a message to the journal with additional structured fields. Field names must be uppercase, following the journald API. ```python from systemd import journal journal.send('Hello, again, world', FIELD2='Greetings!', FIELD3='Guten tag') ``` -------------------------------- ### Read Journal Entries Between Two Timestamps Source: https://github.com/systemd/python-systemd/blob/main/README.md Filter journal entries to display only those within a specific time range. The loop breaks if the entry's timestamp exceeds the 'until' time. ```python from systemd import journal from datetime import datetime, timedelta j = journal.Reader() since = datetime.now() - timedelta(minutes=50) until = datetime.now() - timedelta(minutes=10) j.seek_realtime(since) for entry in j: if entry['__REALTIME_TIMESTAMP'] > until: break print(entry['MESSAGE']) ``` -------------------------------- ### journal.Reader.query_unique() Source: https://context7.com/systemd/python-systemd/llms.txt Lists unique values for a specified field across all journal entries, irrespective of current filter matches. ```APIDOC ## journal.Reader.query_unique() ### Description Returns the set of all distinct values recorded for a given field name across the entire journal (ignoring current matches). ### Usage Examples **List unique units that logged today:** ```python from systemd import journal with journal.Reader() as j: units = j.query_unique('_SYSTEMD_UNIT') print("Units that logged today:") for u in sorted(units): print(' ', u) ``` **Count distinct PIDs seen:** ```python from systemd import journal with journal.Reader() as j: pids = j.query_unique('_PID') print(f"Distinct PIDs seen: {len(pids)}") ``` ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.