Add auto-validation, SSE, stream_file, after_request, route groups

Five new features:

1. **Pydantic auto-validation** — if request_model is set, request
   bodies are validated automatically and 422 returned on failure.
   If response_model is set, resp.media is serialized through the
   model (extra fields stripped, types enforced).

2. **Server-Sent Events** — resp.sse for real-time streaming:

       @resp.sse
       async def stream():
           yield {"event": "update", "data": "hello"}

3. **resp.stream_file()** — stream large files without loading
   into memory, with automatic content-type detection.

4. **after_request hooks** — run code after every request:

       @api.after_request()
       def add_request_id(req, resp):
           resp.headers["X-Request-ID"] = str(uuid.uuid4())

5. **Route groups** — organize routes with shared prefixes:

       v1 = api.group("/v1")

       @v1.route("/users")
       def list_users(req, resp): ...

Also fix streaming responses not sending Content-Type headers.

172 tests, 95% coverage.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

responder/api.py

@@ -150,6 +150,23 @@ class API:
 
         return decorator
 
+    def after_request(self):
+        """Register a function to run after every request.
+
+        Usage::
+
+            @api.after_request()
+            def add_request_id(req, resp):
+                resp.headers["X-Request-ID"] = str(uuid.uuid4())
+
+        """
+
+        def decorator(f):
+            self.router.after_request(f)
+            return f
+
+        return decorator
+
     def add_middleware(self, middleware_cls, **middleware_config):
         self.app = middleware_cls(self.app, **middleware_config)
 
@@ -476,5 +493,38 @@ class API:
             kwargs.update({"debug": self.debug})
         self.serve(**kwargs)
 
+    def group(self, prefix):
+        """Create a route group with a shared URL prefix.
+
+        Usage::
+
+            v1 = api.group("/v1")
+
+            @v1.route("/users")
+            def list_users(req, resp):
+                resp.media = []
+
+            @v1.route("/users/{id:int}")
+            def get_user(req, resp, *, id):
+                resp.media = {"id": id}
+
+        """
+        return RouteGroup(api=self, prefix=prefix)
+
     async def __call__(self, scope, receive, send):
         await self.app(scope, receive, send)
+
+
+class RouteGroup:
+    """A group of routes with a shared URL prefix."""
+
+    def __init__(self, api, prefix):
+        self.api = api
+        self.prefix = prefix.rstrip("/")
+
+    def route(self, route=None, **options):
+        full_route = f"{self.prefix}{route}"
+        return self.api.route(full_route, **options)
+
+    def before_request(self, **kwargs):
+        return self.api.before_request(**kwargs)

responder/models.py

@@ -357,6 +357,80 @@ class Response:
 
         return func
 
+    def sse(self, func, *args, **kwargs):
+        """Set up Server-Sent Events streaming.
+
+        Usage::
+
+            @api.route("/events")
+            async def events(req, resp):
+                @resp.sse
+                async def stream():
+                    for i in range(10):
+                        yield {"data": f"message {i}"}
+
+        Each yielded dict can have: data, event, id, retry.
+        Yielding a string is treated as data.
+        """
+        assert inspect.isasyncgenfunction(func)
+
+        async def sse_generator():
+            async for event in func(*args, **kwargs):
+                if isinstance(event, str):
+                    yield f"data: {event}\n\n".encode()
+                elif isinstance(event, dict):
+                    parts = []
+                    if "event" in event:
+                        parts.append(f"event: {event['event']}")
+                    if "id" in event:
+                        parts.append(f"id: {event['id']}")
+                    if "retry" in event:
+                        parts.append(f"retry: {event['retry']}")
+                    data = event.get("data", "")
+                    for line in str(data).split("\n"):
+                        parts.append(f"data: {line}")
+                    parts.append("")
+                    parts.append("")
+                    yield "\n".join(parts).encode()
+                else:
+                    yield f"data: {event}\n\n".encode()
+
+        self._stream = sse_generator
+        self.mimetype = "text/event-stream"
+        self.headers["Cache-Control"] = "no-cache"
+        self.headers["Connection"] = "keep-alive"
+
+        return func
+
+    def stream_file(self, path, *, content_type=None, chunk_size=8192):
+        """Stream a file without loading it entirely into memory.
+
+        :param path: Path to the file.
+        :param content_type: Optional MIME type override.
+        :param chunk_size: Size of chunks to read (default 8192 bytes).
+        """
+        from pathlib import Path as PathType
+
+        path = PathType(path)
+
+        if content_type:
+            self.mimetype = content_type
+        else:
+            import mimetypes
+
+            guessed = mimetypes.guess_type(str(path))[0]
+            self.mimetype = guessed or "application/octet-stream"
+
+        async def file_generator():
+            with open(path, "rb") as f:
+                while True:
+                    chunk = f.read(chunk_size)
+                    if not chunk:
+                        break
+                    yield chunk
+
+        self._stream = file_generator
+
     def file(self, path, *, content_type=None):
         """Serve a file from disk as the response.
 
@@ -385,7 +459,10 @@ class Response:
     @property
     async def body(self):
         if self._stream is not None:
-            return (self._stream(), {})
+            headers = {}
+            if self.mimetype is not None:
+                headers["Content-Type"] = self.mimetype
+            return (self._stream(), headers)
 
         if self.content is not None:
             headers = {}

responder/routes.py

@@ -123,6 +123,23 @@ class Route(BaseRoute):
                 await response(scope, receive, send)
                 return
 
+        # Auto-validate request body with Pydantic model
+        req_model = getattr(self.endpoint, "_request_model", None)
+        if req_model is not None and request.method in ("post", "put", "patch"):
+            try:
+                body = await request.media()
+                req_model(**body)
+            except Exception as exc:
+                response.status_code = 422
+                errors = []
+                if hasattr(exc, "errors"):
+                    errors = exc.errors()
+                else:
+                    errors = [{"msg": str(exc)}]
+                response.media = {"errors": errors}
+                await response(scope, receive, send)
+                return
+
         views = []
 
         if inspect.isclass(self.endpoint):
@@ -150,6 +167,23 @@ class Route(BaseRoute):
             else:
                 await run_in_threadpool(view, request, response, **path_params)
 
+        # Auto-serialize response with Pydantic model
+        resp_model = getattr(self.endpoint, "_response_model", None)
+        if resp_model is not None and response.media is not None:
+            try:
+                validated = resp_model(**response.media)
+                response.media = validated.model_dump()
+            except Exception:
+                pass  # Don't break the response if serialization fails
+
+        # Run after-request hooks
+        after_requests = scope.get("after_requests", [])
+        for after_request in after_requests:
+            if asyncio.iscoroutinefunction(after_request):
+                await after_request(request, response)
+            else:
+                await run_in_threadpool(after_request, request, response)
+
         if response.status_code is None:
             response.status_code = status_codes.HTTP_200  # type: ignore[attr-defined]
 
@@ -231,6 +265,7 @@ class Router:
         self.before_requests = (
             {"http": [], "ws": []} if before_requests is None else before_requests
         )
+        self.after_requests: list = []
         self.events = defaultdict(list)
         self._lifespan_handler = lifespan
 
@@ -297,6 +332,9 @@ class Router:
         else:
             self.before_requests.setdefault("http", []).append(endpoint)
 
+    def after_request(self, endpoint):
+        self.after_requests.append(endpoint)
+
     def url_for(self, endpoint, **params):
         for route in self.routes:
             if endpoint in (route.endpoint, route.endpoint.__name__):
@@ -371,6 +409,7 @@ class Router:
         route = self._resolve_route(scope)
 
         scope["before_requests"] = self.before_requests
+        scope["after_requests"] = self.after_requests
 
         if route is not None:
             await route(scope, receive, send)

tests/test_new_features.py

@@ -0,0 +1,215 @@
+"""Tests for new features: validation, SSE, after_request, route groups, stream_file."""
+
+import pytest
+from pydantic import BaseModel
+from starlette.testclient import TestClient as StarletteTestClient
+
+import responder
+
+
+# --- Pydantic auto-validation ---
+
+
+class ItemIn(BaseModel):
+    name: str
+    price: float
+
+
+class ItemOut(BaseModel):
+    id: int
+    name: str
+    price: float
+
+
+def test_pydantic_request_validation():
+    """Auto-validate request body against request_model."""
+    api = responder.API(allowed_hosts=[";"])
+
+    @api.route("/items", methods=["POST"], request_model=ItemIn)
+    async def create(req, resp):
+        data = await req.media()
+        resp.media = {"id": 1, **data}
+
+    # Valid request
+    r = api.requests.post("http://;/items", json={"name": "widget", "price": 9.99})
+    assert r.status_code == 200
+    assert r.json()["name"] == "widget"
+
+    # Invalid request — missing required field
+    r = api.requests.post("http://;/items", json={"name": "widget"})
+    assert r.status_code == 422
+    assert "errors" in r.json()
+
+    # Invalid request — wrong type
+    r = api.requests.post("http://;/items", json={"name": "widget", "price": "not_a_number"})
+    assert r.status_code == 422
+
+
+def test_pydantic_response_serialization():
+    """Auto-serialize response through response_model."""
+    api = responder.API(allowed_hosts=[";"])
+
+    @api.route("/items", methods=["POST"],
+               request_model=ItemIn, response_model=ItemOut)
+    async def create(req, resp):
+        data = await req.media()
+        # Include an extra field that should be stripped by the model
+        resp.media = {"id": 1, "secret": "hidden", **data}
+
+    r = api.requests.post("http://;/items", json={"name": "widget", "price": 9.99})
+    assert r.status_code == 200
+    data = r.json()
+    assert data == {"id": 1, "name": "widget", "price": 9.99}
+    assert "secret" not in data
+
+
+def test_pydantic_validation_skipped_for_get():
+    """GET requests don't trigger request body validation."""
+    api = responder.API(allowed_hosts=[";"])
+
+    @api.route("/items", methods=["GET"], request_model=ItemIn)
+    def list_items(req, resp):
+        resp.media = []
+
+    r = api.requests.get("http://;/items")
+    assert r.status_code == 200
+
+
+# --- SSE streaming ---
+
+
+def test_sse_streaming(api):
+    """Server-Sent Events with resp.sse."""
+
+    @api.route("/events")
+    async def events(req, resp):
+        @resp.sse
+        async def stream():
+            yield {"data": "hello"}
+            yield {"event": "update", "data": "world"}
+            yield "simple"
+
+    r = api.requests.get(api.url_for(events))
+    assert r.status_code == 200
+    assert "text/event-stream" in r.headers.get("content-type", "")
+    assert "data: hello" in r.text
+    assert "event: update" in r.text
+    assert "data: world" in r.text
+    assert "data: simple" in r.text
+
+
+def test_sse_with_id_and_retry(api):
+    """SSE events with id and retry fields."""
+
+    @api.route("/events")
+    async def events(req, resp):
+        @resp.sse
+        async def stream():
+            yield {"data": "msg", "id": "1", "retry": "5000"}
+
+    r = api.requests.get(api.url_for(events))
+    assert "id: 1" in r.text
+    assert "retry: 5000" in r.text
+
+
+# --- stream_file ---
+
+
+def test_stream_file(api, tmp_path):
+    """Stream a file without loading into memory."""
+    big_file = tmp_path / "data.bin"
+    big_file.write_bytes(b"x" * 10000)
+
+    @api.route("/download")
+    def download(req, resp):
+        resp.stream_file(big_file)
+
+    r = api.requests.get(api.url_for(download))
+    assert len(r.content) == 10000
+    assert r.content == b"x" * 10000
+
+
+def test_stream_file_content_type(api, tmp_path):
+    """stream_file detects content type."""
+    css = tmp_path / "style.css"
+    css.write_text("body { color: red; }")
+
+    @api.route("/css")
+    def serve_css(req, resp):
+        resp.stream_file(css)
+
+    r = api.requests.get(api.url_for(serve_css))
+    assert "text/css" in r.headers.get("content-type", "")
+
+
+# --- after_request hooks ---
+
+
+def test_after_request(api):
+    """after_request hook runs after route handler."""
+
+    @api.after_request()
+    def add_header(req, resp):
+        resp.headers["X-After"] = "yes"
+
+    @api.route("/")
+    def view(req, resp):
+        resp.text = "hello"
+
+    r = api.requests.get(api.url_for(view))
+    assert r.text == "hello"
+    assert r.headers["X-After"] == "yes"
+
+
+def test_after_request_async(api):
+    """Async after_request hook."""
+
+    @api.after_request()
+    async def add_header(req, resp):
+        resp.headers["X-Async-After"] = "yes"
+
+    @api.route("/")
+    def view(req, resp):
+        resp.text = "hello"
+
+    r = api.requests.get(api.url_for(view))
+    assert r.headers["X-Async-After"] == "yes"
+
+
+# --- Route groups ---
+
+
+def test_route_group(api):
+    """Route group with shared prefix."""
+    v1 = api.group("/v1")
+
+    @v1.route("/users")
+    def list_users(req, resp):
+        resp.media = [{"name": "alice"}]
+
+    @v1.route("/users/{user_id:int}")
+    def get_user(req, resp, *, user_id):
+        resp.media = {"id": user_id}
+
+    r = api.requests.get("http://;/v1/users")
+    assert r.json() == [{"name": "alice"}]
+
+    r = api.requests.get("http://;/v1/users/42")
+    assert r.json() == {"id": 42}
+
+
+def test_multiple_route_groups(api):
+    """Multiple route groups coexist."""
+    v1 = api.group("/v1")
+    v2 = api.group("/v2")
+
+    @v1.route("/status")
+    def v1_status(req, resp):
+        resp.media = {"version": 1}
+
+    @v2.route("/status")
+    def v2_status(req, resp):
+        resp.media = {"version": 2}
+
+    assert api.requests.get("http://;/v1/status").json() == {"version": 1}
+    assert api.requests.get("http://;/v2/status").json() == {"version": 2}