Process utilities for working with system commands.¶

Module Contents¶

Classes¶
`Command`	System command class.
`Daemon`	A generic Unix daemon abstract base class.

Functions¶
`get_real_users`()	Return a set of all real user accounts on the…

get_real_users() → set[str][source]¶

Return a set of all real user accounts on the system.

Retrieves a set of all user accounts that have a valid login shell. On Linux, /etc/login.defs is parsed for the UID_MIN value. On other POSIX systems (macOS, BSD), a default minimum UID of 500 is used, which correctly includes real users starting at 501.

System accounts with a UID below the minimum are excluded, as are accounts with nologin-style shells.

Note

Availability: Unix, not WASI, not iOS

Returns:¶: set [str ] – A set of usernames for all real user accounts on the system.

class Command(name: str, *, path: os.PathLike[str] | None = None, user: str | None = None)[source]¶

System command class.

Represents a system command and provides methods for running the command and handling the output.

It also supports specifying the user to run the command as on POSIX systems. The actual implementation use the sudo command to execute the command if user is specified.

Commands are never executed through a shell. On POSIX systems, if a user is specified, the command is executed via sudo -u <user>.

This class supports both synchronous execution via __call__() and asynchronous execution via async_call() using asyncio.

Parameters:¶

name: str¶: The name of the command.
path: os.PathLike[str] | None = None¶: The path to the command executable. Default: None.
user: str | None = None¶: The user to run the command as. Requires sudo to be available on the system. Default: None.

Raises:¶

Command.Error – If the command is not found on the system.
NotImplementedError – if user is specified on non POSIX systems.

Examples:

>>> cmd = Command("ls")
>>> cmd("-la", "/tmp")

User Switching¶

The user parameter is only available on POSIX systems (Linux, macOS, BSD). It is not supported on Windows. This section explains why and provides guidance for cross-platform development.

Why Windows is Different

On POSIX systems, sudo -u <user> allows running a command as a different user without knowing their password, provided the invoking user has sudo privileges. This model assumes:

A centralized authentication system (PAM, /etc/passwd)
Passwordless sudo configuration (/etc/sudoers)
A flat UID-based user model (integer UIDs)

Windows uses a fundamentally different security model:

UAC (User Account Control): Elevation is per-process, not per-command. The runas verb triggers a UAC prompt, but it elevates the current user to Administrator — it does not switch to a different user account.
No passwordless elevation: Unlike sudo, Windows runas requires the user’s password interactively. There is no equivalent of /etc/sudoers for passwordless cross-user execution.
Separate security contexts: An elevated process runs in a different security token. Standard subprocess cannot capture stdout/stderr from an elevated process because the pipes cannot cross the security boundary.
SID-based identity: Windows identifies users by SID (Security Identifier), not integer UIDs. User enumeration requires Windows API calls (NetUserEnum) rather than parsing /etc/passwd.

Cross-Platform Guidance

If you need to run commands with elevated privileges across platforms, use platform-specific branches:

import sys
from deluxe.process import Command

if sys.platform == "win32":
    # Windows: use PowerShell for elevation
    pws = Command("powershell")
    pws("-Command", "Start-Process regedit -Verb RunAs -Wait")
else:
    # POSIX: use sudo
    apt = Command("apt")
    apt.user = "root"
    apt("update")

Summary of Platform Differences

Feature	POSIX	Windows
Run as different user	`sudo -u <user>`	Not possible without password (use `runas`)
Elevate same user	`sudo <cmd>`	`runas` verb (UAC prompt)
Capture stdout	✅ Direct	❌ Separate security context
Pass stdin	✅ Direct	❌ Not possible
Wait for exit code	✅ Direct	⚠️ Requires pywin32

property user : str | None¶

The user associated with this command.

Returns:¶: str | None – The username, or None if not set.

command : str = "''"¶

name : str¶

async async_call( *args: str, input: bytes | None = None, capture: bool = True, cwd: deluxe.types.AnyFilePath | None = None, env: collections.abc.Mapping[str, str] | None = None, ) → asyncio.Task[asyncio.Future[bytes]][source]¶

Run this command asynchronously.

Executes the command as an asyncio.subprocess.Process and returns a asyncio.Task wrapping a asyncio.Future that resolves to the command’s stdout bytes.

Parameters:¶

*args: str¶: The arguments to pass to the command.
input: bytes | None = None¶: The input to pass to the command’s stdin. Default: None.
capture: bool = True¶: Whether to capture the command output. Default: True.
cwd: deluxe.types.AnyFilePath | None = None¶: The current working directory for the command. Default: None.
env: collections.abc.Mapping[str, str] | None = None¶: The environment variables for the command. Default: None.

Returns:¶

asyncio.Task [asyncio.Future [bytes ] ] – A task that resolves to the command’s stdout bytes.

Raises:¶

Command.Error – A dynamically created subclass if the process
exits due to a signal (negative return code). –

class Daemon[source]¶

Extends: abc.ABC

A generic Unix daemon abstract base class.

Make instance of this class execute in a daemonized process using an Unix double fork mechanism.

Note

Availability: Unix, not WASI

Subclasses must implement the run() method, which contains the daemon’s working logic. User-defined Daemon instances should not call run() directly. The daemon writes a pidfile at its start to prevent multiple instances from running simultaneously. Once daemonized, the run() method is called with no parameters. The daemon executes in its own detached session with no tty attached, so it will not inherit the standard files from the python interpreter where it was instancied. Code instantiating the daemon will receive a functional instance of the defined class. This instance acts as a daemon controller.

Keyword Arguments:¶: workpath (Path | str) – The working directory for the daemon process. Must be an existing directory. Default: "/".

Interprocess Communication¶

Daemon subclasses will end up with two instances: the controller living in the calling process, and the worker living in its own detached session. The controller provides a built-in set of methods to manage the daemon’s life cycle: start(), stop(), restart(), signal_user1(), and signal_user2() (see next section for details).

For richer communication channels (bidirectional pipes, shared memory, sockets, etc.), this class makes no provision for a specific protocol; it is up to the subclass implementation. Common Python options include:

multiprocessing.Queue — thread-safe message passing between the controller and the daemon worker.
multiprocessing.connection.Connection (via multiprocessing.Pipe()) — lightweight bidirectional byte stream.
multiprocessing.shared_memory — zero-copy shared data between processes (Python 3.8+).
socket — Unix domain sockets for structured protocols or TCP sockets for network-accessible daemons.
asyncio event loop with asyncio.Queue — for daemons built on asynchronous I/O.

Controller Semantics¶

Instantiating a Daemon subclass returns a controller — a lightweight handle that manages the daemon’s lifecycle. The controller is not the daemon itself; it is a proxy that communicates with the daemon through the pidfile and signals.

Multiple controllers are allowed. Several controller instances can coexist for the same daemon class within a single process or across processes. All controllers for a given class point to the same daemon process. Calling stop(), start(), restart(), signal_user1(), or signal_user2() on any controller affects the shared daemon.

The daemon is a system-level singleton. Only one daemon process runs at a time for a given class, enforced by the pidfile. When a new controller is created while the daemon is already running, the singleton guard prevents a duplicate daemon from starting. The new controller simply becomes another handle to the existing daemon.

Constructor arguments are preserved per controller. Each controller retains the *args and **kwds passed to its constructor. When start() is called to (re)launch the daemon, it uses the calling controller’s original arguments. This means different controllers may hold different argument sets; the last controller to call start() determines the daemon’s configuration.

Warning

Constructor arguments are stored in memory for the lifetime of the controller and are never written to disk. If the controller is garbage collected, its arguments are lost. If a daemon dies and needs to be restarted, the controller calling start() must have been created with the intended arguments.

Concurrency safety. A process-level lock prevents two controllers in the same process from racing through start() concurrently. Cross-process races are handled by the pidfile singleton guard.

User Signals¶

The controller’s signal_user1() and signal_user2() methods send SIGUSR1 and SIGUSR2 to the daemon process. When the daemon receives one of these signals, it calls the overridable hook method on_user1() or on_user2(). The default implementations are no-ops; override them in your subclass to define custom behavior.

import time
from deluxe.process import Daemon

class Worker(Daemon):
    def run(self):
        while True:
            time.sleep(1)

    def on_user1(self):
        # Called when controller sends SIGUSR1
        self.reloading = True

daemon = Worker()
daemon.start()
daemon.signal_user1()  # sends SIGUSR1 -> on_user1()
daemon.stop()

Note

The on_user1 / on_user2 hooks run inside a signal handler. Keep them lightweight: set a flag or event, and defer heavy work to the main run() loop.

About The Unix Double Fork Mechanism¶

In Unix every process belongs to a group which in turn belongs to a session (session (SID) -> process Group (PGID) -> process (PID)). The first process in the session becomes the session leader. Every session can have one TTY associated with it and only a session leader can take control of a TTY.

Normally, when launching a daemon, setsid is called (from the child process after calling fork) to dissociate the daemon from its controlling terminal. However, calling setsid also means that the calling process will be the session leader of the new session, which leaves open the possibility that the daemon could reacquire a controlling terminal in the future.

The double-fork technique ensures that the daemon process is no longer a session leader, making the init process responsible for its cleanup. Forking a second child and exiting immediately prevents zombies and causes the second child process to be orphaned, preventing it from ever acquiring inadvertently a controlling terminal.

property pid : int¶

The PID of this daemon.

Returns:¶: int – The PID of the daemon if running, 0 otherwise.

stop() → None[source]¶

Stop the daemon.

Sends SIGTERM to the daemon process and waits for it to terminate. If the daemon does not terminate within a timeout, SIGKILL is sent as a last resort. If the daemon is not running, a warning is issued and the pidfile is cleaned up if present.

Raises:¶: OSError – If the daemon process could not be killed for a reason other than it not existing.

start() → int[source]¶

Start the daemon.

If the daemon is not already running, it is daemonized. If it is already running, a warning is issued.

Returns:¶: int – The PID of the already-running daemon, or 0 if a new daemon was started.

restart() → None[source]¶

Restart the daemon.

Stops the daemon if running, then starts it again.

signal_user1() → None[source]¶

Send SIGUSR1 to the daemon process.

The daemon’s on_user1() method is invoked when this signal is received. Override on_user1() in your subclass to define the response.

If the daemon is not running, a warning is issued.

signal_user2() → None[source]¶

Send SIGUSR2 to the daemon process.

The daemon’s on_user2() method is invoked when this signal is received. Override on_user2() in your subclass to define the response.

If the daemon is not running, a warning is issued.

atexit() → None[source]¶

Called when the daemon terminates.

Override this method to include cleanup code. This method is registered via atexit.register() and will be executed upon normal interpreter termination.