Docs plain plain views README.md

Views

Take a request, return a response.

Overview
HTTP methods map to class methods
Template-rendering views
RedirectView
Async views
ServerSentEventsView
WebSockets
Lifecycle hooks
ResponseException
Error views
View patterns
FAQs
Installation

Overview

Plain views are class-based, with a straightforward API that keeps simple views simple while giving you the full power of a class for complex cases.

from plain.http import Response
from plain.views import View


class ExampleView(View):
    def get(self):
        return Response("<html><body>Hello, world!</body></html>")

View handlers return a Response (or subclass like JsonResponse, RedirectResponse, etc.). To return a dict/list shorthand for JSON APIs, use APIView.

HTTP methods map to class methods

The HTTP method of the request maps directly to a class method of the same name. Define only the methods you want to support.

from plain.views import View


class ExampleView(View):
    def get(self):
        pass

    def post(self):
        pass

    def put(self):
        pass

    def patch(self):
        pass

    def delete(self):
        pass

If a request comes in for a method your view doesn't implement, Plain returns a 405 Method Not Allowed response automatically.

The base View class provides default options and head behavior, but you can override these too.

Template-rendering views

Views that render Jinja templates (TemplateView, FormView, DetailView, CreateView, UpdateView, DeleteView, ListView) live in the plain.templates package. Install plain.templates and import them from plain.templates.views:

from plain.templates.views import TemplateView


class ExampleView(TemplateView):
    template_name = "example.html"

    def get_template_context(self):
        context = super().get_template_context()
        context["message"] = "Hello, world!"
        return context

RedirectView

RedirectView redirects to another URL.

from plain.views import RedirectView


class ExampleRedirectView(RedirectView):
    url = "/new-location/"

Set status_code = 301 for permanent redirects (default is 302).

For simple redirects, configure the view directly in your URL routes.

from plain.views import RedirectView
from plain.urls import path, Router


class AppRouter(Router):
    routes = [
        path(
            "/old-location/",
            RedirectView.as_view(url="/new-location/", status_code=301),
        ),
    ]

You can also redirect to a named URL using url_name, or preserve query parameters with preserve_query_params=True.

Async views

Any view method defined with async def runs directly on the worker's event loop. This enables non-blocking I/O patterns like SSE and async HTTP clients.

Important: Blocking calls in async views freeze the entire worker process — no other requests can be processed until the blocking call returns. Plain's ORM, sessions, and auth layers are all synchronous and must not be called directly from async views.

Common mistakes:

User.query.get(pk=1) — blocks the event loop
time.sleep(1) — use await asyncio.sleep(1) instead
requests.get(...) — use an async HTTP client instead

To wrap a blocking call safely: await asyncio.to_thread(blocking_fn). Use asyncio.to_thread rather than loop.run_in_executor(None, blocking_fn) — it copies the request context, so context-local state like the per-request database connection follows the call into the thread.

Use async views only for true async I/O (SSE, async HTTP clients). For standard request/response views that use the ORM, use regular sync views — they run in the thread pool and don't block other connections.

In development (DEBUG=True), the server enables asyncio debug mode which logs warnings when a callback blocks the event loop for more than 100ms.

Type checkers will flag async def get as an LSP violation against the sync View.get base stub. Add # ty: ignore[invalid-method-override] on the async handler line — the dispatch layer handles sync and async uniformly at runtime.

ServerSentEventsView

ServerSentEventsView provides Server-Sent Events streaming. Subclass it and implement stream() as an async generator.

import asyncio
from datetime import datetime
from plain.views import ServerSentEvent, ServerSentEventsView


class ClockView(ServerSentEventsView):
    async def stream(self):
        while True:
            yield ServerSentEvent(data={"time": datetime.now().isoformat()})
            await asyncio.sleep(1)

The stream() method must yield ServerSentEvent instances. The data argument accepts strings, dicts, and lists (dicts/lists are JSON-serialized). You can also set optional event, id, and retry fields.

from plain.views import ServerSentEvent


class NotificationView(ServerSentEventsView):
    async def stream(self):
        yield ServerSentEvent(data="hello")  # Simple string
        yield ServerSentEvent(data={"count": 1})  # JSON data
        yield ServerSentEvent(data="update", event="status")  # Named event type
        yield ServerSentEvent(data="msg", id="42", retry=5000)  # With id and retry
        yield ServerSentEvent.comment("keepalive")  # SSE comment (keepalive)

Connect from JavaScript using the standard EventSource API:

const source = new EventSource("/events/");
source.onmessage = (event) => {
    console.log(event.data);
};
// Listen for named event types
source.addEventListener("status", (event) => {
    console.log("Status:", event.data);
});

Send ServerSentEvent.comment() periodically as a keepalive to prevent proxies and browsers from closing idle connections.

ServerSentEventsView only accepts GET requests. The stream() method runs on the event loop — use await for any I/O and avoid blocking calls. Use await asyncio.sleep() instead of time.sleep(), and await asyncio.to_thread() to wrap blocking operations.

Note: browsers limit HTTP/1.1 to 6 SSE connections per domain. Use HTTP/2 to avoid this limit.

WebSockets

A view handles a WebSocket by defining async def websocket(self, ws). The same URL keeps serving its page from get() — the browser opens the socket against the URL it is already on — and the handshake runs through the normal pipeline (middleware, before_request, URL kwargs), so the socket is authenticated exactly like the page is.

from plain.http import WebSocket
from plain.templates.views import TemplateView


class LiveView(TemplateView):
    template_name = "live.html"

    async def websocket(self, ws: WebSocket) -> None:
        async for message in ws:
            await ws.send(f"echo: {message}")

const socket = new WebSocket(`wss://${location.host}${location.pathname}`);
socket.addEventListener("message", (event) => console.log(event.data));

A plain GET to LiveView renders the template as usual. A view that defines only websocket() answers a plain GET with 405 — websocket is not an HTTP method and never appears in Allow — and an upgrade request to a view that doesn't define websocket() is served as an ordinary GET.

Only a complete RFC 6455 opening handshake is treated as an upgrade: a GET with Upgrade: websocket, a Connection header carrying the upgrade token, Sec-WebSocket-Version: 13, and a Sec-WebSocket-Key that decodes to 16 bytes. Anything short of that — including a different version — is served as a normal GET.

The ws object (WebSocket) is the entire surface a view gets:

async for message in ws — complete messages, str for text frames and bytes for binary ones. Fragments are reassembled and control frames (ping, pong, close) are answered for you.
await ws.send(message) — a str goes out as a text frame, bytes as a binary one. Outbound messages have no size limit.
await ws.close(code=1000, reason="") — send the close frame, wait for the peer's, then drop the connection. It's idempotent, and safe to call from another task while something else is iterating.
ws.subprotocol — the negotiated subprotocol, or None.
ws.closed — whether the socket is over; ws.close_code and ws.close_reason say how (1006 when the peer vanished without a close frame).

Returning from websocket() closes the socket.

A peer going away is a normal ending, not an error. The most common way a socket ends is the browser leaving. When that happens the iteration simply stops, and ws.send() raises WebSocketClosed (carrying .code and .reason) — a raw socket error never reaches the view. The server logs nothing for either.

import asyncio
from datetime import datetime

from plain.http import WebSocket, WebSocketClosed
from plain.views import View


class ClockView(View):
    async def websocket(self, ws: WebSocket) -> None:
        try:
            while True:
                await ws.send(datetime.now().isoformat())
                await asyncio.sleep(1)
        except WebSocketClosed:
            return  # The browser left — nothing to report.

Any other exception out of websocket() is a real failure: it's logged like a 500 and the socket is closed with 1011.

Subprotocols are declared on the view and matched case-sensitively against what the client offers, in the client's preference order. The first offered name the view supports is echoed back in Sec-WebSocket-Protocol and exposed as ws.subprotocol; if nothing matches, no subprotocol is echoed (a browser then fails the connection itself, per spec).

class LiveView(TemplateView):
    template_name = "live.html"
    websocket_subprotocols = ("binary",)

Inbound messages are capped by websocket_max_message_size (1 MiB by default). A frame that declares more than the cap is refused before its payload is read, and a reassembled message over the cap closes the socket with 1009. The cap applies to what the client sends, not to what the view sends.

class LiveView(TemplateView):
    template_name = "live.html"
    websocket_max_message_size = 16 * 1024 * 1024

Auth works through the normal hooks. before_request runs before the upgrade is accepted, so an AuthViewMixin-style check rejects the handshake with its usual 403 or redirect and websocket() never runs.

Session cookies ride along on a websocket handshake the same way they do on a POST, so a cross-origin handshake is refused with 403 by the same decision the CSRF middleware makes: CSRF_TRUSTED_ORIGINS first, then Sec-Fetch-Site (browsers send same-origin on a same-origin handshake), then Origin against Host for requests with no fetch metadata. A request with neither Origin nor Sec-Fetch-Site — a non-browser client — is allowed. To accept handshakes from another origin, add it to CSRF_TRUSTED_ORIGINS; there is no websocket-specific setting.

Middleware sees the handshake as a response. The accepted upgrade is a bodiless 101 that flows through every middleware's after_response, so cookies and headers set there land on it, and resource closers registered on it run when the socket ends.

Keepalive is automatic. The server sends a ping every 20 seconds and closes the socket with 1001 if no pong comes back within 20 seconds of it. A send that the peer does not drain within 20 seconds ends the socket the same way. Nothing to configure, and it keeps proxies that drop idle connections from cutting a quiet socket.

A view that stops reading gets closed. Inbound messages are buffered for a view that is not iterating (16 of them), and once that buffer is full nothing more is read from the peer, pongs included. If the peer keeps sending to a view that never comes back to async for, the socket is closed with 1008 at the next ping. A push-only view whose peer never sends is unaffected.

When the socket ends, the view ends. If the peer leaves or keepalive fails while websocket() is awaiting something other than ws, the coroutine is cancelled a moment later; a view that returns to ws first sees the iteration end or send raise.

Worker shutdown is cooperative. When a worker starts draining — SIGTERM, a deploy, or recycling at SERVER_MAX_REQUESTS — every open socket is sent a 1001 close and its websocket() coroutine is cancelled, rather than being cut off mid-frame at the end of the graceful timeout. Clients should reconnect when they see 1001; a draining worker refuses new upgrades with a 503.

Databases. The connection the handshake used is returned to the pool before the socket starts, so a socket that only pumps bytes holds no database connection however long it stays open. The first query inside websocket() checks a connection out again, and that one is held until the socket ends.

The ORM is synchronous, so a query inside websocket() blocks the event loop for every connection on that worker. Move it to a thread with await asyncio.to_thread(fn), which copies the request context — loop.run_in_executor(None, fn) does not, and a query run without it escapes the per-request connection handling entirely.

Testing uses Client().websocket(), which runs the handshake through the same pipeline as any test-client request and then drives the view's coroutine in-process — see WebSockets in the test docs.

Lifecycle hooks

Every view has three hooks around its handler: before_request runs before the HTTP method handler, after_response runs after the response is built (including error responses), and handle_exception converts any exception raised during dispatch into a response.

`before_request`

Runs before the HTTP method handler (get, post, etc.). Default is a no-op. Raise to reject the request — the exception flows through handle_exception.

from plain.http import ForbiddenError403


class MyView(View):
    def before_request(self):
        if self.request.user and self.request.user.is_banned:
            raise ForbiddenError403("Banned")

Use it for auth checks, rate limiting, or any precondition. AuthView overrides it to call check_auth(); APIKeyView uses it to validate the API key.

`after_response`

Runs after the response is built — for successes, responses from handle_exception, and 405 method-not-allowed. Return the response (mutated or replaced). Default is a no-op.

from plain.http import Response
from plain.utils.cache import patch_cache_control


class MyView(View):
    def after_response(self, response: Response) -> Response:
        patch_cache_control(response, private=True)
        return response

Exceptions raised inside after_response are not routed through handle_exception — they escape to the framework's error renderer. Guard in before_request or inside the handler for anything that might raise.

`handle_exception`

Converts an exception raised during before_request or the handler into a response. Subclasses override it to format errors for their clients.

from plain.http import Response


class MyView(View):
    def handle_exception(self, exc: Exception) -> Response:
        if isinstance(exc, MyAppError):
            return JsonResponse({"error": str(exc)}, status_code=400)
        return super().handle_exception(exc)

The base View.handle_exception re-raises, so unhandled cases fall through to the framework default — which returns a plain-text status line (404 Not Found, 500 Internal Server Error, etc.). View subclasses that want a richer format override the hook: TemplateView renders {status}.html, APIView emits JSON.

ResponseException is unwrapped by get_response before handle_exception runs, so you don't need to handle it in overrides. Returning a 5xx response is treated as a real failure: the framework logs the exception once (via log_exception) and attaches it to response.exception so observability tooling can record it. Returning a 4xx response is silent — the view chose to render it as a handled outcome.

Plain's HTTP exceptions (NotFoundError404, ForbiddenError403, BadRequestError400, etc.) inherit HTTPException and carry their own status_code. Subclass HTTPException to define your own:

from plain.http import HTTPException


class PaymentRequiredError402(HTTPException):
    status_code = 402

ResponseException

At any point during request handling, you can raise a ResponseException to immediately return a response. This is useful for authorization checks or rate limiting in nested helper functions.

from plain.views import View
from plain.views.exceptions import ResponseException
from plain.http import Response


class ExampleView(View):
    def get(self):
        if self.request.user and self.request.user.exceeds_rate_limit:
            raise ResponseException(Response("Rate limit exceeded", status_code=429))

        return Response("ok")

Error views

Plain core's exception handler returns plain text — 404 Not Found, 500 Internal Server Error, etc. Styled error pages live in plain.templates: TemplateView.handle_exception looks for {status_code}.html and renders it with request, status_code, exception, and DEBUG in context, falling back to plain text on TemplateFileMissing.

To get a styled 404 for URL-resolution failures (where no view runs), mount NotFoundView as the last route:

from plain.templates.views import NotFoundView
from plain.urls import path

urls = (
    # ... your routes ...
    path("<path:_>", NotFoundView),
)

The resolver treats a sole-segment terminal <path:> as a catchall: slash-agnostic, and yields to slash-mismatch redirects from specific routes. So you still get /login → 308 → /login/ for path("login/", LoginView), not a stray 404 from the catchall.

View patterns

Don't evaluate querysets at class level

Class attributes execute at import time, not per request. Queries belong in methods.

# Bad — runs once at import, stale forever
class DashboardView(View):
    recent_users = User.query.order_by("-created_at")[:5]


# Good — fresh per request
class DashboardView(View):
    def get_template_context(self):
        return {"recent_users": User.query.order_by("-created_at")[:5]}

Paginate list views

Always paginate querysets in list views. Unbounded queries get slower as data grows.

from plain.paginator import Paginator


def get_template_context(self):
    paginator = Paginator(Item.query.all(), per_page=25)
    page = paginator.get_page(self.request.query_params.get("page"))
    return {"page": page}

Wrap multi-step writes in transactions

Use transaction.atomic() when creating or updating related objects together.

from plain.postgres import transaction

with transaction.atomic():
    order = Order(user=user)
    order.create()
    payment = Payment(order=order, amount=total)
    payment.create()

FAQs

How do I exempt a view from CSRF protection?

Use the CSRF_EXEMPT_PATHS setting to specify path patterns that should bypass CSRF protection. For example:

# app/settings.py
CSRF_EXEMPT_PATHS = [
    r"^/api/",  # Exempt all API routes
    r"^/webhooks/",  # Exempt webhook endpoints
]

How do I access URL parameters?

URL parameters are available via self.url_kwargs.

class ExampleView(View):
    def get(self):
        user_id = self.url_kwargs["id"]
        return f"User ID: {user_id}"

How do I access the request object?

The request is available as self.request after the view is set up.

class ExampleView(View):
    def get(self):
        return f"Path: {self.request.path}"

Can I customize view initialization?

Yes, define your own __init__ method to accept custom arguments passed via as_view().

class CustomView(View):
    def __init__(self, feature_enabled=False):
        self.feature_enabled = feature_enabled


# In URLs
path("/custom/", CustomView.as_view(feature_enabled=True))

Installation

Views are included with the core plain package. No additional installation is required.

On this page