trading_agent.extensions.systemd_user_units

Reads and changes the state of systemd user units, for the apps that packages bring.

Apps such as sridhara already run as systemd user units, which start at login and restart after a crash. trading-agent reports whether each app's unit is running and can start, stop or restart it, but only units named by an installed package's app.toml, and only with systemctl --user, never as root.

Typical usage example:

units = SystemdUserUnits() state = await units.state('sridhara.service') await units.act('sridhara.service', 'restart')

ACTIONS module-attribute

ACTIONS = ('start', 'stop', 'restart')

UnitError

Bases: RuntimeError

systemctl failed, or was asked about a unit name it should not be given.

UnitState dataclass

UnitState(load_state, active_state, sub_state)

What systemd says about one unit.

Attributes:

Name	Type	Description
load_state	str	"loaded" when systemd knows the unit, "not-found" when it does not.
active_state	str	Such as "active", "inactive" or "failed".
sub_state	str	Such as "running" or "dead".

SystemdUserUnits

Talks to the user's systemd through systemctl --user.

state async

state(unit)

Asks systemd for a unit's state.

Parameters:

Name	Type	Description	Default
unit	str	The unit, such as sridhara.service.	required

Returns:

Type	Description
UnitState	The unit's state.

Raises:

Type	Description
UnitError	The name is not a service unit, or systemctl failed.

Source code in src/trading_agent/extensions/systemd_user_units.py

async def state(self, unit: str) -> UnitState:
    """Asks systemd for a unit's state.

    Args:
        unit (str): The unit, such as sridhara.service.

    Returns:
        UnitState: The unit's state.

    Raises:
        UnitError: The name is not a service unit, or systemctl failed.
    """
    self._check(unit)
    output = await self._systemctl(
        'show',
        unit,
        '--property=LoadState,ActiveState,SubState',
    )
    values = {}
    for line in output.splitlines():
        name, _, value = line.partition('=')
        values[name] = value
    return UnitState(
        load_state=values.get('LoadState', 'unknown'),
        active_state=values.get('ActiveState', 'unknown'),
        sub_state=values.get('SubState', 'unknown'),
    )

act async

act(unit, action)

Starts, stops or restarts a unit.

Parameters:

Name	Type	Description	Default
unit	str	The unit.	required
action	str	"start", "stop" or "restart".	required

Raises:

Type	Description
UnitError	The action is unknown, the name is not a service unit, or systemctl failed.

Source code in src/trading_agent/extensions/systemd_user_units.py

async def act(self, unit: str, action: str) -> None:
    """Starts, stops or restarts a unit.

    Args:
        unit (str): The unit.
        action (str): "start", "stop" or "restart".

    Raises:
        UnitError: The action is unknown, the name is not a service unit, or systemctl failed.
    """
    if action not in ACTIONS:
        raise UnitError(
            f'Unknown action {action!r}; use start, stop or restart'
        )
    self._check(unit)
    await self._systemctl(action, unit)

enable async

enable(unit, unit_file)

Links a unit file if one is given, then enables and starts the unit.

Parameters:

Name	Type	Description	Default
unit	str	The unit.	required
unit_file	Path | None	The unit file to link first, or None when systemd already knows the unit.	required

Raises:

Type	Description
UnitError	The name is not a service unit, the file's name does not match it, or systemctl failed.

Source code in src/trading_agent/extensions/systemd_user_units.py

async def enable(self, unit: str, unit_file: Path | None) -> None:
    """Links a unit file if one is given, then enables and starts the unit.

    Args:
        unit (str): The unit.
        unit_file (Path | None): The unit file to link first, or None when systemd already knows the unit.

    Raises:
        UnitError: The name is not a service unit, the file's name does not match it, or systemctl failed.
    """
    self._check(unit)
    if unit_file is not None:
        if unit_file.name != unit:
            raise UnitError(
                f'The unit file {unit_file} must be named {unit}'
            )
        if not unit_file.is_file():
            raise UnitError(f'The unit file {unit_file} does not exist')
        await self._systemctl('link', str(unit_file.resolve()))
    await self._systemctl('enable', '--now', unit)

disable async

disable(unit)

Stops a unit and stops it starting at login.

Parameters:

Name	Type	Description	Default
unit	str	The unit.	required

Raises:

Type	Description
UnitError	The name is not a service unit, or systemctl failed.

Source code in src/trading_agent/extensions/systemd_user_units.py

async def disable(self, unit: str) -> None:
    """Stops a unit and stops it starting at login.

    Args:
        unit (str): The unit.

    Raises:
        UnitError: The name is not a service unit, or systemctl failed.
    """
    self._check(unit)
    await self._systemctl('disable', '--now', unit)