Skip to content

trading_agent.gateway.routes.app_routes

Routes for the packages' web apps.

GET /api/apps every app with its address, whether it answers and its unit's state POST /api/apps/{package}/{action} start, stop or restart an app's systemd unit

Only units named by an installed package's app.toml can be started or stopped here.

Typical usage example:

application.include_router(AppRoutes(guard, catalog).router())

AppRoutes 

AppRoutes(guard, catalog)

Bases: RouteGroup

The app routes.

Creates the routes.

Parameters:

Name Type Description Default
guard AccessGuard

Checks the access token.

 required
catalog AppCatalog

The packages' apps.

 required
Source code in src/trading_agent/gateway/routes/app_routes.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    guard: access_guard.AccessGuard,
    catalog: app_catalog_module.AppCatalog,
):
    """Creates the routes.

    Args:
        guard (access_guard.AccessGuard): Checks the access token.
        catalog (app_catalog_module.AppCatalog): The packages' apps.
    """
    super().__init__(guard)
    self._catalog = catalog

list_apps async 

list_apps()

Every app and its state, checked all at once.

Returns:

Type Description
dict[str, Any]

The apps.
Source code in src/trading_agent/gateway/routes/app_routes.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def list_apps(self) -> dict[str, Any]:
    """Every app and its state, checked all at once.

    Returns:
        dict[str, Any]: The apps.
    """
    apps = self._catalog.apps()
    statuses = await asyncio.gather(
        *[self._catalog.status(app) for app in apps]
    )
    entries = []
    for app, status in zip(apps, statuses, strict=True):
        entries.append(self._entry(app, status))
    return {
        'apps': entries,
    }

act async 

act(package, action)

Starts, stops or restarts an app's unit.

Parameters:

Name Type Description Default
package str

The package.

 required
action str

"start", "stop" or "restart".

 required

Returns:

Type Description
dict[str, Any]

The app and its state afterwards.

Raises:

Type Description
HTTPException

No such app (status 404), the app has no unit or the action is unknown (status 400), or systemctl failed (status 502).
Source code in src/trading_agent/gateway/routes/app_routes.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def act(self, package: str, action: str) -> dict[str, Any]:
    """Starts, stops or restarts an app's unit.

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

    Returns:
        dict[str, Any]: The app and its state afterwards.

    Raises:
        fastapi.HTTPException: No such app (status 404), the app has no unit or the action is unknown (status 400), or systemctl failed (status 502).
    """
    app = self._catalog.find(package)
    if app is None:
        raise fastapi.HTTPException(
            status_code=404,
            detail=f'No installed package named {package} has an app',
        )
    if app.manifest is None or app.manifest.service is None:
        raise fastapi.HTTPException(
            status_code=400, detail=f'{package} names no systemd unit'
        )
    if action not in systemd_user_units.ACTIONS:
        raise fastapi.HTTPException(
            status_code=400, detail='Use start, stop or restart'
        )
    try:
        await self._catalog.units.act(app.manifest.service.unit, action)
    except systemd_user_units.UnitError as error:
        raise fastapi.HTTPException(
            status_code=502, detail=str(error)
        ) from error
    return self._entry(app, await self._catalog.status(app))