Coverage for nexios\application.py: 55%
161 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-21 20:31 +0100
1from typing import (
2 Any,
3 Awaitable,
4 Callable,
5 Dict,
6 List,
7 Optional,
8 Type,
9 Union,
10 AsyncContextManager,
11)
13from pydantic import BaseModel
14from typing_extensions import Annotated, Doc
16from nexios.config import DEFAULT_CONFIG, MakeConfig
17from nexios.events import AsyncEventEmitter
18from nexios.exception_handler import ExceptionHandlerType, ExceptionMiddleware
19from nexios.logging import create_logger
20from nexios.middlewares.core import BaseMiddleware, Middleware
21from nexios.middlewares.errors.server_error_handler import (
22 ServerErrorMiddleware,
23 ServerErrHandlerType,
24)
25from nexios.openapi._builder import APIDocumentation
26from nexios.openapi.config import OpenAPIConfig
27from nexios.openapi.models import HTTPBearer, Parameter, Path, Schema
28from nexios.structs import URLPath
29from .routing import Router, Routes, WSRouter, WebsocketRoutes
30from .types import (
31 ASGIApp,
32 HandlerType,
33 Message,
34 MiddlewareType,
35 Receive,
36 Scope,
37 Send,
38 WsHandlerType,
39)
42logger = create_logger("nexios")
43lifespan_manager = Callable[["NexiosApp"], AsyncContextManager[bool]]
46class NexiosApp(object):
47 def __init__(
48 self,
49 config: Annotated[
50 Optional[MakeConfig],
51 Doc(
52 """
53 This subclass is derived from the MakeConfig class and is responsible for managing configurations within the Nexios framework. It takes arguments in the form of dictionaries, allowing for structured and flexible configuration handling. By using dictionaries, this subclass makes it easy to pass multiple configuration values at once, reducing complexity and improving maintainability.
55 One of the key advantages of this approach is its ability to dynamically update and modify settings without requiring changes to the core codebase. This is particularly useful in environments where configurations need to be frequently adjusted, such as database settings, API credentials, or feature flags. The subclass can also validate the provided configuration data, ensuring that incorrect or missing values are handled properly.
57 Additionally, this design allows for merging and overriding configurations, making it adaptable for various use cases. Whether used for small projects or large-scale applications, this subclass ensures that configuration management remains efficient and scalable. By extending MakeConfig, it leverages existing functionality while adding new capabilities tailored to Nexios. This makes it an essential component for maintaining structured and well-organized application settings.
58 """
59 ),
60 ] = DEFAULT_CONFIG,
61 title: Annotated[
62 Optional[str],
63 Doc(
64 """
65 The title of the API, used in the OpenAPI documentation.
66 """
67 ),
68 ] = None,
69 version: Annotated[
70 Optional[str],
71 Doc(
72 """
73 The version of the API, used in the OpenAPI documentation.
74 """
75 ),
76 ] = None,
77 description: Annotated[
78 Optional[str],
79 Doc(
80 """
81 A brief description of the API, used in the OpenAPI documentation.
82 """
83 ),
84 ] = None,
85 middlewares: Annotated[
86 List[Middleware],
87 Doc(
88 "A list of middlewares, where each middleware is either a class inherited from BaseMiddleware or an asynchronous callable function that accepts request, response, and callnext"
89 ),
90 ] = [],
91 server_error_handler: Annotated[
92 Optional[ServerErrHandlerType],
93 Doc(
94 """
95 A function in Nexios responsible for handling server-side exceptions by logging errors, reporting issues, or initiating recovery mechanisms. It prevents crashes by intercepting unexpected failures, ensuring the application remains stable and operational. This function provides a structured approach to error management, allowing developers to define custom handling strategies such as retrying failed requests, sending alerts, or gracefully degrading functionality. By centralizing error processing, it improves maintainability and observability, making debugging and monitoring more efficient. Additionally, it ensures that critical failures do not disrupt the entire system, allowing services to continue running while appropriately managing faults and failures."""
96 ),
97 ] = None,
98 lifespan: Optional[lifespan_manager] = None,
99 routes: Optional[List[Routes]] = None,
100 ):
102 self.config = config or DEFAULT_CONFIG
103 from nexios.config import set_config, get_config
105 try:
106 get_config()
107 except RuntimeError:
108 set_config(self.config)
109 self.server_error_handler = None
110 self.ws_router = WSRouter()
111 self.ws_routes: List[WebsocketRoutes] = []
112 self.http_middlewares: List[Middleware] = middlewares or []
113 self.ws_middlewares: List[ASGIApp] = []
114 self.startup_handlers: List[Callable[[], Awaitable[None]]] = []
115 self.shutdown_handlers: List[Callable[[], Awaitable[None]]] = []
116 self.exceptions_handler = ExceptionMiddleware()
117 self.server_error_handler = server_error_handler
119 self.app = Router(routes=routes)
120 self.router = self.app
121 self.route = self.router.route
122 self.lifespan_context: Optional[lifespan_manager] = lifespan
124 openapi_config: Dict[str, Any] = self.config.to_dict().get(
125 "openapi", {}
126 ) # type:ignore
127 self.openapi_config = OpenAPIConfig(
128 title=openapi_config.get("title", title or "Nexios API"),
129 version=openapi_config.get("version", version or "1.0.0"),
130 description=openapi_config.get(
131 "description", description or "Nexios API Documentation"
132 ),
133 license=openapi_config.get("license"),
134 contact=openapi_config.get("contact"),
135 )
137 self.openapi_config.add_security_scheme(
138 "bearerAuth", HTTPBearer(type="http", scheme="bearer", bearerFormat="JWT")
139 )
141 self.docs = APIDocumentation(
142 app=self,
143 config=self.openapi_config,
144 )
146 self.events = AsyncEventEmitter()
147 self.title = title or "Nexios API"
149 def on_startup(self, handler: Callable[[], Awaitable[None]]) -> None:
150 """
151 Registers a startup handler that executes when the application starts.
153 This method allows you to define functions that will be executed before
154 the application begins handling requests. It is useful for initializing
155 resources such as database connections, loading configuration settings,
156 or preparing caches.
158 The provided function must be asynchronous (`async def`) since it
159 will be awaited during the startup phase.
161 Args:
162 handler (Callable): An asynchronous function to be executed at startup.
164 Returns:
165 Callable: The same handler function, allowing it to be used as a decorator.
167 Example:
168 ```python
170 @app.on_startup
171 async def connect_to_db():
172 global db
173 db = await Database.connect("postgres://user:password@localhost:5432/mydb")
174 print("Database connection established.")
176 @app.on_startup
177 async def cache_warmup():
178 global cache
179 cache = await load_initial_cache()
180 print("Cache warmed up and ready.")
181 ```
183 In this example:
184 - `connect_to_db` establishes a database connection before the app starts.
185 - `cache_warmup` preloads data into a cache for faster access.
187 These functions will be executed in the order they are registered when the
188 application starts.
189 """
190 self.startup_handlers.append(handler)
192 def on_shutdown(self, handler: Callable[[], Awaitable[None]]) -> None:
193 """
194 Registers a shutdown handler that executes when the application is shutting down.
196 This method allows you to define functions that will be executed when the
197 application is stopping. It is useful for cleaning up resources such as
198 closing database connections, saving application state, or gracefully
199 terminating background tasks.
201 The provided function must be asynchronous (`async def`) since it will be
202 awaited during the shutdown phase.
204 Args:
205 handler (Callable): An asynchronous function to be executed during shutdown.
207 Returns:
208 Callable: The same handler function, allowing it to be used as a decorator.
210 Example:
211 ```python
212 app = NexioApp()
214 @app.on_shutdown
215 async def disconnect_db():
216 global db
217 await db.disconnect()
218 print("Database connection closed.")
220 @app.on_shutdown
221 async def clear_cache():
222 global cache
223 await cache.clear()
224 print("Cache cleared before shutdown.")
225 ```
227 In this example:
228 - `disconnect_db` ensures that the database connection is properly closed.
229 - `clear_cache` removes cached data to free up memory before the app stops.
231 These functions will be executed in the order they are registered when the
232 application is shutting down.
233 """
234 self.shutdown_handlers.append(handler)
236 async def _startup(self) -> None:
237 """Execute all startup handlers sequentially"""
238 for handler in self.startup_handlers:
239 try:
240 await handler()
241 except Exception as e:
242 raise e
244 async def _shutdown(self) -> None:
245 """Execute all shutdown handlers sequentially with error handling"""
246 for handler in self.shutdown_handlers:
247 try:
248 await handler()
249 except Exception as e:
250 raise e
252 async def handle_lifespan(self, receive: Receive, send: Send) -> None:
253 """Handle ASGI lifespan protocol events."""
254 self._setup_openapi()
256 try:
257 while True:
258 message: Message = await receive()
259 if message["type"] == "lifespan.startup":
260 try:
261 if self.lifespan_context:
262 # If a lifespan context manager is provided, use it
263 self.lifespan_manager: Any = self.lifespan_context(self)
264 await self.lifespan_manager.__aenter__()
265 else:
266 # Otherwise, fall back to the default startup handlers
267 await self._startup()
268 await send({"type": "lifespan.startup.complete"})
269 except Exception as e:
270 await send(
271 {"type": "lifespan.startup.failed", "message": str(e)}
272 )
273 return
275 elif message["type"] == "lifespan.shutdown":
276 try:
277 if self.lifespan_context:
278 # If a lifespan context manager is provided, use it
279 await self.lifespan_manager.__aexit__(None, None, None)
280 else:
281 # Otherwise, fall back to the default shutdown handlers
282 await self._shutdown()
283 await send({"type": "lifespan.shutdown.complete"})
284 return
285 except Exception as e:
286 await send(
287 {"type": "lifespan.shutdown.failed", "message": str(e)}
288 )
289 return
291 except Exception as e:
292 if message["type"].startswith("lifespan.startup"): # type: ignore
293 await send({"type": "lifespan.startup.failed", "message": str(e)})
294 else:
295 await send({"type": "lifespan.shutdown.failed", "message": str(e)})
297 def _setup_openapi(self) -> None:
298 """Set up automatic OpenAPI documentation"""
299 docs = self.docs
300 for route in self.get_all_routes():
301 if route.exlude_from_schema:
302 continue
303 for method in route.methods:
305 parameters = [
306 Path(name=x, schema=Schema(type="string"), schema_=None) # type: ignore
307 for x in route.param_names
308 ]
310 parameters.extend(route.parameters) # type: ignore
311 docs.document_endpoint(
312 path=route.raw_path,
313 method=method,
314 tags=route.tags,
315 security=route.security,
316 summary=route.summary or "",
317 description=route.description,
318 request_body=route.request_model,
319 parameters=parameters, # type:ignore
320 deprecated=route.deprecated,
321 operation_id=route.operation_id,
322 responses=route.responses,
323 )(route.handler)
325 def add_middleware(
326 self,
327 middleware: Annotated[
328 MiddlewareType,
329 Doc(
330 "A callable middleware function that processes requests and responses."
331 ),
332 ],
333 ) -> None:
334 """
335 Adds middleware to the application.
337 Middleware functions are executed in the request-response lifecycle, allowing
338 modifications to requests before they reach the route handler and responses
339 before they are sent back to the client.
341 Args:
342 middleware (MiddlewareType): A callable that takes a `Request`, `Response`,
343 and a `Callable` (next middleware or handler) and returns a `Response`.
345 Returns:
346 None
348 Example:
349 ```python
350 def logging_middleware(request: Request, response: Response, next_call: Callable) -> Response:
351 print(f"Request received: {request.method} {request.url}")
352 return next_call(request, response)
354 app.add_middleware(logging_middleware)
355 ```
356 """
358 self.http_middlewares.insert(
359 0, Middleware(BaseMiddleware, dispatch=middleware) # type:ignore
360 )
362 def add_ws_route(
363 self,
364 route: Annotated[
365 WebsocketRoutes,
366 Doc("An instance of the Routes class representing a WebSocket route."),
367 ],
368 ) -> None:
369 """
370 Adds a WebSocket route to the application.
372 This method registers a WebSocket route, allowing the application to handle WebSocket connections.
374 Args:
375 route (Routes): The WebSocket route configuration.
377 Returns:
378 None
380 Example:
381 ```python
382 route = Routes("/ws/chat", chat_handler)
383 app.add_ws_route(route)
384 ```
385 """
386 self.ws_router.add_ws_route(route)
388 def mount_router(self, router: Router, path: Optional[str] = None) -> None:
389 """
390 Mounts a router and all its routes to the application.
392 This method allows integrating another `Router` instance, registering all its
393 defined routes into the current application. It is useful for modularizing routes
394 and organizing large applications.
396 Args:
397 router (Router): The `Router` instance whose routes will be added.
399 Returns:
400 None
402 Example:
403 ```python
404 user_router = Router()
406 @user_router.route("/users", methods=["GET"])
407 def get_users(request, response):
408 response.json({"users": ["Alice", "Bob"]})
410 app.mount_router(user_router) # Mounts the user routes into the main app
411 ```
412 """
413 self.router.mount_router(router, path=path)
415 def mount_ws_router(
416 self,
417 router: Annotated[
418 WSRouter,
419 Doc("An instance of Router containing multiple routes to be mounted."),
420 ],
421 ) -> None:
422 """
423 Mounts a router and all its routes to the application.
425 This method allows integrating another `Router` instance, registering all its
426 defined routes into the current application. It is useful for modularizing routes
427 and organizing large applications.
429 Args:
430 router (Router): The `Router` instance whose routes will be added.
432 Returns:
433 None
435 Example:
436 ```python
437 chat_router = WSRouter()
439 @chat_router.ws("/users")
440 def get_users(ws):
441 ...
443 app.mount_ws_router(chat_router) # Mounts the user routes into the main app
444 ```
445 """
446 self.ws_router.mount_router(router)
448 async def handle_websocket(
449 self, scope: Scope, receive: Receive, send: Send
450 ) -> None:
451 app = self.ws_router
452 for mdw in reversed(self.ws_middlewares):
453 app = mdw(app) # type:ignore
454 await app(scope, receive, send)
456 def add_ws_middleware(
457 self,
458 middleware: Annotated[
459 ASGIApp,
460 Doc(
461 "A callable function that intercepts and processes WebSocket connections."
462 ),
463 ],
464 ) -> None:
465 """
466 Adds a WebSocket middleware to the application.
468 WebSocket middleware functions allow pre-processing of WebSocket requests before they
469 reach their final handler. Middleware can be used for authentication, logging, or
470 modifying the WebSocket request/response.
472 Args:
473 middleware (Callable): A callable function that handles WebSocket connections.
475 Returns:
476 None
478 Example:
479 ```python
480 def ws_auth_middleware(ws, next_handler):
481 if not ws.headers.get("Authorization"):
482 ...
483 return next_handler(ws)
485 app.add_ws_middleware(ws_auth_middleware)
486 ```
487 """
488 self.ws_middlewares.append(middleware)
490 def handle_http_request(self, scope: Scope, receive: Receive, send: Send):
491 app = self.app
492 middleware = (
493 [
494 Middleware(
495 BaseMiddleware,
496 dispatch=ServerErrorMiddleware(handler=self.server_error_handler),
497 )
498 ]
499 + self.http_middlewares
500 + [
501 Middleware(
502 BaseMiddleware, dispatch=self.exceptions_handler
503 ) # type:ignore
504 ]
505 )
506 for cls, args, kwargs in reversed(middleware):
507 app = cls(app, *args, **kwargs)
508 return app(scope, receive, send)
510 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
511 """ASGI application callable"""
512 scope["app"] = self
513 scope["base_app"] = self
514 if scope["type"] == "lifespan":
515 await self.handle_lifespan(receive, send)
516 elif scope["type"] == "http":
517 await self.handle_http_request(scope, receive, send)
519 else:
521 await self.handle_websocket(scope, receive, send)
523 def get(
524 self,
525 path: Annotated[
526 str,
527 Doc(
528 """
529 URL path pattern for the GET endpoint.
530 Supports path parameters using {param} syntax.
531 Example: '/users/{user_id}'
532 """
533 ),
534 ],
535 handler: Annotated[
536 Optional[HandlerType],
537 Doc(
538 """
539 Async handler function for GET requests.
540 Receives (request, response) and returns response or raw data.
542 Example:
543 async def get_user(request, response):
544 user = await get_user_from_db(request.path_params['user_id'])
545 return response.json(user)
546 """
547 ),
548 ] = None,
549 name: Annotated[
550 Optional[str],
551 Doc(
552 """
553 Unique route identifier for URL generation.
554 Example: 'get-user-by-id'
555 """
556 ),
557 ] = None,
558 summary: Annotated[
559 Optional[str],
560 Doc(
561 """
562 Brief summary for OpenAPI documentation.
563 Example: 'Retrieves a user by ID'
564 """
565 ),
566 ] = None,
567 description: Annotated[
568 Optional[str],
569 Doc(
570 """
571 Detailed description for OpenAPI documentation.
572 Example: 'Returns full user details including profile information'
573 """
574 ),
575 ] = None,
576 responses: Annotated[
577 Optional[Dict[int, Any]],
578 Doc(
579 """
580 Response models by status code.
581 Example:
582 {
583 200: UserSchema,
584 404: {"description": "User not found"},
585 500: {"description": "Server error"}
586 }
587 """
588 ),
589 ] = None,
590 request_model: Annotated[
591 Optional[Type[BaseModel]],
592 Doc(
593 """
594 Pydantic model for request validation (query params).
595 Example:
596 class UserQuery(BaseModel):
597 active_only: bool = True
598 limit: int = 100
599 """
600 ),
601 ] = None,
602 middlewares: Annotated[
603 List[Any],
604 Doc(
605 """
606 List of route-specific middleware functions.
607 Example: [auth_required, rate_limit]
608 """
609 ),
610 ] = [],
611 tags: Annotated[
612 Optional[List[str]],
613 Doc(
614 """
615 OpenAPI tags for grouping related endpoints.
616 Example: ["Users", "Public"]
617 """
618 ),
619 ] = None,
620 security: Annotated[
621 Optional[List[Dict[str, List[str]]]],
622 Doc(
623 """
624 Security requirements for OpenAPI docs.
625 Example: [{"BearerAuth": []}]
626 """
627 ),
628 ] = None,
629 operation_id: Annotated[
630 Optional[str],
631 Doc(
632 """
633 Unique operation identifier for OpenAPI.
634 Example: 'users.get_by_id'
635 """
636 ),
637 ] = None,
638 deprecated: Annotated[
639 bool,
640 Doc(
641 """
642 Mark endpoint as deprecated in docs.
643 Example: True
644 """
645 ),
646 ] = False,
647 parameters: Annotated[
648 List[Parameter],
649 Doc(
650 """
651 Additional OpenAPI parameter definitions.
652 Example: [Parameter(name="fields", in_="query", description="Fields to include")]
653 """
654 ),
655 ] = [],
656 exclude_from_schema: Annotated[
657 bool,
658 Doc(
659 """
660 Exclude this route from OpenAPI docs.
661 Example: True for internal endpoints
662 """
663 ),
664 ] = False,
665 **kwargs: Annotated[
666 Dict[str, Any],
667 Doc(
668 """
669 Additional route metadata.
670 Example: {"x-internal": True}
671 """
672 ),
673 ],
674 ) -> Callable[..., Any]:
675 """
676 Register a GET endpoint with comprehensive OpenAPI support.
678 Examples:
679 1. Basic GET endpoint:
680 @router.get("/users")
681 async def get_users(request: Request, response: Response):
682 users = await get_all_users()
683 return response.json(users)
685 2. GET with path parameter and response model:
686 @router.get(
687 "/users/{user_id}",
688 responses={
689 200: UserResponse,
690 404: {"description": "User not found"}
691 }
692 )
693 async def get_user(request: Request, response: Response):
694 user_id = request.path_params['user_id']
695 user = await get_user_by_id(user_id)
696 if not user:
697 return response.status(404).json({"error": "User not found"})
698 return response.json(user)
700 3. GET with query parameters:
701 class UserQuery(BaseModel):
702 active: bool = True
703 limit: int = 100
705 @router.get("/users/search", request_model=UserQuery)
706 async def search_users(request: Request, response: Response):
707 query = request.query_params
708 users = await search_users(
709 active=query['active'],
710 limit=query['limit']
711 )
712 return response.json(users)
713 """
715 return self.route(
716 path=path,
717 handler=handler,
718 methods=["GET"],
719 name=name,
720 summary=summary,
721 description=description,
722 responses=responses,
723 request_model=request_model,
724 middlewares=middlewares,
725 tags=tags,
726 security=security,
727 operation_id=operation_id,
728 deprecated=deprecated,
729 parameters=parameters,
730 exclude_from_schema=exclude_from_schema,
731 **kwargs,
732 )
734 def post(
735 self,
736 path: Annotated[
737 str,
738 Doc(
739 """
740 URL path pattern for the POST endpoint.
741 Example: '/api/v1/users'
742 """
743 ),
744 ],
745 handler: Annotated[
746 Optional[HandlerType],
747 Doc(
748 """
749 Async handler function for POST requests.
750 Example:
751 async def create_user(request, response):
752 user_data = request.json()
753 return response.json(user_data, status=201)
754 """
755 ),
756 ] = None,
757 name: Annotated[
758 Optional[str],
759 Doc(
760 """
761 Unique route name for URL generation.
762 Example: 'api-v1-create-user'
763 """
764 ),
765 ] = None,
766 summary: Annotated[
767 Optional[str],
768 Doc(
769 """
770 Brief endpoint summary.
771 Example: 'Create new user'
772 """
773 ),
774 ] = None,
775 description: Annotated[
776 Optional[str],
777 Doc(
778 """
779 Detailed endpoint description.
780 Example: 'Creates new user with provided data'
781 """
782 ),
783 ] = None,
784 responses: Annotated[
785 Optional[Dict[int, Any]],
786 Doc(
787 """
788 Response schemas by status code.
789 Example: {
790 201: UserSchema,
791 400: {"description": "Invalid input"},
792 409: {"description": "User already exists"}
793 }
794 """
795 ),
796 ] = None,
797 request_model: Annotated[
798 Optional[Type[BaseModel]],
799 Doc(
800 """
801 Model for request body validation.
802 Example:
803 class UserCreate(BaseModel):
804 username: str
805 email: EmailStr
806 password: str
807 """
808 ),
809 ] = None,
810 middlewares: Annotated[
811 List[Any],
812 Doc(
813 """
814 Route-specific middleware.
815 Example: [rate_limit(10), validate_content_type('json')]
816 """
817 ),
818 ] = [],
819 tags: Annotated[
820 Optional[List[str]],
821 Doc(
822 """
823 OpenAPI tags for grouping.
824 Example: ["User Management"]
825 """
826 ),
827 ] = None,
828 security: Annotated[
829 Optional[List[Dict[str, List[str]]]],
830 Doc(
831 """
832 Security requirements.
833 Example: [{"BearerAuth": []}]
834 """
835 ),
836 ] = None,
837 operation_id: Annotated[
838 Optional[str],
839 Doc(
840 """
841 Unique operation ID.
842 Example: 'createUser'
843 """
844 ),
845 ] = None,
846 deprecated: Annotated[
847 bool,
848 Doc(
849 """
850 Mark as deprecated.
851 Example: False
852 """
853 ),
854 ] = False,
855 parameters: Annotated[
856 List[Parameter],
857 Doc(
858 """
859 Additional parameters.
860 Example: [Parameter(name="X-Request-ID", in_="header")]
861 """
862 ),
863 ] = [],
864 exclude_from_schema: Annotated[
865 bool,
866 Doc(
867 """
868 Hide from OpenAPI docs.
869 Example: False
870 """
871 ),
872 ] = False,
873 **kwargs: Annotated[
874 Dict[str, Any],
875 Doc(
876 """
877 Additional metadata.
878 Example: {"x-audit-log": True}
879 """
880 ),
881 ],
882 ) -> Callable[..., Any]:
883 """
884 Register a POST endpoint with the application.
886 Examples:
887 1. Simple POST endpoint:
888 @router.post("/messages")
889 async def create_message(request, response):
890 message = await Message.create(**request.json())
891 return response.json(message, status=201)
893 2. POST with request validation:
894 class ProductCreate(BaseModel):
895 name: str
896 price: float
897 category: str
899 @router.post(
900 "/products",
901 request_model=ProductCreate,
902 responses={201: ProductSchema}
903 )
904 async def create_product(request, response):
905 product = await Product.create(**request.validated_data)
906 return response.json(product, status=201)
908 3. POST with file upload:
909 @router.post("/upload")
910 async def upload_file(request, response):
911 file = request.files.get('file')
912 # Process file upload
913 return response.json({"filename": file.filename})
914 """
915 return self.route(
916 path=path,
917 methods=["POST"],
918 handler=handler,
919 name=name,
920 summary=summary,
921 description=description,
922 responses=responses,
923 request_model=request_model,
924 middlewares=middlewares,
925 tags=tags,
926 security=security,
927 operation_id=operation_id,
928 deprecated=deprecated,
929 parameters=parameters,
930 exclude_from_schema=exclude_from_schema,
931 **kwargs,
932 )
934 def delete(
935 self,
936 path: Annotated[
937 str,
938 Doc(
939 """
940 URL path pattern for the DELETE endpoint.
941 Example: '/api/v1/users/{id}'
942 """
943 ),
944 ],
945 handler: Annotated[
946 Optional[HandlerType],
947 Doc(
948 """
949 Async handler function for DELETE requests.
950 Example:
951 async def delete_user(request, response):
952 user_id = request.path_params['id']
953 return response.json({"deleted": user_id})
954 """
955 ),
956 ] = None,
957 name: Annotated[
958 Optional[str],
959 Doc(
960 """
961 Unique route name for URL generation.
962 Example: 'api-v1-delete-user'
963 """
964 ),
965 ] = None,
966 summary: Annotated[
967 Optional[str],
968 Doc(
969 """
970 Brief endpoint summary.
971 Example: 'Delete user account'
972 """
973 ),
974 ] = None,
975 description: Annotated[
976 Optional[str],
977 Doc(
978 """
979 Detailed endpoint description.
980 Example: 'Permanently deletes user account and all associated data'
981 """
982 ),
983 ] = None,
984 responses: Annotated[
985 Optional[Dict[int, Any]],
986 Doc(
987 """
988 Response schemas by status code.
989 Example: {
990 204: None,
991 404: {"description": "User not found"},
992 403: {"description": "Forbidden"}
993 }
994 """
995 ),
996 ] = None,
997 request_model: Annotated[
998 Optional[Type[BaseModel]],
999 Doc(
1000 """
1001 Model for request validation.
1002 Example:
1003 class DeleteConfirmation(BaseModel):
1004 confirm: bool
1005 """
1006 ),
1007 ] = None,
1008 middlewares: Annotated[
1009 List[Any],
1010 Doc(
1011 """
1012 Route-specific middleware.
1013 Example: [admin_required, confirm_action]
1014 """
1015 ),
1016 ] = [],
1017 tags: Annotated[
1018 Optional[List[str]],
1019 Doc(
1020 """
1021 OpenAPI tags for grouping.
1022 Example: ["User Management"]
1023 """
1024 ),
1025 ] = None,
1026 security: Annotated[
1027 Optional[List[Dict[str, List[str]]]],
1028 Doc(
1029 """
1030 Security requirements.
1031 Example: [{"BearerAuth": []}]
1032 """
1033 ),
1034 ] = None,
1035 operation_id: Annotated[
1036 Optional[str],
1037 Doc(
1038 """
1039 Unique operation ID.
1040 Example: 'deleteUser'
1041 """
1042 ),
1043 ] = None,
1044 deprecated: Annotated[
1045 bool,
1046 Doc(
1047 """
1048 Mark as deprecated.
1049 Example: False
1050 """
1051 ),
1052 ] = False,
1053 parameters: Annotated[
1054 List[Parameter],
1055 Doc(
1056 """
1057 Additional parameters.
1058 Example: [Parameter(name="confirm", in_="query")]
1059 """
1060 ),
1061 ] = [],
1062 exclude_from_schema: Annotated[
1063 bool,
1064 Doc(
1065 """
1066 Hide from OpenAPI docs.
1067 Example: False
1068 """
1069 ),
1070 ] = False,
1071 **kwargs: Annotated[
1072 Dict[str, Any],
1073 Doc(
1074 """
1075 Additional metadata.
1076 Example: {"x-destructive": True}
1077 """
1078 ),
1079 ],
1080 ) -> Callable[..., Any]:
1081 """
1082 Register a DELETE endpoint with the application.
1084 Examples:
1085 1. Simple DELETE endpoint:
1086 @router.delete("/users/{id}")
1087 async def delete_user(request, response):
1088 await User.delete(request.path_params['id'])
1089 return response.status(204)
1091 2. DELETE with confirmation:
1092 @router.delete(
1093 "/account",
1094 responses={
1095 204: None,
1096 400: {"description": "Confirmation required"}
1097 }
1098 )
1099 async def delete_account(request, response):
1100 if not request.query_params.get('confirm'):
1101 return response.status(400)
1102 await request.user.delete()
1103 return response.status(204)
1105 3. Soft DELETE:
1106 @router.delete("/posts/{id}")
1107 async def soft_delete_post(request, response):
1108 await Post.soft_delete(request.path_params['id'])
1109 return response.json({"status": "archived"})
1110 """
1111 return self.route(
1112 path=path,
1113 methods=["DELETE"],
1114 handler=handler,
1115 name=name,
1116 summary=summary,
1117 description=description,
1118 responses=responses,
1119 request_model=request_model,
1120 middlewares=middlewares,
1121 tags=tags,
1122 security=security,
1123 operation_id=operation_id,
1124 deprecated=deprecated,
1125 parameters=parameters,
1126 exclude_from_schema=exclude_from_schema,
1127 **kwargs,
1128 )
1130 def put(
1131 self,
1132 path: Annotated[
1133 str,
1134 Doc(
1135 """
1136 URL path pattern for the PUT endpoint.
1137 Example: '/api/v1/users/{id}'
1138 """
1139 ),
1140 ],
1141 handler: Annotated[
1142 Optional[HandlerType],
1143 Doc(
1144 """
1145 Async handler function for PUT requests.
1146 Example:
1147 async def update_user(request, response):
1148 user_id = request.path_params['id']
1149 return response.json({"updated": user_id})
1150 """
1151 ),
1152 ] = None,
1153 name: Annotated[
1154 Optional[str],
1155 Doc(
1156 """
1157 Unique route name for URL generation.
1158 Example: 'api-v1-update-user'
1159 """
1160 ),
1161 ] = None,
1162 summary: Annotated[
1163 Optional[str],
1164 Doc(
1165 """
1166 Brief endpoint summary.
1167 Example: 'Update user details'
1168 """
1169 ),
1170 ] = None,
1171 description: Annotated[
1172 Optional[str],
1173 Doc(
1174 """
1175 Detailed endpoint description.
1176 Example: 'Full update of user resource'
1177 """
1178 ),
1179 ] = None,
1180 responses: Annotated[
1181 Optional[Dict[int, Any]],
1182 Doc(
1183 """
1184 Response schemas by status code.
1185 Example: {
1186 200: UserSchema,
1187 400: {"description": "Invalid input"},
1188 404: {"description": "User not found"}
1189 }
1190 """
1191 ),
1192 ] = None,
1193 request_model: Annotated[
1194 Optional[Type[BaseModel]],
1195 Doc(
1196 """
1197 Model for request body validation.
1198 Example:
1199 class UserUpdate(BaseModel):
1200 email: Optional[EmailStr]
1201 password: Optional[str]
1202 """
1203 ),
1204 ] = None,
1205 middlewares: Annotated[
1206 List[Any],
1207 Doc(
1208 """
1209 Route-specific middleware.
1210 Example: [owner_required, validate_etag]
1211 """
1212 ),
1213 ] = [],
1214 tags: Annotated[
1215 Optional[List[str]],
1216 Doc(
1217 """
1218 OpenAPI tags for grouping.
1219 Example: ["User Management"]
1220 """
1221 ),
1222 ] = None,
1223 security: Annotated[
1224 Optional[List[Dict[str, List[str]]]],
1225 Doc(
1226 """
1227 Security requirements.
1228 Example: [{"BearerAuth": []}]
1229 """
1230 ),
1231 ] = None,
1232 operation_id: Annotated[
1233 Optional[str],
1234 Doc(
1235 """
1236 Unique operation ID.
1237 Example: 'updateUser'
1238 """
1239 ),
1240 ] = None,
1241 deprecated: Annotated[
1242 bool,
1243 Doc(
1244 """
1245 Mark as deprecated.
1246 Example: False
1247 """
1248 ),
1249 ] = False,
1250 parameters: Annotated[
1251 List[Parameter],
1252 Doc(
1253 """
1254 Additional parameters.
1255 Example: [Parameter(name="If-Match", in_="header")]
1256 """
1257 ),
1258 ] = [],
1259 exclude_from_schema: Annotated[
1260 bool,
1261 Doc(
1262 """
1263 Hide from OpenAPI docs.
1264 Example: False
1265 """
1266 ),
1267 ] = False,
1268 **kwargs: Annotated[
1269 Dict[str, Any],
1270 Doc(
1271 """
1272 Additional metadata.
1273 Example: {"x-idempotent": True}
1274 """
1275 ),
1276 ],
1277 ) -> Callable[..., Any]:
1278 """
1279 Register a PUT endpoint with the application.
1281 Examples:
1282 1. Simple PUT endpoint:
1283 @router.put("/users/{id}")
1284 async def update_user(request, response):
1285 user_id = request.path_params['id']
1286 await User.update(user_id, **request.json())
1287 return response.json({"status": "updated"})
1289 2. PUT with full resource replacement:
1290 @router.put(
1291 "/articles/{slug}",
1292 request_model=ArticleUpdate,
1293 responses={
1294 200: ArticleSchema,
1295 404: {"description": "Article not found"}
1296 }
1297 )
1298 async def replace_article(request, response):
1299 article = await Article.replace(
1300 request.path_params['slug'],
1301 request.validated_data
1302 )
1303 return response.json(article)
1305 3. PUT with conditional update:
1306 @router.put("/resources/{id}")
1307 async def update_resource(request, response):
1308 if request.headers.get('If-Match') != expected_etag:
1309 return response.status(412)
1310 # Process update
1311 return response.json({"status": "success"})
1312 """
1313 return self.route(
1314 path=path,
1315 methods=["PUT"],
1316 handler=handler,
1317 name=name,
1318 summary=summary,
1319 description=description,
1320 responses=responses,
1321 request_model=request_model,
1322 middlewares=middlewares,
1323 tags=tags,
1324 security=security,
1325 operation_id=operation_id,
1326 deprecated=deprecated,
1327 parameters=parameters,
1328 exclude_from_schema=exclude_from_schema,
1329 **kwargs,
1330 )
1332 def patch(
1333 self,
1334 path: Annotated[
1335 str,
1336 Doc(
1337 """
1338 URL path pattern for the PATCH endpoint.
1339 Example: '/api/v1/users/{id}'
1340 """
1341 ),
1342 ],
1343 handler: Annotated[
1344 Optional[HandlerType],
1345 Doc(
1346 """
1347 Async handler function for PATCH requests.
1348 Example:
1349 async def partial_update_user(request, response):
1350 user_id = request.path_params['id']
1351 return response.json({"updated": user_id})
1352 """
1353 ),
1354 ] = None,
1355 name: Annotated[
1356 Optional[str],
1357 Doc(
1358 """
1359 Unique route name for URL generation.
1360 Example: 'api-v1-partial-update-user'
1361 """
1362 ),
1363 ] = None,
1364 summary: Annotated[
1365 Optional[str],
1366 Doc(
1367 """
1368 Brief endpoint summary.
1369 Example: 'Partially update user details'
1370 """
1371 ),
1372 ] = None,
1373 description: Annotated[
1374 Optional[str],
1375 Doc(
1376 """
1377 Detailed endpoint description.
1378 Example: 'Partial update of user resource'
1379 """
1380 ),
1381 ] = None,
1382 responses: Annotated[
1383 Optional[Dict[int, Any]],
1384 Doc(
1385 """
1386 Response schemas by status code.
1387 Example: {
1388 200: UserSchema,
1389 400: {"description": "Invalid input"},
1390 404: {"description": "User not found"}
1391 }
1392 """
1393 ),
1394 ] = None,
1395 request_model: Annotated[
1396 Optional[Type[BaseModel]],
1397 Doc(
1398 """
1399 Model for request body validation.
1400 Example:
1401 class UserPatch(BaseModel):
1402 email: Optional[EmailStr] = None
1403 password: Optional[str] = None
1404 """
1405 ),
1406 ] = None,
1407 middlewares: Annotated[
1408 List[Any],
1409 Doc(
1410 """
1411 Route-specific middleware.
1412 Example: [owner_required, validate_patch]
1413 """
1414 ),
1415 ] = [],
1416 tags: Annotated[
1417 Optional[List[str]],
1418 Doc(
1419 """
1420 OpenAPI tags for grouping.
1421 Example: ["User Management"]
1422 """
1423 ),
1424 ] = None,
1425 security: Annotated[
1426 Optional[List[Dict[str, List[str]]]],
1427 Doc(
1428 """
1429 Security requirements.
1430 Example: [{"BearerAuth": []}]
1431 """
1432 ),
1433 ] = None,
1434 operation_id: Annotated[
1435 Optional[str],
1436 Doc(
1437 """
1438 Unique operation ID.
1439 Example: 'partialUpdateUser'
1440 """
1441 ),
1442 ] = None,
1443 deprecated: Annotated[
1444 bool,
1445 Doc(
1446 """
1447 Mark as deprecated.
1448 Example: False
1449 """
1450 ),
1451 ] = False,
1452 parameters: Annotated[
1453 List[Parameter],
1454 Doc(
1455 """
1456 Additional parameters.
1457 Example: [Parameter(name="fields", in_="query")]
1458 """
1459 ),
1460 ] = [],
1461 exclude_from_schema: Annotated[
1462 bool,
1463 Doc(
1464 """
1465 Hide from OpenAPI docs.
1466 Example: False
1467 """
1468 ),
1469 ] = False,
1470 **kwargs: Annotated[
1471 Dict[str, Any],
1472 Doc(
1473 """
1474 Additional metadata.
1475 Example: {"x-partial-update": True}
1476 """
1477 ),
1478 ],
1479 ) -> Callable[..., Any]:
1480 """
1481 Register a PATCH endpoint with the application.
1483 Examples:
1484 1. Simple PATCH endpoint:
1485 @router.patch("/users/{id}")
1486 async def update_user(request, response):
1487 user_id = request.path_params['id']
1488 await User.partial_update(user_id, **request.json())
1489 return response.json({"status": "updated"})
1491 2. PATCH with JSON Merge Patch:
1492 @router.patch(
1493 "/articles/{id}",
1494 request_model=ArticlePatch,
1495 responses={200: ArticleSchema}
1496 )
1497 async def patch_article(request, response):
1498 article = await Article.patch(
1499 request.path_params['id'],
1500 request.validated_data
1501 )
1502 return response.json(article)
1504 3. PATCH with selective fields:
1505 @router.patch("/profile")
1506 async def update_profile(request, response):
1507 allowed_fields = {'bio', 'avatar_url'}
1508 updates = {k: v for k, v in request.json().items()
1509 if k in allowed_fields}
1510 await Profile.update(request.user.id, **updates)
1511 return response.json(updates)
1512 """
1513 return self.route(
1514 path=path,
1515 methods=["PATCH"],
1516 handler=handler,
1517 name=name,
1518 summary=summary,
1519 description=description,
1520 responses=responses,
1521 request_model=request_model,
1522 middlewares=middlewares,
1523 tags=tags,
1524 security=security,
1525 operation_id=operation_id,
1526 deprecated=deprecated,
1527 parameters=parameters,
1528 exclude_from_schema=exclude_from_schema,
1529 **kwargs,
1530 )
1532 def options(
1533 self,
1534 path: Annotated[
1535 str,
1536 Doc(
1537 """
1538 URL path pattern for the OPTIONS endpoint.
1539 Example: '/api/v1/users'
1540 """
1541 ),
1542 ],
1543 handler: Annotated[
1544 Optional[HandlerType],
1545 Doc(
1546 """
1547 Async handler function for OPTIONS requests.
1548 Example:
1549 async def user_options(request, response):
1550 response.headers['Allow'] = 'GET, POST, OPTIONS'
1551 return response
1552 """
1553 ),
1554 ] = None,
1555 name: Annotated[
1556 Optional[str],
1557 Doc(
1558 """
1559 Unique route name for URL generation.
1560 Example: 'api-v1-user-options'
1561 """
1562 ),
1563 ] = None,
1564 summary: Annotated[
1565 Optional[str],
1566 Doc(
1567 """
1568 Brief endpoint summary.
1569 Example: 'Get supported operations'
1570 """
1571 ),
1572 ] = None,
1573 description: Annotated[
1574 Optional[str],
1575 Doc(
1576 """
1577 Detailed endpoint description.
1578 Example: 'Returns supported HTTP methods and CORS headers'
1579 """
1580 ),
1581 ] = None,
1582 responses: Annotated[
1583 Optional[Dict[int, Any]],
1584 Doc(
1585 """
1586 Response schemas by status code.
1587 Example: {
1588 200: None,
1589 204: None
1590 }
1591 """
1592 ),
1593 ] = None,
1594 request_model: Annotated[
1595 Optional[Type[BaseModel]],
1596 Doc(
1597 """
1598 Model for request validation.
1599 Example:
1600 class OptionsQuery(BaseModel):
1601 detailed: bool = False
1602 """
1603 ),
1604 ] = None,
1605 middlewares: Annotated[
1606 List[Any],
1607 Doc(
1608 """
1609 Route-specific middleware.
1610 Example: [cors_middleware]
1611 """
1612 ),
1613 ] = [],
1614 tags: Annotated[
1615 Optional[List[str]],
1616 Doc(
1617 """
1618 OpenAPI tags for grouping.
1619 Example: ["CORS"]
1620 """
1621 ),
1622 ] = None,
1623 security: Annotated[
1624 Optional[List[Dict[str, List[str]]]],
1625 Doc(
1626 """
1627 Security requirements.
1628 Example: []
1629 """
1630 ),
1631 ] = None,
1632 operation_id: Annotated[
1633 Optional[str],
1634 Doc(
1635 """
1636 Unique operation ID.
1637 Example: 'userOptions'
1638 """
1639 ),
1640 ] = None,
1641 deprecated: Annotated[
1642 bool,
1643 Doc(
1644 """
1645 Mark as deprecated.
1646 Example: False
1647 """
1648 ),
1649 ] = False,
1650 parameters: Annotated[
1651 List[Parameter],
1652 Doc(
1653 """
1654 Additional parameters.
1655 Example: [Parameter(name="Origin", in_="header")]
1656 """
1657 ),
1658 ] = [],
1659 exclude_from_schema: Annotated[
1660 bool,
1661 Doc(
1662 """
1663 Hide from OpenAPI docs.
1664 Example: True
1665 """
1666 ),
1667 ] = False,
1668 **kwargs: Annotated[
1669 Dict[str, Any],
1670 Doc(
1671 """
1672 Additional metadata.
1673 Example: {"x-cors": True}
1674 """
1675 ),
1676 ],
1677 ) -> Callable[..., Any]:
1678 """
1679 Register an OPTIONS endpoint with the application.
1681 Examples:
1682 1. Simple OPTIONS endpoint:
1683 @router.options("/users")
1684 async def user_options(request, response):
1685 response.headers['Allow'] = 'GET, POST, OPTIONS'
1686 return response
1688 2. CORS OPTIONS handler:
1689 @router.options("/{path:path}")
1690 async def cors_options(request, response):
1691 response.headers.update({
1692 'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE',
1693 'Access-Control-Allow-Headers': 'Content-Type',
1694 'Access-Control-Max-Age': '86400'
1695 })
1696 return response.status(204)
1698 3. Detailed OPTIONS response:
1699 @router.options("/resources")
1700 async def resource_options(request, response):
1701 return response.json({
1702 "methods": ["GET", "POST"],
1703 "formats": ["application/json"],
1704 "limits": {"max_size": "10MB"}
1705 })
1706 """
1707 return self.route(
1708 path=path,
1709 methods=["OPTIONS"],
1710 handler=handler,
1711 name=name,
1712 summary=summary,
1713 description=description,
1714 responses=responses,
1715 request_model=request_model,
1716 middlewares=middlewares,
1717 tags=tags,
1718 security=security,
1719 operation_id=operation_id,
1720 deprecated=deprecated,
1721 parameters=parameters,
1722 exclude_from_schema=exclude_from_schema,
1723 **kwargs,
1724 )
1726 def head(
1727 self,
1728 path: Annotated[
1729 str,
1730 Doc(
1731 """
1732 URL path pattern for the HEAD endpoint.
1733 Example: '/api/v1/resources/{id}'
1734 """
1735 ),
1736 ],
1737 handler: Annotated[
1738 Optional[HandlerType],
1739 Doc(
1740 """
1741 Async handler function for HEAD requests.
1742 Example:
1743 async def check_resource(request, response):
1744 exists = await Resource.exists(request.path_params['id'])
1745 return response.status(200 if exists else 404)
1746 """
1747 ),
1748 ] = None,
1749 name: Annotated[
1750 Optional[str],
1751 Doc(
1752 """
1753 Unique route name for URL generation.
1754 Example: 'api-v1-check-resource'
1755 """
1756 ),
1757 ] = None,
1758 summary: Annotated[
1759 Optional[str],
1760 Doc(
1761 """
1762 Brief endpoint summary.
1763 Example: 'Check resource existence'
1764 """
1765 ),
1766 ] = None,
1767 description: Annotated[
1768 Optional[str],
1769 Doc(
1770 """
1771 Detailed endpoint description.
1772 Example: 'Returns headers only to check if resource exists'
1773 """
1774 ),
1775 ] = None,
1776 responses: Annotated[
1777 Optional[Dict[int, Any]],
1778 Doc(
1779 """
1780 Response schemas by status code.
1781 Example: {
1782 200: None,
1783 404: None
1784 }
1785 """
1786 ),
1787 ] = None,
1788 request_model: Annotated[
1789 Optional[Type[BaseModel]],
1790 Doc(
1791 """
1792 Model for request validation.
1793 Example:
1794 class ResourceCheck(BaseModel):
1795 check_children: bool = False
1796 """
1797 ),
1798 ] = None,
1799 middlewares: Annotated[
1800 List[Any],
1801 Doc(
1802 """
1803 Route-specific middleware.
1804 Example: [cache_control('public')]
1805 """
1806 ),
1807 ] = [],
1808 tags: Annotated[
1809 Optional[List[str]],
1810 Doc(
1811 """
1812 OpenAPI tags for grouping.
1813 Example: ["Resource Management"]
1814 """
1815 ),
1816 ] = None,
1817 security: Annotated[
1818 Optional[List[Dict[str, List[str]]]],
1819 Doc(
1820 """
1821 Security requirements.
1822 Example: [{"ApiKeyAuth": []}]
1823 """
1824 ),
1825 ] = None,
1826 operation_id: Annotated[
1827 Optional[str],
1828 Doc(
1829 """
1830 Unique operation ID.
1831 Example: 'checkResource'
1832 """
1833 ),
1834 ] = None,
1835 deprecated: Annotated[
1836 bool,
1837 Doc(
1838 """
1839 Mark as deprecated.
1840 Example: False
1841 """
1842 ),
1843 ] = False,
1844 parameters: Annotated[
1845 List[Parameter],
1846 Doc(
1847 """
1848 Additional parameters.
1849 Example: [Parameter(name="X-Check-Type", in_="header")]
1850 """
1851 ),
1852 ] = [],
1853 exclude_from_schema: Annotated[
1854 bool,
1855 Doc(
1856 """
1857 Hide from OpenAPI docs.
1858 Example: False
1859 """
1860 ),
1861 ] = False,
1862 **kwargs: Annotated[
1863 Dict[str, Any],
1864 Doc(
1865 """
1866 Additional metadata.
1867 Example: {"x-head-only": True}
1868 """
1869 ),
1870 ],
1871 ) -> Callable[..., Any]:
1872 """
1873 Register a HEAD endpoint with the application.
1875 Examples:
1876 1. Simple HEAD endpoint:
1877 @router.head("/resources/{id}")
1878 async def check_resource(request, response):
1879 exists = await Resource.exists(request.path_params['id'])
1880 return response.status(200 if exists else 404)
1882 2. HEAD with cache headers:
1883 @router.head("/static/{path:path}")
1884 async def check_static(request, response):
1885 path = request.path_params['path']
1886 if not static_file_exists(path):
1887 return response.status(404)
1888 response.headers['Last-Modified'] = get_last_modified(path)
1889 return response.status(200)
1891 3. HEAD with metadata:
1892 @router.head("/documents/{id}")
1893 async def document_metadata(request, response):
1894 doc = await Document.metadata(request.path_params['id'])
1895 if not doc:
1896 return response.status(404)
1897 response.headers['X-Document-Size'] = str(doc.size)
1898 return response.status(200)
1899 """
1900 return self.route(
1901 path=path,
1902 methods=["HEAD"],
1903 handler=handler,
1904 name=name,
1905 summary=summary,
1906 description=description,
1907 responses=responses,
1908 request_model=request_model,
1909 middlewares=middlewares,
1910 tags=tags,
1911 security=security,
1912 operation_id=operation_id,
1913 deprecated=deprecated,
1914 parameters=parameters,
1915 exclude_from_schema=exclude_from_schema,
1916 **kwargs,
1917 )
1919 def add_route(
1920 self,
1921 route: Annotated[
1922 Routes, Doc("An instance of the Routes class representing an HTTP route.")
1923 ],
1924 ) -> None:
1925 """
1926 Adds an HTTP route to the application.
1928 This method registers an HTTP route, allowing the application to handle requests for a specific URL path.
1930 Args:
1931 route (Routes): The HTTP route configuration.
1933 Returns:
1934 None
1936 Example:
1937 ```python
1938 route = Routes("/home", home_handler, methods=["GET", "POST"])
1939 app.add_route(route)
1940 ```
1941 """
1943 self.router.add_route(route)
1945 def add_exception_handler(
1946 self,
1947 exc_class_or_status_code: Union[Type[Exception], int],
1948 handler: Optional[ExceptionHandlerType] = None,
1949 ) -> Any:
1950 if handler is None:
1951 # If handler is not given yet, return a decorator
1952 def decorator(func: ExceptionHandlerType) -> Any:
1953 self.exceptions_handler.add_exception_handler(
1954 exc_class_or_status_code, func
1955 )
1956 return func
1958 return decorator
1959 else:
1960 # Normal direct handler registration
1961 self.exceptions_handler.add_exception_handler(
1962 exc_class_or_status_code, handler
1963 )
1965 def url_for(self, _name: str, **path_params: Dict[str, Any]) -> URLPath:
1966 return self.router.url_for(_name, **path_params)
1968 def wrap_asgi(
1969 self,
1970 middleware_cls: Annotated[
1971 Callable[[ASGIApp], Any],
1972 Doc(
1973 "An ASGI middleware class or callable that takes an app as its first argument and returns an ASGI app"
1974 ),
1975 ],
1976 **kwargs: Dict[str, Any],
1977 ) -> None:
1978 """
1979 Wraps the entire application with an ASGI middleware.
1981 This method allows adding middleware at the ASGI level, which intercepts all requests
1982 (HTTP, WebSocket, and Lifespan) before they reach the application.
1984 Args:
1985 middleware_cls: An ASGI middleware class or callable that follows the ASGI interface
1986 *args: Additional positional arguments to pass to the middleware
1987 **kwargs: Additional keyword arguments to pass to the middleware
1989 Returns:
1990 NexiosApp: The application instance for method chaining
1993 """
1994 self.app = middleware_cls(self.app, **kwargs)
1995 return None
1997 def get_all_routes(self) -> List[Routes]:
1998 """
1999 Returns all routes registered in the application.
2001 This method retrieves a list of all HTTP and WebSocket routes defined in the application.
2003 Returns:
2004 List[Routes]: A list of all registered routes.
2006 Example:
2007 ```python
2008 routes = app.get_all_routes()
2009 for route in routes:
2010 print(route.path, route.methods)
2011 ```
2012 """
2013 return self.router.get_all_routes()
2015 def ws_route(
2016 self,
2017 path: Annotated[
2018 str,
2019 Doc(
2020 """
2021 URL path pattern for the WebSocket route.
2022 Example: '/ws/chat/{room_id}'
2023 """
2024 ),
2025 ],
2026 handler: Annotated[
2027 Optional[WsHandlerType],
2028 Doc(
2029 """
2030 Async handler function for WebSocket connections.
2031 Example:
2032 async def chat_handler(websocket, path):
2033 await websocket.send("Welcome to the chat!")
2034 """
2035 ),
2036 ] = None,
2037 ):
2038 """
2039 Register a WebSocket route with the application.
2041 Args:
2042 path (str): URL path pattern for the WebSocket route.
2043 handler (Callable): Async handler function for WebSocket connections.
2044 Example: async def chat_handler(websocket, path): pass
2046 Returns:
2047 Callable: A decorator to register the WebSocket route.
2048 """
2049 return self.ws_router.ws_route(
2050 path=path,
2051 handler=handler,
2052 )
2054 def __str__(self) -> str:
2055 return f"<NexiosApp: {self.title}>"