Coverage for nexios\openapi\_builder.py: 31%
89 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-21 20:31 +0100
1from functools import wraps
2from typing import Optional, Dict, List, Type, Union
3from uuid import uuid4
4from pydantic import BaseModel
5from nexios.http import Request, Response
6from .config import OpenAPIConfig
7from .models import (
8 Components,
9 Parameter,
10 ExternalDocumentation,
11 RequestBody,
12 MediaType,
13 Response as OpenAPIResponse,
14 Operation,
15 Schema,
16 PathItem,
17)
20class APIDocumentation:
21 _instance = None
23 def __new__(cls, *args, **kwargs):
24 if not cls._instance:
25 cls._instance = super().__new__(cls)
26 return cls._instance
28 def __init__(
29 self, app: Optional["NexiosApp"] = None, config: Optional[OpenAPIConfig] = None
30 ):
31 self.app = app
32 self.config = config or OpenAPIConfig()
33 if app:
34 self._setup_doc_routes()
36 def _setup_doc_routes(self):
37 """Set up routes for serving OpenAPI specification"""
39 @self.app.get("/openapi.json", exclude_from_schema=True) # type:ignore
40 async def serve_openapi(request: Request, response: Response):
41 openapi_json = self.config.openapi_spec.model_dump(
42 by_alias=True, exclude_none=True
43 )
44 return response.json(openapi_json)
46 @self.app.get("/docs", exclude_from_schema=True) # type:ignore
47 async def swagger_ui(request: Request, response: Response):
48 return response.html(self._generate_swagger_ui())
50 @classmethod
51 def get_instance(cls):
52 return cls._instance
54 def _generate_swagger_ui(self) -> str:
55 """Generate Swagger UI HTML"""
56 return f"""
57 <!DOCTYPE html>
58 <html>
59 <head>
60 <title>{self.config.openapi_spec.info.title} - Docs</title>
61 <link rel="stylesheet" href="https://unpkg.com/swagger-ui-dist@4.18.3/swagger-ui.css">
62 </head>
63 <body>
64 <div id="swagger-ui"></div>
65 <script src="https://unpkg.com/swagger-ui-dist@4/swagger-ui-bundle.js"></script>
66 <script>
67 window.onload = function() {
68 SwaggerUIBundle({
69 url: '/openapi.json',
70 dom_id: '#swagger-ui',
71 presets: [
72 SwaggerUIBundle.presets.apis,
73 SwaggerUIBundle.SwaggerUIStandalonePreset
74 ],
75 layout: "BaseLayout"
76 } );
77 }
78 </script>
79 </body>
80 </html>
81 """
83 def document_endpoint(
84 self,
85 path: str,
86 method: str,
87 summary: str = "",
88 description: Optional[str] = None,
89 parameters: Optional[List[Parameter]] = None,
90 request_body: Optional[Type[BaseModel]] = None,
91 responses: Optional[Union[BaseModel, Dict[int, BaseModel]]] = None,
92 tags: Optional[List[str]] = None,
93 security: Optional[List[Dict[str, List[str]]]] = None,
94 operation_id: Optional[str] = None,
95 deprecated: bool = False,
96 external_docs: Optional[ExternalDocumentation] = None,
97 ):
98 """
99 Decorator to document API endpoints with OpenAPI specification
101 :param path: URL path of the endpoint
102 :param method: HTTP method (get, post, put, delete, etc.)
103 :param summary: Short summary of the endpoint
104 :param description: Detailed description of the endpoint
105 :param parameters: List of parameters
106 :param request_body: Pydantic model for request body
107 :param responses: Response model(s)
108 :param tags: Categorization tags for the endpoint
109 :param security: Security requirements
110 :param operation_id: Unique identifier for the operation
111 :param deprecated: Whether the endpoint is deprecated
112 :param external_docs: External documentation reference
113 """
115 def decorator(func):
116 # Prepare request body specification
117 request_body_spec = None
119 if request_body:
120 self.add_schema(request_body)
121 if request_body:
122 request_body_spec = RequestBody(
123 content={
124 "application/json": MediaType( # type:ignore
125 schema=Schema( # type:ignore
126 **request_body.model_json_schema()
127 )
128 )
129 }
130 )
131 else:
132 if method.upper() not in ["GET", "DELETE"]:
133 request_body_spec = RequestBody(
134 content={
135 "application/json": MediaType( # type:ignore
136 schema=Schema(
137 example={
138 "example": "This is an example request body"
139 }
140 ) # type:ignore
141 )
142 }
143 )
145 # self.config.openapi_spec.components.schemas[request_body.__name__] = request_body
146 # Prepare responses specification
147 responses_spec = {}
148 if responses:
149 if isinstance(responses, type) and issubclass(responses, BaseModel):
150 responses_spec["200"] = OpenAPIResponse(
151 description="Successful Response",
152 content={
153 "application/json": MediaType( # type:ignore
154 schema=Schema( # type:ignore
155 **responses.model_json_schema(),
156 )
157 )
158 },
159 )
160 # Handle List[Model] case
161 elif hasattr(responses, "__origin__") and responses.__origin__ is list:
162 item_model = responses.__args__[0]
163 if issubclass(item_model, BaseModel):
164 responses_spec["200"] = OpenAPIResponse(
165 description="Successful Response",
166 content={
167 "application/json": MediaType( # type:ignore
168 schema=Schema( # type:ignore
169 type="array",
170 items=Schema(**item_model.model_json_schema()),
171 )
172 )
173 },
174 )
175 # Multiple responses case
176 elif isinstance(responses, dict):
177 for status_code, model in responses.items():
178 # Handle direct model
179 if isinstance(model, type) and issubclass(model, BaseModel):
180 responses_spec[str(status_code)] = OpenAPIResponse(
181 description=f"Response for status code {status_code}",
182 content={
183 "application/json": MediaType( # type:ignore
184 schema=Schema( # type:ignore
185 **model.model_json_schema()
186 )
187 )
188 },
189 )
190 # Handle List[Model]
191 elif hasattr(model, "__origin__") and model.__origin__ is list:
192 item_model = model.__args__[0]
193 if issubclass(item_model, BaseModel):
194 responses_spec[str(status_code)] = OpenAPIResponse(
195 description=f"Response for status code {status_code}",
196 content={
197 "application/json": MediaType( # type:ignore
198 schema=Schema( # type:ignore
199 type="array",
200 items=Schema(
201 **item_model.model_json_schema()
202 ),
203 )
204 )
205 },
206 )
207 else:
208 # Default response if no responses specified
209 responses_spec["200"] = OpenAPIResponse(
210 description="Successful Response",
211 content={
212 "application/json": MediaType( # type:ignore
213 schema=Schema(
214 example={"example": "This is an example response"},
215 type="object",
216 ) # type:ignore
217 )
218 },
219 )
221 if not responses_spec:
222 responses_spec["200"] = OpenAPIResponse(
223 description="Successful Response",
224 content={
225 "application/json": MediaType( # type:ignore
226 schema=Schema(
227 example={"example": "This is an example response"},
228 type="object",
229 ) # type:ignore
230 )
231 },
232 )
234 if responses:
235 if isinstance(responses, type) and issubclass(responses, BaseModel):
236 self.add_schema(responses)
237 elif hasattr(responses, "__origin__") and responses.__origin__ is list:
238 item_model = responses.__args__[0]
239 if issubclass(item_model, BaseModel):
240 self.add_schema(item_model)
241 elif isinstance(responses, dict):
242 for model in responses.values():
243 if isinstance(model, type) and issubclass(model, BaseModel):
244 self.add_schema(model)
245 elif hasattr(model, "__origin__") and model.__origin__ is list:
246 item_model = model.__args__[0]
247 if issubclass(item_model, BaseModel):
248 self.add_schema(item_model)
250 operation = Operation(
251 summary=summary,
252 description=description,
253 responses=responses_spec,
254 tags=tags,
255 parameters=parameters or [], # type:ignore
256 requestBody=request_body_spec,
257 security=security,
258 operationId=operation_id or str(uuid4()),
259 deprecated=deprecated,
260 externalDocs=external_docs,
261 )
263 # Add operation to the OpenAPI specification
264 if path not in self.config.openapi_spec.paths:
265 self.config.openapi_spec.paths[path] = PathItem()
267 setattr(self.config.openapi_spec.paths[path], method.lower(), operation)
269 @wraps(func)
270 async def wrapper(*args, **kwargs):
271 return await func(*args, **kwargs)
273 return wrapper
275 return decorator
277 def add_schema(self, schema: Type[BaseModel]) -> None:
278 """Add a schema to the OpenAPI components section
280 Args:
281 schema: A Pydantic model to be added to the components/schemas section
282 """
283 if not self.config.openapi_spec.components:
284 self.config.openapi_spec.components = Components()
286 if not self.config.openapi_spec.components.schemas:
287 self.config.openapi_spec.components.schemas = {}
289 self.config.openapi_spec.components.schemas[schema.__name__] = Schema(
290 **schema.model_json_schema()
291 )