Python with Starlette

Cover Page

Back-end Page

Add dependencies

Change to your chatterd directory and add two packages including the PostgreSQL adaptor:

server$ cd ~/reactive/chatterd
server$ uv add pydantic-core '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()

Continuing in main.py, find the routing table routes and add two routes right after the routes 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 routing table:

# must come after routes and lifespan definitions
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 psycopg.errors import StringDataRightTruncation
from pydantic_core import to_jsonable_python
from typing import Optional
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:
    name: 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 ` to_jsonable_python() from the pydantic-core 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 name, message, id, time FROM chatts ORDER BY time ASC;')
                return JSONResponse(to_jsonable_python(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 (name, message, id) VALUES '
                    '(%s, %s, gen_random_uuid());', (chatt.name, 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$

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 January 23rd, 2026