Skip to content

trading_agent.gateway.gateway_application

Builds the gateway's FastAPI application and runs the desks inside it.

The desks' loops are started when the web server starts and cancelled when it stops, so the gateway is one process that owns every desk.

Typical usage example:

gateway = GatewayApplication(event_bus, desks, guard) uvicorn.run(gateway.build(), host='127.0.0.1', port=8120)

BackgroundWorker

Bases: Protocol

Anything with a run coroutine that loops until it is cancelled.

run async 

run()

Runs until cancelled.

Source code in src/trading_agent/gateway/gateway_application.py 
35
36
async def run(self) -> None:
    """Runs until cancelled."""

GatewayApplication 

GatewayApplication(event_bus, desks, guard, background_workers=(), redactor=None, webhook_token='', route_groups=(), frontend_directory=None, allowed_origins=())

The gateway: its web routes and the desks it runs.

Creates the gateway without starting anything.

Parameters:

Name Type Description Default
event_bus EventBus

The bus every component publishes on.

 required
desks Mapping[str, DeskRuntime]

The desks to run, keyed by name.

 required
guard AccessGuard

The guard that checks the access token.

 required
background_workers Sequence[BackgroundWorker]

Other loops to run beside the desks, such as the trade journal.

 ()
redactor SecretRedactor | None

Masks secrets in the live event stream, or None to send events unchanged.

 None
webhook_token str

The token incoming webhooks must present, or an empty string to switch them off.

 ''
route_groups Sequence[RouteGroup]

The API's route groups; when empty, only the basic desk routes are offered.

 ()
frontend_directory Path | None

The built web page to serve at every other address, or None to serve no page.

 None
allowed_origins Sequence[str]

The other web origins whose pages may call the API, such as the Android app's "https://localhost"; empty allows none.

 ()
Source code in src/trading_agent/gateway/gateway_application.py 
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
68
69
70
71
72
73
74
75
def __init__(
    self,
    event_bus: event_bus_module.EventBus,
    desks: Mapping[str, desk_runtime.DeskRuntime],
    guard: access_guard.AccessGuard,
    background_workers: Sequence[BackgroundWorker] = (),
    redactor: secret_redactor.SecretRedactor | None = None,
    webhook_token: str = '',
    route_groups: Sequence[route_group.RouteGroup] = (),
    frontend_directory: Path | None = None,
    allowed_origins: Sequence[str] = (),
):
    """Creates the gateway without starting anything.

    Args:
        event_bus (event_bus_module.EventBus): The bus every component publishes on.
        desks (Mapping[str, desk_runtime.DeskRuntime]): The desks to run, keyed by name.
        guard (access_guard.AccessGuard): The guard that checks the access token.
        background_workers (Sequence[BackgroundWorker]): Other loops to run beside the desks, such as the trade journal.
        redactor (secret_redactor.SecretRedactor | None): Masks secrets in the live event stream, or None to send events unchanged.
        webhook_token (str): The token incoming webhooks must present, or an empty string to switch them off.
        route_groups (Sequence[route_group.RouteGroup]): The API's route groups; when empty, only the basic desk routes are offered.
        frontend_directory (Path | None): The built web page to serve at every other address, or None to serve no page.
        allowed_origins (Sequence[str]): The other web origins whose pages may call the API, such as the Android app's "https://localhost"; empty allows none.
    """
    self._event_bus = event_bus
    self._desks = desks
    self._guard = guard
    self._background_workers = list(background_workers)
    self._redactor = redactor
    self._webhook_token = webhook_token
    self._route_groups = list(route_groups)
    self._frontend_directory = frontend_directory
    self._allowed_origins = list(allowed_origins)

build 

build()

Builds the FastAPI application with every route.

Returns:

Type Description
FastAPI

The application, ready for uvicorn.
Source code in src/trading_agent/gateway/gateway_application.py 
 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
104
105
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
132
133
def build(self) -> fastapi.FastAPI:
    """Builds the FastAPI application with every route.

    Returns:
        fastapi.FastAPI: The application, ready for uvicorn.
    """
    application = fastapi.FastAPI(
        title='trading-agent gateway',
        lifespan=self._lifespan,
    )
    if self._allowed_origins:
        application.add_middleware(
            cors.CORSMiddleware,
            allow_origins=self._allowed_origins,
            allow_methods=[
                'GET',
                'POST',
                'PUT',
                'PATCH',
                'DELETE',
            ],
            allow_headers=[
                'Authorization',
                'Content-Type',
            ],
            allow_credentials=False,
        )
    application.include_router(
        health_routes.HealthRoutes(self._desks).router()
    )
    application.include_router(
        webhook_routes.WebhookRoutes(
            self._desks,
            self._webhook_token,
        ).router()
    )
    route_groups = self._route_groups
    if not route_groups:
        route_groups = [
            desk_routes.DeskRoutes(self._desks, self._guard),
        ]
    for group in route_groups:
        application.include_router(group.router())
    application.include_router(
        event_routes.EventRoutes(
            self._event_bus,
            self._guard,
            self._redactor,
        ).router()
    )
    if self._frontend_directory is not None:
        application.include_router(
            frontend_routes.FrontendRoutes(
                self._frontend_directory
            ).router()
        )
    return application