Skip to content

athena.server

ServerBuilder

Builder for configuring the server

Source code in athena/server.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
class ServerBuilder():
    """Builder for configuring the server
    """
    def __init__(self):
        self._configurators = []
        self._claimed_ports = set()

    def add_server(self, configure: Callable[[ServerConfigurator], ServerConfigurator]):
        """Add a new server configuration

        Args:
            configure (Callable[[ServerConfigurator], ServerConfigurator]): configure the server
        """
        configurator = configure(ServerConfigurator())
        if configurator._port is not None:
            if configurator._port in self._claimed_ports:
                raise AthenaException(f'Multiple servers are trying to claim port {configurator._port}')
            self._claimed_ports.add(configurator._port)
        self._configurators.append(configurator)
        return self

    def _build(self):
        available_port = 5000
        for configurator in self._configurators:
            if configurator._port is not None:
                continue

            while available_port in self._claimed_ports:
                available_port += 1
            configurator.port(available_port)
            self._claimed_ports.add(available_port)
            available_port += 1

        start_functions = []
        for configurator in self._configurators:
            def start_function(cfg):
                server = cfg._build()
                def after_request(response):
                    _logger.info(f'[{request.host}] [{request.method}] {request.path} {response.status}')
                    return response
                server.after_request(after_request)

                server.run(port=cfg._port, host=cfg._host, debug=False)
            start_functions.append((start_function, (configurator,)))

        return start_functions

add_server(configure)

Add a new server configuration

Parameters:

Name Type Description Default
configure Callable[[ServerConfigurator], ServerConfigurator]

configure the server

 required
Source code in athena/server.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
def add_server(self, configure: Callable[[ServerConfigurator], ServerConfigurator]):
    """Add a new server configuration

    Args:
        configure (Callable[[ServerConfigurator], ServerConfigurator]): configure the server
    """
    configurator = configure(ServerConfigurator())
    if configurator._port is not None:
        if configurator._port in self._claimed_ports:
            raise AthenaException(f'Multiple servers are trying to claim port {configurator._port}')
        self._claimed_ports.add(configurator._port)
    self._configurators.append(configurator)
    return self

ServerConfigurator

Builder for creating a server configuration

Source code in athena/server.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
class ServerConfigurator():
    """Builder for creating a server configuration
    """
    def __init__(self):
        self._routes: list[tuple[tuple[str, ...], str, Callable[[RouteBuilder], RouteBuilder]]] = []
        self._host: str | None = None
        self._port: int | None = None

    def host(self, host: str):
        """Set the host for the server.

        Args:
            host (str): host the server is running on. Set to 0.0.0.0 to make the server publicly available.
        """
        self._host = host
        return self

    def port(self, port: int):
        """Set the port for the server.

        Args:
            port (int): port the server is running on.
        """
        self._port = port
        return self

    def send(self, method: str | list[str], path: str, build_route: Callable[[RouteBuilder], RouteBuilder]):
        """
        Add an http request handler.

        Args:
            method (str): HTTP method ('GET', 'POST', etc).
            path (str): endpoint for the request
            build_route (Callable[[RouteBuilder], RouteBuilder]):
                Function to build or modify the request.
        """
        if path[0] != '/':
            path = f'/{path}'

        if isinstance(method, str):
            method_tuple = (method,)
        else:
            method_tuple = tuple(method)
        self._routes.append((method_tuple, path, build_route))
        return self

    def get(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
        """
        Add an http get request handler.

        Args:
            path (str): endpoint for the request
            build_route (Callable[[RouteBuilder], RouteBuilder]):
                Function to build or modify the request.
        """
        self.send('GET', path, build_route)
        return self
    def post(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
        """
        Add an http post request handler.

        Args:
            path (str): endpoint for the request
            build_route (Callable[[RouteBuilder], RouteBuilder]):
                Function to build or modify the request.
        """
        self.send('POST', path, build_route)
        return self
    def put(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
        """
        Add an http put request handler.

        Args:
            path (str): endpoint for the request
            build_route (Callable[[RouteBuilder], RouteBuilder]):
                Function to build or modify the request.
        """
        self.send('PUT', path, build_route)
        return self
    def delete(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
        """
        Add an http delete request handler.

        Args:
            path (str): endpoint for the request
            build_route (Callable[[RouteBuilder], RouteBuilder]):
                Function to build or modify the request.
        """
        self.send('DELETE', path, build_route)
        return self

    def _build(self) -> Flask:
        def create_route_handler(func):
            def handler():
                builder = RouteBuilder(ServerRequest(request))
                try:
                    return func(builder)._complete()
                except:
                    exc_type, exc_value, exc_traceback = sys.exc_info()
                    response = flask_jsonify({
                        'error': exc_type.__name__ if exc_type is not None else None,
                        'message': str(exc_value),
                        'stacktrace': ''.join(traceback.format_exception(exc_type, exc_value, exc_traceback))
                    })
                    response.status_code = 500
                    return response
            return handler
        app = Flask(__name__)
        for methods, url, func in self._routes:
            app.route(url, methods=methods, endpoint=str(uuid.uuid4()))(create_route_handler(func))

        return app

delete(path, build_route)

Add an http delete request handler.

Parameters:

Name Type Description Default
path str

endpoint for the request

 required
build_route Callable[[RouteBuilder], RouteBuilder]

Function to build or modify the request.

 required
Source code in athena/server.py 
260
261
262
263
264
265
266
267
268
269
270
def delete(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
    """
    Add an http delete request handler.

    Args:
        path (str): endpoint for the request
        build_route (Callable[[RouteBuilder], RouteBuilder]):
            Function to build or modify the request.
    """
    self.send('DELETE', path, build_route)
    return self

get(path, build_route)

Add an http get request handler.

Parameters:

Name Type Description Default
path str

endpoint for the request

 required
build_route Callable[[RouteBuilder], RouteBuilder]

Function to build or modify the request.

 required
Source code in athena/server.py 
227
228
229
230
231
232
233
234
235
236
237
def get(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
    """
    Add an http get request handler.

    Args:
        path (str): endpoint for the request
        build_route (Callable[[RouteBuilder], RouteBuilder]):
            Function to build or modify the request.
    """
    self.send('GET', path, build_route)
    return self

host(host)

Set the host for the server.

Parameters:

Name Type Description Default
host str

host the server is running on. Set to 0.0.0.0 to make the server publicly available.

 required
Source code in athena/server.py 
189
190
191
192
193
194
195
196
def host(self, host: str):
    """Set the host for the server.

    Args:
        host (str): host the server is running on. Set to 0.0.0.0 to make the server publicly available.
    """
    self._host = host
    return self

port(port)

Set the port for the server.

Parameters:

Name Type Description Default
port int

port the server is running on.

 required
Source code in athena/server.py 
198
199
200
201
202
203
204
205
def port(self, port: int):
    """Set the port for the server.

    Args:
        port (int): port the server is running on.
    """
    self._port = port
    return self

post(path, build_route)

Add an http post request handler.

Parameters:

Name Type Description Default
path str

endpoint for the request

 required
build_route Callable[[RouteBuilder], RouteBuilder]

Function to build or modify the request.

 required
Source code in athena/server.py 
238
239
240
241
242
243
244
245
246
247
248
def post(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
    """
    Add an http post request handler.

    Args:
        path (str): endpoint for the request
        build_route (Callable[[RouteBuilder], RouteBuilder]):
            Function to build or modify the request.
    """
    self.send('POST', path, build_route)
    return self

put(path, build_route)

Add an http put request handler.

Parameters:

Name Type Description Default
path str

endpoint for the request

 required
build_route Callable[[RouteBuilder], RouteBuilder]

Function to build or modify the request.

 required
Source code in athena/server.py 
249
250
251
252
253
254
255
256
257
258
259
def put(self, path, build_route: Callable[[RouteBuilder], RouteBuilder]):
    """
    Add an http put request handler.

    Args:
        path (str): endpoint for the request
        build_route (Callable[[RouteBuilder], RouteBuilder]):
            Function to build or modify the request.
    """
    self.send('PUT', path, build_route)
    return self

send(method, path, build_route)

Add an http request handler.

Parameters:

Name Type Description Default
method str

HTTP method ('GET', 'POST', etc).

 required
path str

endpoint for the request

 required
build_route Callable[[RouteBuilder], RouteBuilder]

Function to build or modify the request.

 required
Source code in athena/server.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
def send(self, method: str | list[str], path: str, build_route: Callable[[RouteBuilder], RouteBuilder]):
    """
    Add an http request handler.

    Args:
        method (str): HTTP method ('GET', 'POST', etc).
        path (str): endpoint for the request
        build_route (Callable[[RouteBuilder], RouteBuilder]):
            Function to build or modify the request.
    """
    if path[0] != '/':
        path = f'/{path}'

    if isinstance(method, str):
        method_tuple = (method,)
    else:
        method_tuple = tuple(method)
    self._routes.append((method_tuple, path, build_route))
    return self

RouteBuilder

Builder for configuring an http route

Parameters:

Name Type Description Default
request ServerRequest

accessor for the incoming http request

 required
body ResponseBodyBuilder

builder for configuring the response body

 required
Source code in athena/server.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class RouteBuilder:
    """Builder for configuring an http route

        Args:
            request (ServerRequest): accessor for the incoming http request
            body (ResponseBodyBuilder): builder for configuring the response body
    """
    def __init__(self, request: ServerRequest):
        self.request = request
        self._response = ServerResponse()
        self.body = ResponseBodyBuilder(self)

    def status(self, status_code: int):
        """Set the response status code"""
        self._response.status_code(status_code)
        return self

    def header(self, header_key, header_value):
        """Add a header to the response."""
        self._response.header(header_key, header_value)
        return self

    def _complete(self):
        return self._response._to_flask_response()

header(header_key, header_value)

Add a header to the response.

Source code in athena/server.py 
173
174
175
176
def header(self, header_key, header_value):
    """Add a header to the response."""
    self._response.header(header_key, header_value)
    return self

status(status_code)

Set the response status code

Source code in athena/server.py 
168
169
170
171
def status(self, status_code: int):
    """Set the response status code"""
    self._response.status_code(status_code)
    return self

ResponseBodyBuilder

Builder for configuring the response body

Source code in athena/server.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class ResponseBodyBuilder:
    """Builder for configuring the response body
    """
    def __init__(self, parent: RouteBuilder):
        self._parent = parent

    def json(self, payload) -> RouteBuilder:
        """set the json payload"""
        self._parent._response.json(payload)
        return self._parent

    def text(self, payload) -> RouteBuilder:
        """set the text payload"""
        self._parent._response.text(payload)
        return self._parent

    def html(self, payload) -> RouteBuilder:
        """set the html payload"""
        self._parent._response.html(payload)
        return self._parent

    def data(self, mimetype: str, payload: bytes) -> RouteBuilder:
        """set the raw payload"""
        self._parent._response.data(mimetype, payload)
        return self._parent

data(mimetype, payload)

set the raw payload

Source code in athena/server.py 
151
152
153
154
def data(self, mimetype: str, payload: bytes) -> RouteBuilder:
    """set the raw payload"""
    self._parent._response.data(mimetype, payload)
    return self._parent

html(payload)

set the html payload

Source code in athena/server.py 
146
147
148
149
def html(self, payload) -> RouteBuilder:
    """set the html payload"""
    self._parent._response.html(payload)
    return self._parent

json(payload)

set the json payload

Source code in athena/server.py 
136
137
138
139
def json(self, payload) -> RouteBuilder:
    """set the json payload"""
    self._parent._response.json(payload)
    return self._parent

text(payload)

set the text payload

Source code in athena/server.py 
141
142
143
144
def text(self, payload) -> RouteBuilder:
    """set the text payload"""
    self._parent._response.text(payload)
    return self._parent

ServerRequest

Accessor for the request sent to the server

Parameters:

Name Type Description Default
body ServerRequestBody

accessor for the request body

 required
Source code in athena/server.py 
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
76
77
78
79
80
81
82
83
class ServerRequest:
    """Accessor for the request sent to the server

    Args:
        body (ServerRequestBody): accessor for the request body
    """
    def __init__(self, request: Request):
        self._request = request
        self.body = ServerRequestBody(request)

    @property
    def query(self) -> MultiDictProxy:
        """request query data"""
        return MultiDictProxy(MultiDict(list(self._request.args.items(multi=True))))

    @property
    def cookies(self):
        """request cookie jar"""
        return self._request.cookies

    @property
    def headers(self) -> CIMultiDictProxy:
        """request headers"""
        return CIMultiDictProxy(CIMultiDict(self._request.headers.items()))

    @property
    def url(self):
        """request url"""
        return self._request.url

    @property
    def method(self):
        """request method (verb)"""
        return self._request.method

cookies

request cookie jar

headers: CIMultiDictProxy

request headers

method

request method (verb)

query: MultiDictProxy

request query data

url

request url

ServerRequestBody

Accessor for the request body

Source code in athena/server.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class ServerRequestBody:
    """Accessor for the request body"""
    def __init__(self, request: Request):
        self._request = request

    @property
    def form(self) -> MultiDictProxy:
        """form payload"""
        return MultiDictProxy(MultiDict(self._request.form.items()))

    @property
    def files(self):
        """files payload"""
        return self._request.files

    @property
    def data(self) -> bytes:
        """raw bytes payload"""
        return self._request.data

    @property
    def json(self):
        """json payload"""
        return self._request.json

data: bytes

raw bytes payload

files

files payload

form: MultiDictProxy

form payload

json

json payload