trading_agent.extensions.app_catalog

The web apps that installed packages bring, and whether each is up.

An app is up when its page answers, which is checked by asking its health address with a short timeout; any answer below 500 counts, because an app that asks for a login still answers. Its unit's state comes from systemd.

Typical usage example:

catalog = AppCatalog(packages, SystemdUserUnits()) for app in catalog.apps(): status = await catalog.status(app)

InstalledApp dataclass

InstalledApp(package, folder, manifest, error)

One package's app.

Attributes:

Name	Type	Description
package	str	The package's name, which also names the app in addresses.
folder	Path	The installed package's folder.
manifest	AppManifest | None	The app's settings, or None when app.toml cannot be read.
error	str	Why app.toml cannot be read, or an empty string.

AppStatus dataclass

AppStatus(healthy, health_detail, unit_state, unit_error)

Whether one app is up.

Attributes:

Name	Type	Description
healthy	bool	Whether its page answered.
health_detail	str	The answer's status code, or why there was none.
unit_state	UnitState | None	Its unit's state, or None when it has no unit or systemd could not be asked.
unit_error	str	Why systemd could not be asked, or an empty string.

AppCatalog

AppCatalog(packages, units)

Finds the apps of installed packages and checks them.

Creates the catalog.

Parameters:

Name	Type	Description	Default
packages	InstalledPackages	The installed packages.	required
units	SystemdUserUnits	Talks to systemd.	required

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

def __init__(
    self,
    packages: installed_packages_module.InstalledPackages,
    units: systemd_user_units_module.SystemdUserUnits,
):
    """Creates the catalog.

    Args:
        packages (installed_packages_module.InstalledPackages): The installed packages.
        units (systemd_user_units_module.SystemdUserUnits): Talks to systemd.
    """
    self._packages = packages
    self.units = units
    self._reader = app_manifest_module.AppManifestReader()

apps

apps()

Every installed package that has an app.toml, read afresh.

Returns:

Type	Description
list[InstalledApp]	The apps, in package-name order.

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

def apps(self) -> list[InstalledApp]:
    """Every installed package that has an app.toml, read afresh.

    Returns:
        list[InstalledApp]: The apps, in package-name order.
    """
    apps = []
    for folder in self._packages.folders():
        path = folder / app_manifest_module.APP_FILE_NAME
        if not path.is_file():
            continue
        try:
            manifest = self._reader.read(path)
        except app_manifest_module.AppManifestError as error:
            apps.append(InstalledApp(folder.name, folder, None, str(error)))
            continue
        apps.append(InstalledApp(folder.name, folder, manifest, ''))
    return apps

find

find(package)

Finds one package's app.

Parameters:

Name	Type	Description	Default
package	str	The package's name.	required

Returns:

Type	Description
InstalledApp | None	The app, or None when that package has none.

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

def find(self, package: str) -> InstalledApp | None:
    """Finds one package's app.

    Args:
        package (str): The package's name.

    Returns:
        InstalledApp | None: The app, or None when that package has none.
    """
    for app in self.apps():
        if app.package == package:
            return app
    return None

status async

status(app)

Checks whether an app's page answers and asks systemd about its unit, both at once.

Parameters:

Name	Type	Description	Default
app	InstalledApp	The app.	required

Returns:

Type	Description
AppStatus	The app's state.

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

async def status(self, app: InstalledApp) -> AppStatus:
    """Checks whether an app's page answers and asks systemd about its unit, both at once.

    Args:
        app (InstalledApp): The app.

    Returns:
        AppStatus: The app's state.
    """
    if app.manifest is None:
        return AppStatus(False, app.error, None, '')
    health, unit = await asyncio.gather(
        self._check_health(app.manifest.health_address()),
        self._unit_state(app.manifest),
    )
    healthy, detail = health
    unit_state, unit_error = unit
    return AppStatus(healthy, detail, unit_state, unit_error)