Starlette is a lightweight, ASGI-compliant Python framework for asynchronous web applications. It supports REST endpoints and WebSockets, allowing applications to handle asynchronous database or API work without relying only on multithreading or multiprocessing. section 9.0 section 9.3
A Starlette application is organized around a few practical pieces:
- Create a Starlette application with route objects and startup or shutdown coroutines. section 9.3.1
- Map string paths such as /brands to request-handling coroutines that return responses. section 9.3.1
- Store shared resources such as an asyncpg database pool in the application state object. section 9.3.1
- Run the application with Uvicorn and multiple workers when deployment requires additional throughput. section 9.3.1
Starlette also provides an understandable WebSocket interface. A WebSocket keeps one persistent, bidirectional connection open, so the server can push updated information instead of waiting for repeated HTTP polling requests. section 9.3.2
Brands API and live user counter
A database-backed /brands endpoint creates a pool during startup, queries the brand table, and returns JSON. A WebSocketEndpoint can track connected sockets, broadcast the current user count on connection and disconnection, and remove clients when sending fails. section 9.3.1 section 9.3.2
Listing 9.8 A Starlette brands endpoint
import asyncpg
from asyncpg import Record
from asyncpg.pool import Pool
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.routing import Route
from typing import List, Dict
async def create_database_pool():
pool: Pool = await asyncpg.create_pool(host='127.0.0.1',
port=5432,
user='postgres',
password='password',
database='products',
min_size=6,
max_size=6)
app.state.DB = pool
async def destroy_database_pool():
pool = app.state.DB
await pool.close()
async def brands(request: Request) -> Response:
connection: Pool = request.app.state.DB
brand_query = 'SELECT brand_id, brand_name FROM brand'
results: List[Record] = await connection.fetch(brand_query)
result_as_dict: List[Dict] = [dict(brand) for brand in results]
return JSONResponse(result_as_dict)
app = Starlette(routes=[Route('/brands', brands)],
on_startup=[create_database_pool],
on_shutdown=[destroy_database_pool])
Use the complete listing to inspect an asyncpg-backed Starlette application with lifecycle hooks, application state, routing, and a JSON endpoint. section 9.3.1
Listing 9.9 A Starlette WebSocket endpoint
import asyncio
from starlette.applications import Starlette
from starlette.endpoints import WebSocketEndpoint
from starlette.routing import WebSocketRoute
class UserCounter(WebSocketEndpoint):
encoding = 'text'
sockets = []
async def on_connect(self, websocket): #1
await websocket.accept()
UserCounter.sockets.append(websocket)
await self._send_count()
async def on_disconnect(self, websocket, close_code): #2
UserCounter.sockets.remove(websocket)
await self._send_count()
async def on_receive(self, websocket, data):
pass
async def _send_count(self): #3
if len(UserCounter.sockets) > 0:
count_str = str(len(UserCounter.sockets))
task_to_socket = {asyncio.create_task(websocket.send_text(count_str)): websocket
for websocket
in UserCounter.sockets}
done, pending = await asyncio.wait(task_to_socket)
for task in done:
if task.exception() is not None:
if task_to_socket[task] in UserCounter.sockets:
UserCounter.sockets.remove(task_to_socket[task])
app = Starlette(routes=[WebSocketRoute('/counter', UserCounter)])
Use the endpoint listing to follow connection, disconnection, message handling, and broadcast logic for a live WebSocket counter. section 9.3.2
