Cover Page

Backend Page

Python with Starlette

Add dependencies

Change into your chatterd and add the PostgreSQL adaptor:

server$ cd ~/reactive/chatterd
server$ uv add fastapi 'psycopg[binary,pool]'

chatterd app

Edit your main.py file:

server$ vi main.py

Add the following import lines:

from contextlib import asynccontextmanager
from psycopg_pool import AsyncConnectionPool

Add the following lifespan() function to set up a pool of open connections to our PostgreSQL chatterdb database. Maintaining a pool of open connections avoids the cost of opening and closing a connection on every database operation. The password used in creating AsyncConnectionPool must match the one you used when setting up Postgres earlier.

@asynccontextmanager
async def lifespan(server):
    server.pool = AsyncConnectionPool("dbname=chatterdb user=chatter password=chattchatt host=localhost", open=False)
    await server.pool.open()
    yield
    await server.pool.close()

To the routes array, add two routes right after the route for llmprompt. These serve Chatter’s two APIs: HTTP GET request, with URL endpoint getchatts, and HTTP POST request, with endpoint postchatt. With each endpoint, we also specify which HTTP method is allowed for that endpoint and the handler function assigned to it:

    Route('/getchatts', handlers.getchatts, methods=['GET']),
    Route('/postchatt', handlers.postchatt, methods=['POST']),

The functions getchatts() and postchatt() will be implemented in handlers.py.

Find the construction of the web server and provide it with thelifespan function defined above, in addition to the routes array:

server = Starlette(routes=routes, lifespan=lifespan)

We’re done with main.py. Save and exit the file.

handlers.py

Edit the file handlers.py:

server$ vi handlers.py

First add the following imports at the top of the file:

from dataclasses import dataclass
from datetime import datetime
from fastapi.encoders import jsonable_encoder
from typing import Optional
from psycopg.errors import StringDataRightTruncation
from uuid import UUID

import main

We define a Chatt class to help postchatt() deserialize JSON received from clients. Add these linese right below the above:

@dataclass
class Chatt:
    username: str
    message: str

The @dataclass annotations, among other things, automatically provides the class with a replace() constructor, allowing us to deserialize incoming JSON object into the Chatt dataclass directly.

The handler getchatts() uses an open connection from the connection pool to query the database for stored chatts. We obtain and use the connection with its context manager that automatically commits all transactions if no exception has been raised. In the case of the cursor, its context manager releases any resources used. Once all the rows from the database are retrieved, we insert the resulting array of chatts into the response JSON object. We use jsonable_encoder() from the fastapi package to convert UUIDs present in the chatts into strings for JSONResponse(). Python’s built-in json.dumps() method cannot serialize UUIDs. Add getchatts() to your handlers.py after the definition of the llmprompt() function.

async def getchatts(request):
    try:
        async with main.server.pool.connection() as connection:
            async with connection.cursor() as cursor:
                await cursor.execute('SELECT username, message, id, time FROM chatts;')
                return JSONResponse(jsonable_encoder(await cursor.fetchall()))
    except Exception as err:
        print(f'{err=}')
        return JSONResponse(f'{type(err).__name__}: {str(err)}', status_code = 500)

Similarly, postchatt() receives a posted chatt in the expected JSON format, deserializes it into the Chatt class, and inserts it into the database, using a connection from the pool. The UUID and time stamp of each chatt are generated at insertion time. Add postchatt() to the end of your handlers.py:

async def postchatt(request):
    try:
        chatt = Chatt(**(await request.json()))
    except Exception as err:
        print(f'{err=}')
        return JSONResponse(f'Unprocessable entity: {str(err)}', status_code=422)

    try:
        async with main.server.pool.connection() as connection:
            async with connection.cursor() as cursor:
                await cursor.execute('INSERT INTO chatts (username, message, id) VALUES '
                    '(%s, %s, gen_random_uuid());', (chatt.username, chatt.message))
        return JSONResponse({})
    except StringDataRightTruncation as err:
        print(f'Message too long: {str(err)}')
        return JSONResponse(f'Message too long: {str(err)}', status_code = 400)
    except Exception as err:
        print(f'{err=}')
        return JSONResponse(f'{type(err).__name__}: {str(err)}', status_code = 500)

For more Python-PostgreSQL interaction, see Passing parameters to SQL queries.

We’re done with handlers.py. Save and exit the file.

Test run

To test run your server, launch it from the command line:

server$ sudo su
# You are now root, note the command-line prompt changed from '$' or '%' to '#'.
# You can do a lot of harm with all of root's privileges, so be very careful what you do.
server# source .venv/bin/activate
(chattterd) ubuntu@server:/home/ubuntu/reactive/chatterd# granian --host 0.0.0.0 --port 443 --interface asgi --ssl-certificate /home/ubuntu/reactive/chatterd.crt --ssl-keyfile /home/ubuntu/reactive/chatterd.key --access-log --workers-kill-timeout 1 main:server
# Hit ^C to end the test
(chattterd) ubuntu@server:/home/ubuntu/reactive/chatterd# exit
# So that you're no longer root.
server$

When you run source .venv/bin/activate, your prompt should change to indicate that you are now operating within a Python virtual environment. It will look something like this: (chatterd) ubuntu@YOUR_SERVER_IP:/home/ubuntu/reactive/chatterd#.

You can test your implementation following the instructions in the Testing Chatter APIs section.

References

Prepared by Tiberiu Vilcu, Wendan Jiang, Alexander Wu, Benjamin Brengman, Ollie Elmgren, Luke Wassink, Mark Wassink, Nowrin Mohamed, Chenglin Li, Xin Jie ‘Joyce’ Liu, Yibo Pi, and Sugih Jamin Last updated August 29th, 2025