1import inspect
2import json
3import re
4from collections.abc import Sequence
5from typing import Any, get_type_hints
6
7from plain.urls import Router, URLPattern, URLResolver
8
9from .helpers import json_content
10from .utils import merge_data, schema_from_type, typed_dict_from_annotation
11
12# A leading `GET /path/` line in a docstring is dropped — the URL is already in `paths`.
13_LEADING_HTTP_METHOD = re.compile(
14 r"^(GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS)\s+\S+\s*$",
15 re.IGNORECASE,
16)
17
18
19def _merge_parameters(*sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
20 """Merge parameter lists by `(name, in)` (or `$ref`). Later sources override earlier ones."""
21 by_key: dict[tuple[str, str], dict[str, Any]] = {}
22 for source in sources:
23 for p in source:
24 key = ("$ref", p["$ref"]) if "$ref" in p else (p["name"], p["in"])
25 by_key[key] = p
26 return list(by_key.values())
27
28
29def _build_operation_id(view_class: type, method: str) -> str:
30 return f"{view_class.__name__}_{method}"
31
32
33def _schema_for_converter(converter: Any) -> dict[str, Any]:
34 if converter.keyword == "int":
35 return {"type": "integer"}
36 if converter.keyword == "uuid":
37 return {"type": "string", "format": "uuid"}
38 return {"type": "string", "pattern": converter.regex}
39
40
41def _security_schemes_for_view(view_class: type) -> dict[str, dict[str, Any]]:
42 """Collect `openapi_security_schemes` declared on a view's MRO."""
43 schemes: dict[str, dict[str, Any]] = {}
44 for cls in reversed(view_class.__mro__):
45 declared = cls.__dict__.get("openapi_security_schemes")
46 if declared:
47 schemes.update(declared)
48 return schemes
49
50
51def _parameters_for_view(view_class: type) -> list[dict[str, Any]]:
52 """Walk MRO collecting `openapi_parameters`; descendants override by `(name, in)`."""
53 return _merge_parameters(
54 *(
55 cls.__dict__.get("openapi_parameters") or []
56 for cls in reversed(view_class.__mro__)
57 )
58 )
59
60
61def _docstring_summary_description(
62 class_method: Any,
63 view_class: type,
64) -> dict[str, str]:
65 """PEP 257 split: first paragraph → `summary`, rest → `description`. Falls back to the view class."""
66 doc = (inspect.getdoc(class_method) or inspect.getdoc(view_class) or "").strip()
67 if not doc:
68 return {}
69
70 first, _, rest = doc.partition("\n")
71 if _LEADING_HTTP_METHOD.match(first.strip()):
72 doc = rest.lstrip()
73
74 summary, _, description = doc.partition("\n\n")
75 summary = " ".join(summary.split())
76 description = description.strip()
77
78 out: dict[str, str] = {}
79 if summary:
80 out["summary"] = summary
81 if description:
82 out["description"] = description
83 return out
84
85
86class OpenAPISchemaGenerator:
87 def __init__(self, router: Router):
88 # Get initial schema from the router
89 self.schema = getattr(router, "openapi_schema", {}).copy()
90 self.components = getattr(router, "openapi_components", {}).copy()
91
92 self.schema["paths"] = self.get_paths(router.urls)
93
94 if self.components:
95 self.schema["components"] = self.components
96
97 def as_json(self, indent: int) -> str:
98 return json.dumps(self.schema, indent=indent, sort_keys=True)
99
100 def as_yaml(self, indent: int) -> str:
101 import yaml
102
103 # Don't want to get anchors when we dump...
104 cleaned = json.loads(self.as_json(indent=0))
105 return yaml.safe_dump(cleaned, indent=indent, sort_keys=True)
106
107 def get_paths(
108 self,
109 urls: Sequence[URLPattern | URLResolver],
110 ) -> dict[str, dict[str, Any]]:
111 paths = {}
112
113 for url_pattern in urls:
114 if isinstance(url_pattern, URLResolver):
115 paths.update(self.get_paths(url_pattern.url_patterns))
116 elif isinstance(url_pattern, URLPattern):
117 if operations := self.operations_for_url_pattern(url_pattern):
118 path = self.path_from_url_pattern(url_pattern, "/")
119 # TODO could have class level summary/description?
120 paths[path] = operations
121 else:
122 raise TypeError(f"Unknown url pattern: {url_pattern}")
123
124 return paths
125
126 def path_from_url_pattern(self, url_pattern: URLPattern, root_path: str) -> str:
127 path = root_path + url_pattern.raw_route
128 if url_pattern.trailing_slash and url_pattern.raw_route:
129 path += "/"
130
131 for name, converter in url_pattern.converters.items():
132 # Handle both `<type:name>` and the `<name>` shorthand for the default `str` converter.
133 path = path.replace(f"<{converter.keyword}:{name}>", f"{{{name}}}")
134 path = path.replace(f"<{name}>", f"{{{name}}}")
135 return path
136
137 def extract_components(self, obj: Any) -> None:
138 """
139 Extract components from a view or router.
140 """
141 if hasattr(obj, "openapi_components"):
142 self.components = merge_data(
143 self.components,
144 getattr(obj, "openapi_components", {}),
145 )
146
147 def include_view(self, view_class: type) -> bool:
148 """Override to drop a view from the schema."""
149 return True
150
151 def _response_from_return_annotation(
152 self, class_method: Any
153 ) -> dict[str, Any] | None:
154 """Build a 200 response fragment from the method's return annotation if it points at a TypedDict."""
155 try:
156 hints = get_type_hints(class_method)
157 except Exception:
158 return None
159
160 return_type = hints.get("return")
161 if return_type is None:
162 return None
163
164 typed_dict = typed_dict_from_annotation(return_type)
165 if typed_dict is None:
166 return None
167
168 top_ref = schema_from_type(typed_dict, components=self.components)
169 return {
170 "responses": {
171 "200": {
172 "description": "OK",
173 "content": json_content(top_ref),
174 }
175 }
176 }
177
178 def operations_for_url_pattern(
179 self,
180 url_pattern: URLPattern,
181 ) -> dict[str, Any]:
182 operations: dict[str, Any] = {}
183
184 if not self.include_view(url_pattern.view_class):
185 return operations
186
187 # `View` defines runtime stubs for every handler, so gating on
188 # `implemented_methods` is what tells us which verbs the leaf class
189 # actually handles (vs. inheriting a stub that will 405).
190 implemented = getattr(
191 url_pattern.view_class, "implemented_methods", frozenset()
192 )
193
194 inherited_params = _parameters_for_view(url_pattern.view_class)
195 auto_params = self.parameters_from_url_patterns([url_pattern])
196 schemes = _security_schemes_for_view(url_pattern.view_class)
197
198 for vc in reversed(url_pattern.view_class.__mro__):
199 self.extract_components(vc)
200 for method in implemented:
201 class_method = vc.__dict__.get(method)
202 if not class_method:
203 continue
204
205 self.extract_components(class_method)
206 operation = merge_data(
207 getattr(vc, "openapi_schema", {}),
208 getattr(class_method, "openapi_schema", {}),
209 )
210
211 already_has_2xx = any(
212 code.startswith("2") for code in operation.get("responses", {})
213 )
214 if not already_has_2xx:
215 inferred = self._response_from_return_annotation(class_method)
216 if inferred is not None:
217 operation = merge_data(operation, inferred)
218
219 for key, value in _docstring_summary_description(
220 class_method, vc
221 ).items():
222 operation.setdefault(key, value)
223
224 if not operation:
225 continue
226
227 merged_params = _merge_parameters(
228 auto_params,
229 inherited_params,
230 list(operation.get("parameters", [])),
231 )
232 if merged_params:
233 operation["parameters"] = merged_params
234
235 operation.setdefault(
236 "operationId",
237 _build_operation_id(url_pattern.view_class, method),
238 )
239
240 if "security" not in operation and schemes:
241 operation["security"] = [{name: []} for name in schemes]
242 self.components = merge_data(
243 self.components,
244 {"securitySchemes": schemes},
245 )
246
247 # If there are no responses in the 2XX or 3XX range, then don't return it at all.
248 # Most likely the developer didn't define any actual responses for their endpoint,
249 # and all we did was inherit the base error responses.
250 keep_operation = False
251 for status_code in operation.get("responses", {}):
252 if status_code.startswith(("2", "3")):
253 keep_operation = True
254 break
255
256 if operation and keep_operation:
257 if method in operations:
258 # Merge operation with existing data
259 operations[method] = merge_data(operations[method], operation)
260 else:
261 operations[method] = operation
262
263 return operations
264
265 def parameters_from_url_patterns(
266 self, url_patterns: list[URLPattern]
267 ) -> list[dict[str, Any]]:
268 """Need to process any parent/included url patterns too"""
269 parameters = []
270
271 for url_pattern in url_patterns:
272 for name, converter in url_pattern.converters.items():
273 parameters.append(
274 {
275 "name": name,
276 "in": "path",
277 "required": True,
278 "schema": _schema_for_converter(converter),
279 }
280 )
281
282 return parameters