1from __future__ import annotations
2
3import inspect
4from collections.abc import Awaitable, Callable
5from typing import Any, ClassVar
6
7from plain.http import (
8 NotAllowedResponse,
9 Request,
10 Response,
11)
12from plain.logs import get_framework_logger, log_exception
13
14from .exceptions import ResponseException
15
16logger = get_framework_logger("plain.request")
17
18
19# TRACE is an XST-adjacent debugging verb, CONNECT is a proxy concept —
20# neither belongs in an application view. OPTIONS is provided by the base
21# directly; HEAD falls back to GET at dispatch time.
22_HANDLER_NAMES = ("get", "post", "put", "patch", "delete", "head")
23
24
25class View[HandlerResult = Response]:
26 request: Request
27 url_kwargs: dict[str, Any]
28
29 implemented_methods: ClassVar[frozenset[str]] = frozenset()
30
31 def __init_subclass__(cls, **kwargs: Any) -> None:
32 super().__init_subclass__(**kwargs)
33 cls.implemented_methods = frozenset(
34 name
35 for name in _HANDLER_NAMES
36 if getattr(cls, name, None) is not getattr(View, name, None)
37 )
38
39 def get(self) -> HandlerResult:
40 raise NotImplementedError
41
42 def post(self) -> HandlerResult:
43 raise NotImplementedError
44
45 def put(self) -> HandlerResult:
46 raise NotImplementedError
47
48 def patch(self) -> HandlerResult:
49 raise NotImplementedError
50
51 def delete(self) -> HandlerResult:
52 raise NotImplementedError
53
54 def head(self) -> HandlerResult:
55 raise NotImplementedError
56
57 def __init__(
58 self,
59 *,
60 request: Request,
61 url_kwargs: dict[str, Any] | None = None,
62 ) -> None:
63 self.request = request
64 self.url_kwargs = url_kwargs or {}
65
66 def get_request_handler(self) -> Callable[[], Any] | None:
67 """Return the handler for the current request method.
68
69 HEAD falls back to `get` when no explicit `head` handler is defined,
70 per HTTP semantics (HEAD == GET without a response body). The body
71 is stripped at the transport layer, not here.
72 """
73
74 if not self.request.method:
75 raise AttributeError("HTTP method is not set")
76
77 if self.request.method == "OPTIONS":
78 return self.options
79
80 name = self.request.method.lower()
81 if name in self.implemented_methods:
82 return getattr(self, name)
83
84 if self.request.method == "HEAD" and "get" in self.implemented_methods:
85 return self.get
86
87 return None
88
89 def before_request(self) -> None:
90 """Pre-dispatch hook. Raise to reject the request."""
91
92 def after_response(self, response: Response) -> Response:
93 """Post-dispatch hook. Runs for every response — successes, errors, 405s.
94
95 Return the response (possibly mutated or replaced). Exceptions
96 raised here escape to the framework error renderer — they are
97 not routed through `handle_exception`.
98 """
99 return response
100
101 def handle_exception(self, exc: Exception) -> Response:
102 """Translate a raised exception into a response. Re-raise to defer to the framework default.
103
104 Returning a 4xx response treats the exception as a handled outcome
105 (e.g. ValidationError → 400) — no logging, no exception attachment.
106 Returning a 5xx response is treated as a real failure: the framework
107 attaches `response.exception` and calls `log_exception` for you, so
108 observability tooling can record it from the response. Re-raising
109 escapes to the framework error renderer, which logs and renders
110 `{status}.html`.
111 """
112 raise exc
113
114 def get_response(self) -> Response:
115 try:
116 self.before_request()
117
118 handler = self.get_request_handler()
119 if not handler:
120 logger.warning(
121 "Method not allowed",
122 extra={
123 "method": self.request.method,
124 "path": self.request.path,
125 "status_code": 405,
126 },
127 )
128 response: Response = NotAllowedResponse(self._allowed_methods())
129 elif inspect.iscoroutinefunction(handler):
130 return self._dispatch_handler_async(handler) # ty: ignore[invalid-return-type]
131 else:
132 response = self.convert_result_to_response(handler())
133 except Exception as e:
134 response = self._respond_to_exception(e)
135 return self.after_response(response)
136
137 async def _dispatch_handler_async(
138 self, handler: Callable[[], Awaitable[Response]]
139 ) -> Response:
140 try:
141 result = await handler()
142 response = self.convert_result_to_response(result)
143 except Exception as e:
144 response = self._respond_to_exception(e)
145 return self.after_response(response)
146
147 def _respond_to_exception(self, exc: Exception) -> Response:
148 if isinstance(exc, ResponseException):
149 return exc.response
150 response = self.handle_exception(exc)
151 # 5xx responses from handle_exception represent a real failure that
152 # the view chose to render itself. Stamp the response with the
153 # exception and log centrally so subclasses don't each have to
154 # remember (and so the canonical OTel SERVER span can record it via
155 # `_finalize_span`).
156 if response.status_code >= 500:
157 log_exception(self.request, exc)
158 response.exception = exc
159 return response
160
161 def convert_result_to_response(self, result: HandlerResult) -> Response:
162 """Hook for subclasses (e.g. `APIView`) to accept shorthand return types."""
163 if isinstance(result, Response):
164 return result
165
166 raise TypeError(
167 f"{type(self).__name__} handlers must return a Response "
168 f"(got {type(result).__name__}). "
169 "Wrap raw data in a Response/JsonResponse, or use APIView for "
170 "dict/list/tuple shorthand returns."
171 )
172
173 def options(self) -> Response:
174 """Handle responding to requests for the OPTIONS HTTP verb."""
175 response = Response()
176 response.headers["Allow"] = ", ".join(self._allowed_methods())
177 response.headers["Content-Length"] = "0"
178 return response
179
180 def _allowed_methods(self) -> list[str]:
181 methods = {m.upper() for m in self.implemented_methods}
182 if "GET" in methods:
183 methods.add("HEAD")
184 methods.add("OPTIONS")
185 return sorted(methods)