HTTP — Requests, Responses & Views

openviper.http.request

class Request(scope, receive)

Wraps an ASGI scope and receive callable. All body-reading methods are coroutines.

Properties (synchronous):

method -> str

The HTTP method in upper-case (e.g. "GET", "POST").

path -> str

The request URL path (e.g. "/users/42").

root_path -> str

The ASGI root_path (application mount prefix).

url -> URL

Full URL object with scheme, netloc, path, query.

query_params -> QueryParams

Parsed query string as a multi-dict. Supports .get(), .getlist(), and in tests.

headers -> Headers

Case-insensitive, immutable header map. Access like a dict: request.headers["content-type"].

cookies -> dict[str, str]

Parsed Cookie header.

path_params -> dict[str, Any]

Path parameters captured by the router (e.g. {"id": 42}).

client -> tuple[str, int] | None

(ip, port) of the connected client, or None for UNIX sockets.

state -> dict[str, Any]

Per-request mutable storage for middleware to attach data.

user

The authenticated user attached by AuthenticationMiddleware. An AnonymousUser when unauthenticated.

auth

Auth info dict attached by AuthenticationMiddleware (e.g. {"type": "jwt", "claims": {...}}).

Raw header lookup:

header(name: bytes) → bytes | None

O(1) raw header lookup. name must be lower-cased bytes (e.g. b"content-type").

Body reading (all coroutines):

body() → Awaitable[bytes]

Read and cache the full request body. Limited to 10 MB by default. Raises ValueError when Content-Length is exceeded.

json() → Awaitable[Any]

Parse the body as JSON.

form() → Awaitable[ImmutableMultiDict]

Parse application/x-www-form-urlencoded or multipart/form-data. Returns both regular fields and UploadFile objects in the same dict-like structure.

class UploadFile(filename, content_type, file)

Represents an uploaded file from a multipart form submission.

filename -> str

content_type -> str

read(size=-1) → Awaitable[bytes]

Read bytes from the underlying file off-thread (non-blocking).

seek(offset) → Awaitable[None]

Seek within the file off-thread.

close() → Awaitable[None]

Close the file handle.

openviper.http.response

All response classes accept status_code and headers arguments. The headers dict may include any additional response headers.

class Response(content=None, status_code=200, headers=None, media_type=None)

Base ASGI response. content may be bytes, str, or None.

set_cookie(key, value='', max_age=None, expires=None, path='/', domain=None, secure=False, httponly=False, samesite='lax')

Append a Set-Cookie header.

delete_cookie(key, path='/', domain=None)

Append a Set-Cookie header that expires the named cookie.

headers -> MutableHeaders

Mutable response header map. Use .set() or ["name"] = value to add/change headers before the response is sent.

class JSONResponse(content, status_code=200, headers=None, indent=None)

Serialize content to JSON using orjson (C extension). Handles datetime, date, UUID, and FK proxy objects automatically. Pass indent=2 for pretty-printed output.

class HTMLResponse(content=None, status_code=200, headers=None, template=None, context=None, template_dir='templates')

Return HTML. Either pass content as a string, or provide template (a Jinja2 template name) and context for template rendering.

class PlainTextResponse(content, status_code=200, headers=None)

Return a plain-text string with Content-Type: text/plain.

class RedirectResponse(url, status_code=307, headers=None)

HTTP redirect to url. Default status is 307 (Temporary Redirect). Use status_code=301 for permanent redirects.

class StreamingResponse(content, status_code=200, headers=None, media_type=None)

Stream an async generator (or sync iterator) of bytes chunks to the client. content may also be a zero-argument callable that returns an async generator.

class FileResponse(path, status_code=200, headers=None, *, media_type=None, filename=None, allowed_dir=None)

Stream a file from the filesystem. Automatically sets Content-Type, ETag, Last-Modified, and Content-Disposition (when filename is given). Supports If-None-Match and If-Modified-Since conditional requests (returns 304 when appropriate).

Pass allowed_dir to restrict path to a safe directory, preventing path-traversal attacks.

class GZipResponse(content, minimum_size=500, compresslevel=6)

Wrap another Response and gzip-compress its body when its size exceeds minimum_size bytes.

Common HTTP Status Codes

The status_code parameter accepts any integer. Commonly used values:

openviper.http.views

class View

Base class-based view. Subclass and implement one or more HTTP-verb methods (get, post, put, patch, delete, head, options). Unimplemented methods return 405 Method Not Allowed.

Handlers can return a Response object, or a dict/list which is automatically wrapped in a JSONResponse.

Class attributes:

http_method_names: list[str]

Lowercase method names this view handles. Defaults to all standard HTTP verbs.

serializer_class

Optional Pydantic serializer attached for OpenAPI requestBody schema generation.

Methods:

dispatch(request, **kwargs) → Awaitable[Response]

Route request to the appropriate handler method.

classmethod as_view(**initkwargs) → Callable

Return an async callable suitable for use as a route handler. initkwargs are forwarded to __init__ for each request.

classmethod register(router, path, *, name=None, **initkwargs)

Shorthand to register the view on router at path. Automatically determines which HTTP methods are implemented.

@action(methods=None, detail=False, url_path=None, name=None)

Mark a View method as a custom action for automatic routing.

Parameters:

• methods (list[str]) – List of HTTP methods (e.g. ["GET", "POST"]). Defaults to ["GET"].

• detail (bool) –

  • If False (default), the action is for the collection (e.g. /users/search).

  • If True, the action is for a single instance (e.g. /users/{id}/deactivate).

• url_path (str) – Optional override for the URL segment. Defaults to the method name.

• name (str) – Optional name for the reverse URL lookup. Defaults to the method name.

Function-Based Views

from openviper.routing.router import Router
from openviper.http.request import Request
from openviper.http.response import JSONResponse

router = Router()

@router.get("/posts")
async def list_posts(request: Request) -> JSONResponse:
    posts = await Post.objects.filter(is_published=True).order_by("-created_at").all()
    return JSONResponse([p._to_dict() for p in posts])

@router.post("/posts")
async def create_post(request: Request) -> JSONResponse:
    data = await request.json()
    post = await Post.objects.create(**data)
    return JSONResponse(post._to_dict(), status_code=201)

Reading Query Parameters

@router.get("/search")
async def search(request: Request) -> JSONResponse:
    q = request.query_params.get("q", "")
    page = int(request.query_params.get("page", 1))
    return JSONResponse({"query": q, "page": page})

Class-Based Views

from openviper.http.views import View
from openviper.http.response import JSONResponse
from openviper.exceptions import NotFound

class PostDetailView(View):
    async def get(self, request: Request, post_id: int) -> JSONResponse:
        post = await Post.objects.get_or_none(id=post_id)
        if post is None:
            raise NotFound()
        return JSONResponse(post._to_dict())

    async def put(self, request: Request, post_id: int) -> JSONResponse:
        post = await Post.objects.get(id=post_id)
        data = await request.json()
        for k, v in data.items():
            setattr(post, k, v)
        await post.save()
        return JSONResponse(post._to_dict())

    async def delete(self, request: Request, post_id: int) -> JSONResponse:
        post = await Post.objects.get(id=post_id)
        await post.delete()
        return JSONResponse({"deleted": True})

# Register with router
PostDetailView.register(router, "/posts/{post_id:int}")

Extra View Actions

You can add custom endpoints to a View using the @action decorator. These are automatically registered when the view is mounted.

from openviper.http.views import View, action

class UserView(View):
    async def get(self, request):
        """List users."""
        return {"users": []}

    @action(detail=False, methods=["GET"])
    async def search(self, request):
        """Search users: GET /users/search?q=..."""
        q = request.query_params.get("q")
        return {"query": q, "results": []}

    @action(detail=True, methods=["POST"])
    async def deactivate(self, request, id):
        """Deactivate a user: POST /users/{id}/deactivate"""
        return {"id": id, "active": False}

# Registering UserView at "/users" will create:
# GET  /users                  -> UserView.get
# GET  /users/search           -> UserView.search
# POST /users/{id}/deactivate  -> UserView.deactivate
UserView.register(router, "/users")

File Upload

@router.post("/upload")
async def upload(request: Request) -> JSONResponse:
    form = await request.form()
    avatar = form.get("avatar")          # UploadFile instance
    if avatar:
        content = await avatar.read()
        # save content …
    return JSONResponse({"filename": avatar.filename if avatar else None})

Cookie Handling

from openviper.http.response import JSONResponse

@router.post("/set-pref")
async def set_preference(request: Request) -> JSONResponse:
    response = JSONResponse({"status": "ok"})
    response.set_cookie(
        "theme",
        "dark",
        max_age=60 * 60 * 24 * 365,   # 1 year
        httponly=False,
        samesite="lax",
    )
    return response

@router.post("/clear-pref")
async def clear_preference(request: Request) -> JSONResponse:
    response = JSONResponse({"status": "ok"})
    response.delete_cookie("theme")
    return response

Streaming Response

from openviper.http.response import StreamingResponse
import asyncio

async def event_generator():
    for i in range(10):
        yield f"data: {i}\n\n".encode()
        await asyncio.sleep(1)

@router.get("/events")
async def sse(request: Request) -> StreamingResponse:
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
    )

File Download

from openviper.http.response import FileResponse

@router.get("/download/{filename:str}")
async def download(request: Request, filename: str) -> FileResponse:
    return FileResponse(
        f"/media/uploads/{filename}",
        filename=filename,                     # triggers Content-Disposition
        allowed_dir="/media/uploads",          # prevent path traversal
    )

Redirect

from openviper.http.response import RedirectResponse

@router.get("/old-url")
async def old_url(request: Request) -> RedirectResponse:
    return RedirectResponse("/new-url", status_code=301)

Template Rendering

from openviper.http.response import HTMLResponse

@router.get("/")
async def home(request: Request) -> HTMLResponse:
    posts = await Post.objects.filter(is_published=True).limit(10).all()
    return HTMLResponse(template="home.html", context={"posts": posts, "request": request})

GZip Compression

from openviper.http.response import JSONResponse, GZipResponse

@router.get("/large-data")
async def large_data(request: Request) -> GZipResponse:
    data = await fetch_large_dataset()
    return GZipResponse(JSONResponse(data), minimum_size=1024, compresslevel=6)