Compare commits

...

2 Commits

Author SHA1 Message Date
kennethreitz 1c5dc1dfbb Replace HTTP/2 server push with dependency injection in backlog
HTTP/2 server push was removed from the spec and dropped by browsers.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-24 16:02:15 -04:00
kennethreitz cb4bc295b8 Docs: second improvement pass
- tutorial-auth: fix deprecated datetime.utcnow() → datetime.now(timezone.utc),
  add role-based access control section, add auth strategy comparison
- tutorial-websockets: use WebSocketDisconnect instead of bare Exception,
  add connection lifecycle section, add rejected connection test example
- tutorial-sqlalchemy: modernize to mapped_column() / Mapped[] (SQLAlchemy 2.0)
- deployment: use uv in Docker example, fix stale uv.lock reference
- quickstart: link to all tutorials in "next steps"
- sandbox: rewrite with project layout, mypy, and pattern-matching test examples

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-24 15:58:48 -04:00
7 changed files with 173 additions and 37 deletions
+1 -1
View File
@@ -5,4 +5,4 @@
- Per-route rate limiting (different limits for different endpoints)
- Built-in structured logging with request context
- OpenAPI 3.1 support
- HTTP/2 server push
- Dependency injection for route handlers
+7 -5
View File
@@ -30,8 +30,9 @@ Here's a minimal Dockerfile::
FROM python:3.13-slim
WORKDIR /app
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
COPY . .
RUN pip install responder
RUN uv pip install --system responder
ENV PORT=80
EXPOSE 80
CMD ["python", "api.py"]
@@ -42,9 +43,10 @@ Build and run::
$ docker run -p 8000:80 myapi
The ``python:3.13-slim`` image is about 150MB — small enough for fast
deploys but includes everything you need. For even smaller images, you
can use ``python:3.13-alpine``, though some packages may need extra
build dependencies.
deploys but includes everything you need. Using ``uv`` for installs
is significantly faster than pip. For even smaller images, you can use
``python:3.13-alpine``, though some packages may need extra build
dependencies.
Cloud Platforms
@@ -182,4 +184,4 @@ Before going live:
- **Add a health check**``/health`` endpoint for monitoring
- **Enable HTTPS** — via your proxy, cloud platform, or uvicorn's ``--ssl-*`` flags
- **Set up logging** — uvicorn logs requests by default; pipe them to your log aggregator
- **Pin your dependencies**commit ``uv.lock`` for reproducible deploys
- **Pin your dependencies**use a lock file or pinned requirements for reproducible deploys
+6
View File
@@ -376,3 +376,9 @@ jump into the tutorials:
- :doc:`tutorial-rest` — build a full CRUD API with validation
- :doc:`tutorial-sqlalchemy` — connect to a database
- :doc:`tutorial-auth` — add authentication
- :doc:`tutorial-websockets` — real-time communication
- :doc:`tutorial-middleware` — hooks and middleware
- :doc:`tutorial-flask` — migrating from Flask
- :doc:`guide-config` — environment variables and secrets
- :doc:`deployment` — Docker, cloud platforms, and production
- :doc:`testing` — writing tests with pytest
+43 -17
View File
@@ -2,35 +2,61 @@
# Development Sandbox
## Setup
Set up a development sandbox.
Acquire sources and create virtualenv.
Clone the repo and install all dependencies:
```shell
git clone https://github.com/kennethreitz/responder.git
cd responder
uv venv
```
Install project in editable mode, including
all development tools.
```shell
uv venv && source .venv/bin/activate
uv pip install --upgrade --editable '.[develop,docs,release,test]'
```
## Operations
Run tests.
## Running Tests
```shell
source .venv/bin/activate
pytest
pytest # full suite with coverage
pytest tests/test_responder.py -xvs # single file, stop on first failure
pytest -k "test_mount" # run tests matching a pattern
```
Format code.
## Code Formatting
```shell
ruff format .
ruff check --fix .
ruff format . # auto-format
ruff check --fix . # lint and auto-fix
```
Documentation authoring.
## Type Checking
```shell
sphinx-autobuild --open-browser --watch docs/source docs/source docs/build
mypy
```
## Documentation
Live-reloading doc server (opens in browser):
```shell
cd docs
sphinx-autobuild --open-browser --watch source source build
```
Or build once:
```shell
cd docs
make html
# open build/html/index.html
```
## Project Layout
```
responder/
├── responder/ # main package
│ ├── api.py # API class — the entry point
│ ├── routes.py # Router, Route, WebSocketRoute
│ ├── models.py # Request and Response wrappers
│ ├── ext/ # extensions (CLI, GraphQL, OpenAPI, rate limiting)
│ ├── background.py # background task queue
│ └── formats.py # content negotiation (JSON, YAML, msgpack)
├── tests/ # pytest test suite
├── examples/ # runnable example apps
├── docs/source/ # Sphinx documentation
└── pyproject.toml # project metadata and tool config
```
+55 -2
View File
@@ -46,14 +46,14 @@ Install PyJWT::
Create a helper to encode and decode tokens::
import jwt
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
SECRET = "your-secret-key"
def create_token(user_id: int) -> str:
payload = {
"sub": user_id,
"exp": datetime.utcnow() + timedelta(hours=24),
"exp": datetime.now(timezone.utc) + timedelta(hours=24),
}
return jwt.encode(payload, SECRET, algorithm="HS256")
@@ -189,3 +189,56 @@ Remember to set a proper secret key::
The session data is signed (not encrypted) — users can read it but
can't tamper with it. Don't store sensitive data like passwords in
sessions.
Role-Based Access Control
--------------------------
For APIs where different users have different permissions, embed the
role in the token and check it in route-specific guards::
def create_token(user_id: int, role: str = "user") -> str:
payload = {
"sub": user_id,
"role": role,
"exp": datetime.now(timezone.utc) + timedelta(hours=24),
}
return jwt.encode(payload, SECRET, algorithm="HS256")
Create a helper that checks for a specific role::
def require_role(*roles):
"""Before-request hook factory that restricts by role."""
def check(req, resp):
user_role = getattr(req.state, "role", None)
if user_role not in roles:
resp.status_code = 403
resp.media = {"error": "Insufficient permissions"}
return check
Use it on specific routes::
@api.route("/admin/users", before_request=require_role("admin"))
def list_all_users(req, resp):
resp.media = {"users": [...]}
And store the role during token verification::
# In your auth_guard:
req.state.user_id = payload["sub"]
req.state.role = payload.get("role", "user")
Choosing an Auth Strategy
--------------------------
- **API keys** — simplest. Good for server-to-server, CLI tools, and
internal services. No expiration unless you build it.
- **JWT tokens** — standard for SPAs and mobile apps. Stateless, so
they scale well. Downside: you can't revoke them without a blocklist.
- **Sessions** — best for traditional web apps with HTML forms. The
browser manages cookies automatically. Stateful — the server controls
the session lifecycle.
Start with API keys for internal tools, JWT for public APIs, and
sessions for web apps with login pages.
+11 -10
View File
@@ -28,8 +28,8 @@ SQLAlchemy models map Python classes to database tables. Each attribute
becomes a column::
# models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
@@ -37,15 +37,16 @@ becomes a column::
class Book(Base):
__tablename__ = "books"
id = Column(Integer, primary_key=True, autoincrement=True)
title = Column(String, nullable=False)
author = Column(String, nullable=False)
year = Column(Integer, nullable=False)
isbn = Column(String, nullable=True)
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column(String, nullable=False)
author: Mapped[str] = mapped_column(String, nullable=False)
year: Mapped[int] = mapped_column(nullable=False)
isbn: Mapped[str | None] = mapped_column(String, nullable=True)
``DeclarativeBase`` is SQLAlchemy's modern base class (SQLAlchemy 2.0+).
Each model class corresponds to a table, and each ``Column`` corresponds
to a column in that table.
This uses SQLAlchemy 2.0's ``Mapped`` type annotations and
``mapped_column()``, which give you type checker support and cleaner
syntax than the older ``Column()`` style. Each model class corresponds
to a table, and each ``mapped_column()`` corresponds to a column.
Database Setup
+50 -2
View File
@@ -58,6 +58,8 @@ A chat room needs to broadcast messages to all connected clients. We keep
a set of active connections and iterate through them when someone sends
a message::
from starlette.websockets import WebSocketDisconnect
connected = set()
@api.route("/chat", websocket=True)
@@ -70,13 +72,15 @@ a message::
# Broadcast to all connected clients
for client in connected:
await client.send_text(message)
except Exception:
except WebSocketDisconnect:
pass
finally:
connected.discard(ws)
The ``try/finally`` block ensures we remove disconnected clients from
the set, even if the connection drops unexpectedly.
the set, even if the connection drops unexpectedly. Catching
``WebSocketDisconnect`` specifically (rather than bare ``Exception``)
makes the intent clear and avoids swallowing real bugs.
Data Formats
@@ -154,6 +158,40 @@ WebSocket before-request hooks receive the ``ws`` object and must call
``await ws.accept()`` if they want the connection to proceed.
Connection Lifecycle
--------------------
WebSocket connections go through several states:
1. **Connecting** — the client sends an upgrade request
2. **Open** — after ``await ws.accept()``, both sides can send messages
3. **Closing** — either side initiates a close handshake
4. **Closed** — the connection is fully terminated
When a client disconnects (closes the tab, loses network), the next
``await ws.receive_text()`` raises ``WebSocketDisconnect``. Always
handle this — otherwise your server accumulates dead connections::
from starlette.websockets import WebSocketDisconnect
@api.route("/ws", websocket=True)
async def handler(ws):
await ws.accept()
try:
while True:
data = await ws.receive_text()
await ws.send_text(f"Got: {data}")
except WebSocketDisconnect:
print("Client disconnected")
You can also close connections from the server side::
await ws.close(code=1000) # 1000 = normal closure
Common close codes: ``1000`` (normal), ``1001`` (going away),
``1008`` (policy violation), ``1011`` (server error).
Testing WebSockets
------------------
@@ -169,3 +207,13 @@ Use Starlette's ``TestClient`` for WebSocket tests::
The ``websocket_connect`` context manager handles the connection
lifecycle — it connects on enter and disconnects on exit.
You can also test that connections are properly rejected::
from starlette.websockets import WebSocketDisconnect
def test_websocket_404():
client = TestClient(api)
with pytest.raises(WebSocketDisconnect):
with client.websocket_connect("/nonexistent"):
pass