Skip to content

trading_agent.terminal.widgets.pending_files

The files waiting to go with the next message, shown on a line above the message box.

Each file is uploaded as soon as it is added, as on the web page, and shows "uploading…", "ready", "attached, but its text cannot be read", or why it failed. Backspace in the empty message box removes the newest file.

Typical usage example:

tray = PendingFiles(gateway) tray.add('general', Path('chart.png')) ids = await tray.ready_ids()

PendingFile dataclass 

PendingFile(local_id, path, status='uploading', attachment=None, error=None)

One file added to the message being written.

Attributes:

Name Type Description
local_id int

A number for the file while it waits.
path Path

The local file.
status str

"uploading", "ready" or "failed".
attachment dict[str, Any] | None

The gateway's attachment once uploaded, or None.
error str | None

Why it failed, or None.

PendingFiles 

PendingFiles(gateway)

Bases: Static

The tray of files; hidden while empty.

Creates the tray, empty.

Parameters:

Name Type Description Default
gateway GatewayClient

The gateway, to upload to.

 required
Source code in src/trading_agent/terminal/widgets/pending_files.py 
51
52
53
54
55
56
57
58
59
60
61
def __init__(self, gateway: gateway_client.GatewayClient):
    """Creates the tray, empty.

    Args:
        gateway (gateway_client.GatewayClient): The gateway, to upload to.
    """
    super().__init__('', id='pending-files')
    self._gateway = gateway
    self._files: list[PendingFile] = []
    self._numbers = itertools.count(1)
    self._rules = attachable_files.AttachableFiles()

on_mount 

on_mount()

Hides the tray until a file is added.

Source code in src/trading_agent/terminal/widgets/pending_files.py 
63
64
65
def on_mount(self) -> None:
    """Hides the tray until a file is added."""
    self.display = False

add 

add(desk_name, path)

Adds a file and starts uploading it.

Parameters:

Name Type Description Default
desk_name str

The desk the message goes to.

 required
path Path

The local file.

 required

Returns:

Type Description
str | None

Why the file cannot be attached, or None when it was added.
Source code in src/trading_agent/terminal/widgets/pending_files.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def add(self, desk_name: str, path: Path) -> str | None:
    """Adds a file and starts uploading it.

    Args:
        desk_name (str): The desk the message goes to.
        path (Path): The local file.

    Returns:
        str | None: Why the file cannot be attached, or None when it was added.
    """
    problem = self._rules.problem(path)
    if problem is not None:
        return problem
    pending = PendingFile(local_id=next(self._numbers), path=path)
    self._files.append(pending)
    self._show()
    self._upload(desk_name, pending)
    return None

remove_newest 

remove_newest()

Removes the newest file.

Returns:

Type Description
bool

True when a file was removed.
Source code in src/trading_agent/terminal/widgets/pending_files.py 
86
87
88
89
90
91
92
93
94
95
96
def remove_newest(self) -> bool:
    """Removes the newest file.

    Returns:
        bool: True when a file was removed.
    """
    if not self._files:
        return False
    self._files.pop()
    self._show()
    return True

has_files 

has_files()

Says whether any file waits.

Returns:

Type Description
bool

True when the tray is not empty.
Source code in src/trading_agent/terminal/widgets/pending_files.py 
 98
 99
100
101
102
103
104
def has_files(self) -> bool:
    """Says whether any file waits.

    Returns:
        bool: True when the tray is not empty.
    """
    return bool(self._files)

ready_ids async 

ready_ids()

Waits for the uploads to finish.

Returns:

Type Description
tuple[list[str], list[str]]

The ids of the uploaded files, and the problems of the files that failed.
Source code in src/trading_agent/terminal/widgets/pending_files.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
async def ready_ids(self) -> tuple[list[str], list[str]]:
    """Waits for the uploads to finish.

    Returns:
        tuple[list[str], list[str]]: The ids of the uploaded files, and the problems of the files that failed.
    """
    waited = 0.0
    while waited < _LONGEST_WAIT_SECONDS:
        uploading = False
        for pending in self._files:
            if pending.status == 'uploading':
                uploading = True
        if not uploading:
            break
        await asyncio.sleep(_WAIT_STEP_SECONDS)
        waited += _WAIT_STEP_SECONDS
    ready = []
    problems = []
    for pending in self._files:
        if pending.status == 'ready' and pending.attachment is not None:
            ready.append(pending.attachment['attachment_id'])
        else:
            problems.append(
                f'{pending.path.name}: {pending.error or "still uploading"}'
            )
    return ready, problems

clear 

clear()

Empties the tray after the message was sent.

Source code in src/trading_agent/terminal/widgets/pending_files.py 
133
134
135
136
def clear(self) -> None:
    """Empties the tray after the message was sent."""
    self._files = []
    self._show()