Skip to content

trading_agent.gateway.routes.channel_routes

Routes for the messaging channels: Telegram's state and its pairings.

GET /api/channels Telegram's state, paired chats and waiting codes POST /api/channels/telegram/pairings/{code}/approve pair the chat that was given a code DELETE /api/channels/telegram/chats/{chat_id} unpair a chat

Typical usage example:

application.include_router(ChannelRoutes(guard, store, configuration, secrets_present, channel).router())

ChannelRoutes 

ChannelRoutes(guard, store, configuration, token_present, channel)

Bases: RouteGroup

The channel routes.

Creates the routes.

Parameters:

Name Type Description Default
guard AccessGuard

Checks the access token.

 required
store PairingStore

The pairing file.

 required
configuration TelegramConfiguration

The [telegram] settings.

 required
token_present bool

Whether TRADING_AGENT_TELEGRAM_BOT_TOKEN is set.

 required
channel TelegramChannel | None

The running channel, or None when Telegram is off.

 required
Source code in src/trading_agent/gateway/routes/channel_routes.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    guard: access_guard.AccessGuard,
    store: pairing_store_module.PairingStore,
    configuration: agent_configuration.TelegramConfiguration,
    token_present: bool,
    channel: telegram_channel_module.TelegramChannel | None,
):
    """Creates the routes.

    Args:
        guard (access_guard.AccessGuard): Checks the access token.
        store (pairing_store_module.PairingStore): The pairing file.
        configuration (agent_configuration.TelegramConfiguration): The [telegram] settings.
        token_present (bool): Whether TRADING_AGENT_TELEGRAM_BOT_TOKEN is set.
        channel (telegram_channel_module.TelegramChannel | None): The running channel, or None when Telegram is off.
    """
    super().__init__(guard)
    self._store = store
    self._configuration = configuration
    self._token_present = token_present
    self._channel = channel

channels 

channels()

Telegram's state.

Returns:

Type Description
dict[str, Any]

Whether it is enabled and has a token, its connection state, its bot name, the paired chats and the waiting codes.
Source code in src/trading_agent/gateway/routes/channel_routes.py 
68
69
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
def channels(self) -> dict[str, Any]:
    """Telegram's state.

    Returns:
        dict[str, Any]: Whether it is enabled and has a token, its connection state, its bot name, the paired chats and the waiting codes.
    """
    state = 'off'
    error = ''
    username = ''
    if self._channel is not None:
        state = self._channel.state
        error = self._channel.error
        username = self._channel.username
    paired = []
    for chat in self._store.paired_chats():
        paired.append(dataclasses.asdict(chat))
    pending = []
    for waiting in self._store.pending():
        pending.append(dataclasses.asdict(waiting))
    return {
        'telegram': {
            'enabled': self._configuration.enabled,
            'token_present': self._token_present,
            'state': state,
            'error': error,
            'username': username,
            'default_desk': self._configuration.default_desk,
            'notifications': self._configuration.notifications,
            'paired': paired,
            'pending': pending,
        },
    }

approve async 

approve(code)

Pairs the chat that was given a code.

Parameters:

Name Type Description Default
code str

The code.

 required

Returns:

Type Description
dict[str, Any]

The paired chat.

Raises:

Type Description
HTTPException

No valid code matches (status 404).
Source code in src/trading_agent/gateway/routes/channel_routes.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def approve(self, code: str) -> dict[str, Any]:
    """Pairs the chat that was given a code.

    Args:
        code (str): The code.

    Returns:
        dict[str, Any]: The paired chat.

    Raises:
        fastapi.HTTPException: No valid code matches (status 404).
    """
    try:
        if self._channel is not None:
            chat = await self._channel.approve(code)
        else:
            chat = self._store.approve(
                code, self._configuration.default_desk
            )
    except pairing_store_module.PairingError as error:
        raise fastapi.HTTPException(
            status_code=404, detail=str(error)
        ) from error
    return dataclasses.asdict(chat)

revoke 

revoke(chat_id)

Unpairs a chat.

Parameters:

Name Type Description Default
chat_id int

The chat.

 required

Returns:

Type Description
dict[str, Any]

The chat that was removed.

Raises:

Type Description
HTTPException

The chat is not paired (status 404).
Source code in src/trading_agent/gateway/routes/channel_routes.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def revoke(self, chat_id: int) -> dict[str, Any]:
    """Unpairs a chat.

    Args:
        chat_id (int): The chat.

    Returns:
        dict[str, Any]: The chat that was removed.

    Raises:
        fastapi.HTTPException: The chat is not paired (status 404).
    """
    try:
        chat = self._store.revoke(chat_id)
    except pairing_store_module.PairingError as error:
        raise fastapi.HTTPException(
            status_code=404, detail=str(error)
        ) from error
    return dataclasses.asdict(chat)