Skip to content

trading_agent.client.app_commands

The trading-agent apps commands: list the packages' apps, and enable or disable their systemd units.

Enabling links the unit file the app's app.toml names (when systemd does not know the unit yet), then enables and starts it, so the app starts at every login. This is a command rather than a web page button, because it adds a program to the machine's start-up.

Typical usage example:

exit_code = asyncio.run(AppCommands(catalog).enable('sridhara'))

AppCommands 

AppCommands(catalog, output=stdout)

Lists and sets up the packages' apps for the user.

Creates the commands.

Parameters:

Name Type Description Default
catalog AppCatalog

The packages' apps.

 required
output TextIO

Where to print.

 stdout
Source code in src/trading_agent/client/app_commands.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
def __init__(
    self,
    catalog: app_catalog_module.AppCatalog,
    output: TextIO = sys.stdout,
):
    """Creates the commands.

    Args:
        catalog (app_catalog_module.AppCatalog): The packages' apps.
        output (TextIO): Where to print.
    """
    self._catalog = catalog
    self._output = output

list async 

list()

Prints each app with its address, whether it answers and its unit's state.

Returns:

Type Description
int

0 when every app.toml is readable, 1 otherwise.
Source code in src/trading_agent/client/app_commands.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
async def list(self) -> int:
    """Prints each app with its address, whether it answers and its unit's state.

    Returns:
        int: 0 when every app.toml is readable, 1 otherwise.
    """
    apps = self._catalog.apps()
    if not apps:
        self._print('No installed package has an app.')
    failed = False
    for app in apps:
        if app.manifest is None:
            self._print(f'{app.package}: cannot read app.toml: {app.error}')
            failed = True
            continue
        status = await self._catalog.status(app)
        health = (
            'answers'
            if status.healthy
            else f'does not answer ({status.health_detail})'
        )
        self._print(
            f'{app.package}: {app.manifest.app.title} at {app.manifest.app.url}, {health}'
        )
        if status.unit_state is not None:
            state = status.unit_state
            self._print(
                f'  unit {app.manifest.service.unit}: {state.load_state}, {state.active_state} ({state.sub_state})'
            )
        elif status.unit_error:
            self._print(f'  unit: {status.unit_error}')
    if failed:
        return 1
    return 0

enable async 

enable(package)

Links, enables and starts an app's unit.

Parameters:

Name Type Description Default
package str

The package.

 required

Returns:

Type Description
int

0 when the unit is running, 1 otherwise.
Source code in src/trading_agent/client/app_commands.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def enable(self, package: str) -> int:
    """Links, enables and starts an app's unit.

    Args:
        package (str): The package.

    Returns:
        int: 0 when the unit is running, 1 otherwise.
    """
    app = self._service_app(package)
    if app is None:
        return 1
    try:
        await self._catalog.units.enable(
            app.manifest.service.unit,
            app.manifest.unit_file_path(app.folder),
        )
    except systemd_user_units.UnitError as error:
        self._print(str(error))
        return 1
    self._print(
        f'Enabled and started {app.manifest.service.unit}; it now starts at every login.'
    )
    return 0

disable async 

disable(package)

Stops an app's unit and stops it starting at login.

Parameters:

Name Type Description Default
package str

The package.

 required

Returns:

Type Description
int

0 when done, 1 otherwise.
Source code in src/trading_agent/client/app_commands.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def disable(self, package: str) -> int:
    """Stops an app's unit and stops it starting at login.

    Args:
        package (str): The package.

    Returns:
        int: 0 when done, 1 otherwise.
    """
    app = self._service_app(package)
    if app is None:
        return 1
    try:
        await self._catalog.units.disable(app.manifest.service.unit)
    except systemd_user_units.UnitError as error:
        self._print(str(error))
        return 1
    self._print(f'Stopped and disabled {app.manifest.service.unit}.')
    return 0