atoti.Session.endpoint()#

Session.endpoint(route, *, method='GET')#

Create a custom endpoint at /atoti/pyapi/{route}.

This is useful to reuse Atoti’s built-in server instead of adding a FastAPI or Flask server to the project. This way, when deploying the project in a container or a VM, only one port (the one of the Atoti server) can be exposed instead of two. Since custom endpoints are exposed by Atoti’s server, they automatically inherit from the configured atoti.Session()’s authentication and https parameter.

The decorated function must take three parameters with types atoti.pyapi.User, atoti.pyapi.HttpRequest, and atoti.Session and return a response body as a Python data structure that can be converted to JSON.

Parameters:

  • route (str) –

    The path suffix after /atoti/pyapi/. For instance, if custom/search is passed, a request to /atoti/pyapi/custom/search?query=test#results will match. The route should not contain the query (?) or fragment (#).

    Path parameters can be configured by wrapping their name in curly braces in the route.

  • method (Literal['POST', 'GET', 'PUT', 'DELETE']) – The HTTP method the request must be using to trigger this endpoint. DELETE, POST, and PUT requests can have a body but it must be JSON.

Return type:

Callable[[Callable[[…], Any]], Callable[[…], Any]]

Example

>>> import httpx
>>> df = pd.DataFrame(
...     columns=["Year", "Month", "Day", "Quantity"],
...     data=[
...         (2019, 7, 1, 15),
...         (2019, 7, 2, 20),
...     ],
... )
>>> table = session.read_pandas(df, table_name="Quantity")
>>> table.head()
Year  Month  Day  Quantity
0  2019      7    1        15
1  2019      7    2        20
>>> endpoints_base_url = f"{session.url}/atoti/pyapi"
>>> @session.endpoint("tables/{table_name}/count", method="GET")
... def get_table_row_count(request, user, session):
...     table_name = request.path_parameters["table_name"]
...     return session.tables[table_name].row_count
>>> httpx.get(
...     f"{endpoints_base_url}/tables/Quantity/count"
... ).raise_for_status().json()
2
>>> @session.endpoint("tables/{table_name}/rows", method="POST")
... def append_rows_to_table(request, user, session):
...     rows = request.body
...     table_name = request.path_parameters["table_name"]
...     session.tables[table_name].append(*rows)
>>> httpx.post(
...     f"{endpoints_base_url}/tables/Quantity/rows",
...     json=[
...         {"Year": 2021, "Month": 5, "Day": 19, "Quantity": 50},
...         {"Year": 2021, "Month": 5, "Day": 20, "Quantity": 6},
...     ],
... ).status_code
200
>>> httpx.get(
...     f"{endpoints_base_url}/tables/Quantity/count"
... ).raise_for_status().json()
4
>>> table.head()
Year  Month  Day  Quantity
0  2019      7    1        15
1  2019      7    2        20
2  2021      5   19        50
3  2021      5   20         6