Starlette
Lightweight ASGI framework and toolkit with WebSocket and GraphQL support. Foundation technology for FastAPI.
GitHub Overview
encode/starlette
The little ASGI framework that shines. 🌟
Topics
Star History
Framework
Starlette
Overview
Starlette is a lightweight ASGI framework for developing high-performance asynchronous web applications in Python. It is used as the foundation technology for FastAPI.
Details
Starlette is a lightweight ASGI (Asynchronous Server Gateway Interface) framework developed by Tom Christie (creator of Django REST framework). With the catchphrase "The little ASGI framework that shines," it provides the minimal functionality needed for asynchronous web application development. It's adopted as the foundation for many higher-level frameworks like FastAPI and Responder, offering high performance and flexibility as a minimalist framework. While providing basic web development features such as routing, middleware system, WebSocket support, static file serving, template engine integration, authentication system, test client, and background tasks, the framework itself is designed to be extremely lightweight. Running on ASGI servers (Uvicorn, Hypercorn, etc.), it achieves performance comparable to Node.js and Go through true asynchronous processing. With minimal dependencies and a pure Python implementation, it's ideal for projects that prioritize customizability and extensibility, microservices, API development, and real-time application development.
Merits and Demerits
Merits
- Extremely lightweight: Achieves high performance with minimal feature set
- Asynchronous native: Supports true asynchronous processing with ASGI compliance
- High performance: Excellent execution speed and memory efficiency through lightweight design
- Flexible architecture: Ability to use only necessary features in combination
- WebSocket support: Real-time communication features built-in
- Powerful middleware: Highly extensible middleware system
- FastAPI foundation: Proven reliability as the foundation for FastAPI
Demerits
- Limited features: No built-in database ORM, authentication, or form processing
- Learning resources: Limited Japanese information as a relatively new framework
- Development efficiency: High-level features require self-implementation or library addition
- Ecosystem: Relatively few third-party libraries
- Asynchronous complexity: Understanding of async/await programming required
Main Links
- Starlette Official Site
- Starlette Official Documentation
- Starlette GitHub Repository
- Uvicorn ASGI Server
- FastAPI Official Site
- Awesome Starlette
Code Examples
Hello World
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
async def homepage(request):
return JSONResponse({'hello': 'world'})
app = Starlette(debug=True, routes=[
Route('/', homepage),
])
# Server startup: uvicorn main:app --reload
Routing and Response
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse, JSONResponse
from starlette.routing import Route
async def homepage(request):
return PlainTextResponse("Homepage")
async def about(request):
return PlainTextResponse("About")
async def user_profile(request):
username = request.path_params['username']
return JSONResponse({
'username': username,
'message': f'Hello, {username}!'
})
async def search(request):
query = request.query_params.get('q', '')
return JSONResponse({
'query': query,
'results': f'Search results for: {query}'
})
routes = [
Route("/", endpoint=homepage),
Route("/about", endpoint=about),
Route("/user/{username}", endpoint=user_profile),
Route("/search", endpoint=search),
]
app = Starlette(routes=routes)
Middleware and CORS
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
from starlette.responses import JSONResponse
from starlette.routing import Route
async def homepage(request):
# Session usage example
request.session['visits'] = request.session.get('visits', 0) + 1
return JSONResponse({
'message': 'Hello, world!',
'visits': request.session['visits']
})
# Middleware configuration
middleware = [
Middleware(
CORSMiddleware,
allow_origins=['*'],
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*']
),
Middleware(SessionMiddleware, secret_key='your-secret-key'),
Middleware(
TrustedHostMiddleware,
allowed_hosts=['example.com', '*.example.com']
)
]
routes = [
Route("/", endpoint=homepage)
]
app = Starlette(debug=True, routes=routes, middleware=middleware)
Custom Middleware
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.responses import JSONResponse
from starlette.routing import Route
import time
import uuid
# Custom middleware: Request time measurement
class TimingMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
await self.app(scope, receive, send)
return
start_time = time.time()
async def send_wrapper(message):
if message["type"] == "http.response.start":
process_time = time.time() - start_time
headers = dict(message["headers"])
headers[b"x-process-time"] = str(process_time).encode()
message["headers"] = list(headers.items())
await send(message)
await self.app(scope, receive, send_wrapper)
# Request ID middleware
class RequestIDMiddleware:
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
scope["request_id"] = str(uuid.uuid4())
await self.app(scope, receive, send)
async def homepage(request):
request_id = request.scope.get("request_id", "unknown")
return JSONResponse({
'message': 'Hello, world!',
'request_id': request_id
})
middleware = [
Middleware(TimingMiddleware),
Middleware(RequestIDMiddleware)
]
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=middleware
)
WebSocket Support
from starlette.applications import Starlette
from starlette.routing import Route, WebSocketRoute
from starlette.responses import HTMLResponse
from starlette.websockets import WebSocket, WebSocketDisconnect
from typing import List
# Connection management class
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
async def websocket_endpoint(websocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.send_personal_message(f"You wrote: {data}", websocket)
await manager.broadcast(f"Someone says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("Someone left the chat")
async def websocket_user(websocket):
client_id = websocket.path_params["client_id"]
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Client {client_id}: {data}")
except WebSocketDisconnect:
print(f"Client {client_id} disconnected")
async def homepage(request):
return HTMLResponse("""
<!DOCTYPE html>
<html>
<head>
<title>WebSocket Test</title>
</head>
<body>
<h1>WebSocket Test</h1>
<div id="messages"></div>
<input type="text" id="messageText" placeholder="Type a message...">
<button onclick="sendMessage()">Send</button>
<script>
const ws = new WebSocket("ws://localhost:8000/ws");
ws.onmessage = function(event) {
const messages = document.getElementById('messages');
messages.innerHTML += '<div>' + event.data + '</div>';
};
function sendMessage() {
const messageText = document.getElementById('messageText');
ws.send(messageText.value);
messageText.value = '';
}
</script>
</body>
</html>
""")
routes = [
Route("/", endpoint=homepage),
WebSocketRoute("/ws", endpoint=websocket_endpoint),
WebSocketRoute("/ws/{client_id}", endpoint=websocket_user),
]
app = Starlette(debug=True, routes=routes)
Static Files and Templates
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from starlette.responses import HTMLResponse
# Template configuration
templates = Jinja2Templates(directory='templates')
async def homepage(request):
return templates.TemplateResponse(
'index.html',
{'request': request, 'title': 'Starlette App'}
)
async def user_profile(request):
username = request.path_params['username']
user_data = {
'username': username,
'email': f'{username}@example.com',
'created_at': '2025-01-01'
}
return templates.TemplateResponse(
'profile.html',
{'request': request, 'user': user_data}
)
routes = [
Route('/', endpoint=homepage),
Route('/user/{username}', endpoint=user_profile),
Mount('/static', StaticFiles(directory='static'), name='static')
]
app = Starlette(debug=True, routes=routes)
Authentication System
from starlette.applications import Starlette
from starlette.authentication import (
AuthCredentials, AuthenticationBackend, AuthenticationError, SimpleUser
)
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.responses import PlainTextResponse, JSONResponse
from starlette.routing import Route
from starlette.authentication import requires
import base64
import binascii
class BasicAuthBackend(AuthenticationBackend):
async def authenticate(self, conn):
if "Authorization" not in conn.headers:
return
auth = conn.headers["Authorization"]
try:
scheme, credentials = auth.split()
if scheme.lower() != 'basic':
return
decoded = base64.b64decode(credentials).decode("ascii")
except (ValueError, UnicodeDecodeError, binascii.Error):
raise AuthenticationError('Invalid basic auth credentials')
username, _, password = decoded.partition(":")
# In actual implementation, verify user with database
if username == "admin" and password == "secret":
return AuthCredentials(["authenticated"]), SimpleUser(username)
raise AuthenticationError('Invalid credentials')
async def homepage(request):
if request.user.is_authenticated:
return PlainTextResponse(f'Hello, {request.user.display_name}')
return PlainTextResponse('Hello, anonymous user')
@requires("authenticated")
async def protected(request):
return JSONResponse({
"message": f"Hello {request.user.display_name}, this is protected!",
"user": request.user.display_name
})
def on_auth_error(request, exc):
return JSONResponse({"error": str(exc)}, status_code=401)
middleware = [
Middleware(
AuthenticationMiddleware,
backend=BasicAuthBackend(),
on_error=on_auth_error
),
]
routes = [
Route("/", endpoint=homepage),
Route("/protected", endpoint=protected),
]
app = Starlette(routes=routes, middleware=middleware)
Testing
# test_app.py
import pytest
from starlette.testclient import TestClient
from main import app
@pytest.fixture
def client():
return TestClient(app)
def test_homepage(client):
response = client.get("/")
assert response.status_code == 200
assert "Hello" in response.text
def test_json_endpoint(client):
response = client.get("/api")
assert response.status_code == 200
data = response.json()
assert "message" in data
def test_websocket(client):
with client.websocket_connect("/ws") as websocket:
websocket.send_text("Hello WebSocket")
data = websocket.receive_text()
assert "Hello WebSocket" in data
def test_authenticated_endpoint(client):
# Access without authentication
response = client.get("/protected")
assert response.status_code == 401
# Access with Basic authentication
response = client.get(
"/protected",
auth=("admin", "secret")
)
assert response.status_code == 200
def test_static_files(client):
# Static file test (if file exists)
response = client.get("/static/test.txt")
# 200 or 404 depending on file existence
def test_template_response(client):
response = client.get("/user/testuser")
assert response.status_code == 200
assert "testuser" in response.text
# Asynchronous test example
import pytest_asyncio
@pytest_asyncio.async_test
async def test_async_endpoint():
from httpx import AsyncClient
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/slow")
assert response.status_code == 200
# Test execution commands
# pip install pytest httpx pytest-asyncio
# pytest
# pytest -v --cov=main