Starlette

Lightweight ASGI framework and toolkit with WebSocket and GraphQL support. Foundation technology for FastAPI.

PythonframeworkASGIlightweightasynchronousmiddlewareWebSocket

GitHub Overview

encode/starlette

The little ASGI framework that shines. 🌟

Stars11,239
Watchers103
Forks1,020
Created:June 25, 2018
Language:Python
License:BSD 3-Clause "New" or "Revised" License

Topics

asynchttppythonwebsockets

Star History

encode/starlette Star History
Data as of: 7/17/2025, 10:32 AM

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

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