Coverage for nexios\routing.py: 64%
433 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-21 20:31 +0100
1from __future__ import annotations
2import inspect
3from typing import (
4 Any,
5 List,
6 Optional,
7 Pattern,
8 Dict,
9 TypeVar,
10 Tuple,
11 Callable,
12 Union,
13 Type,
14)
15from dataclasses import dataclass
16import re
17import copy
18import warnings, typing
19from enum import Enum
20from abc import abstractmethod, ABC
21import asyncio
22from nexios.events import AsyncEventEmitter
23from nexios.openapi.models import Parameter
24from nexios.types import MiddlewareType, WsMiddlewareType, HandlerType, WsHandlerType
25from nexios.decorators import allowed_methods
26from typing_extensions import Doc, Annotated # type: ignore
27from nexios.structs import URLPath, RouteParam
28from nexios.http import Request, Response
29from nexios.http.response import JSONResponse
30from nexios.types import Scope, Send, Receive, ASGIApp
31from .converters import Convertor, CONVERTOR_TYPES, get_route_path
32from nexios.websockets import WebSocket
33from nexios.middlewares.core import BaseMiddleware
34from nexios.middlewares.core import Middleware, wrap_middleware
35from nexios.exceptions import NotFoundException
36from nexios.websockets.errors import WebSocketErrorMiddleware
37from pydantic import BaseModel
38from nexios.http.response import BaseResponse
39from nexios.dependencies import inject_dependencies
42T = TypeVar("T")
43allowed_methods_default = ["get", "post", "delete", "put", "patch", "options"]
46async def request_response(
47 func: typing.Callable[[Request, Response], typing.Awaitable[Response]],
48) -> ASGIApp:
49 """
50 Takes a function or coroutine `func(request) -> response`,
51 and returns an ASGI application.
52 """
53 assert asyncio.iscoroutinefunction(func), "Endpoints must be async"
55 async def app(scope: Scope, receive: Receive, send: Send) -> None:
56 request = Request(scope, receive, send)
57 response_manager = Response(request)
59 func_result = await func(request, response_manager, **request.path_params)
60 if isinstance(func_result, (dict, list, str)):
61 response_manager.json(func_result)
63 elif isinstance(func_result, BaseResponse):
64 response_manager.make_response(func_result)
65 response = response_manager.get_response()
66 return await response(scope, receive, send)
68 return app
71def websocket_session(
72 func: typing.Callable[[WebSocket], typing.Awaitable[None]],
73) -> ASGIApp:
74 """
75 Takes a coroutine `func(session)`, and returns an ASGI application.
76 """
77 assert asyncio.iscoroutinefunction(func), "WebSocket endpoints must be async"
79 async def app(scope: Scope, receive: Receive, send: Send) -> None:
80 session = WebSocket(scope, receive=receive, send=send)
82 async def app(scope: Scope, receive: Receive, send: Send) -> None:
83 await func(session)
85 # await wrap_app_handling_exceptions(app, session)(scope, receive, send)
86 await app(scope, receive, send)
88 return app
91def replace_params(
92 path: str,
93 param_convertors: dict[str, Convertor[typing.Any]],
94 path_params: dict[str, str],
95) -> tuple[str, dict[str, str]]:
96 for key, value in list(path_params.items()):
97 if "{" + key + "}" in path:
98 convertor = param_convertors[key]
99 value = convertor.to_string(value)
100 path = path.replace("{" + key + "}", value)
101 path_params.pop(key)
102 return path, path_params
105# Match parameters in URL paths, eg. '{param}', and '{param:int}'
106PARAM_REGEX = re.compile("{([a-zA-Z_][a-zA-Z0-9_]*)(:[a-zA-Z_][a-zA-Z0-9_]*)?}")
109def compile_path(
110 path: str,
111) -> tuple[typing.Pattern[str], RouteType, dict[str, Convertor[typing.Any]], List[str]]:
112 """
113 Given a path string, like: "/{username:str}",
114 or a host string, like: "{subdomain}.mydomain.org", return a three-tuple
115 of (regex, format, {param_name:convertor}).
117 regex: "/(?P<username>[^/]+)"
118 format: "/{username}"
119 convertors: {"username": StringConvertor()}
120 """
121 is_host = not path.startswith("/")
123 path_regex = "^"
124 path_format = ""
125 duplicated_params: typing.Set[typing.Any] = set()
127 idx = 0
128 param_convertors = {}
129 param_names: List[str] = []
130 for match in PARAM_REGEX.finditer(path):
131 param_name, convertor_type = match.groups("str")
132 convertor_type = convertor_type.lstrip(":")
133 assert (
134 convertor_type in CONVERTOR_TYPES
135 ), f"Unknown path convertor '{convertor_type}'"
136 convertor = CONVERTOR_TYPES[convertor_type]
138 path_regex += re.escape(path[idx : match.start()])
139 path_regex += f"(?P<{param_name}>{convertor.regex})"
140 path_format += path[idx : match.start()]
141 path_format += "{%s}" % param_name
143 if param_name in param_convertors:
144 duplicated_params.add(param_name)
146 param_convertors[param_name] = convertor
148 idx = match.end()
149 param_names.append(param_name)
151 if duplicated_params:
152 names = ", ".join(sorted(duplicated_params))
153 ending = "s" if len(duplicated_params) > 1 else ""
154 raise ValueError(f"Duplicated param name{ending} {names} at path {path}")
156 if is_host:
157 hostname = path[idx:].split(":")[0]
158 path_regex += re.escape(hostname) + "$"
159 else:
160 path_regex += re.escape(path[idx:]) + "$"
161 path_format += path[idx:]
163 return re.compile(path_regex), path_format, param_convertors, param_names # type: ignore
166class RouteType(Enum):
167 REGEX = "regex"
168 PATH = "path"
169 WILDCARD = "wildcard"
172@dataclass
173class RoutePattern:
174 """Represents a processed route pattern with metadata"""
176 pattern: Pattern[str]
177 raw_path: str
178 param_names: List[str]
179 route_type: RouteType
180 convertor: Dict[str, Convertor[typing.Any]]
183class RouteBuilder:
184 @staticmethod
185 def create_pattern(path: str) -> RoutePattern:
186 path_regex, path_format, param_convertors, param_names = ( # type: ignore
187 compile_path(path)
188 )
189 return RoutePattern(
190 pattern=path_regex,
191 raw_path=path,
192 param_names=param_names,
193 route_type=path_format,
194 convertor=param_convertors,
195 )
198class BaseRouter(ABC):
199 """
200 Base class for routers. This class should not be instantiated directly.
201 Subclasses should implement the `__call__` method to handle specific routing logic.
202 """
204 def __init__(self, prefix: Optional[str] = None):
205 self.prefix = prefix or ""
206 self.routes: List[Any] = []
207 self.middlewares: List[Any] = []
208 self.sub_routers: Dict[str, ASGIApp] = {}
210 if self.prefix and not self.prefix.startswith("/"):
211 warnings.warn("Router prefix should start with '/'")
212 self.prefix = f"/{self.prefix}"
214 @abstractmethod
215 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
216 """
217 Abstract method to handle incoming requests. Subclasses must implement this method.
219 Args:
220 scope: The ASGI scope dictionary.
221 receive: The ASGI receive callable.
222 send: The ASGI send callable.
224 Raises:
225 NotImplementedError: If the method is not implemented by a subclass.
226 """
227 raise NotImplementedError("Subclasses must implement this method")
229 def add_middleware(self, middleware: MiddlewareType) -> None:
230 """
231 Add middleware to the router.
233 Args:
234 middleware: The middleware to add.
235 """
237 self.middlewares.append(middleware)
239 def build_middleware_stack(self, app: ASGIApp) -> ASGIApp:
240 """
241 Builds the middleware stack by applying all registered middlewares to the app.
243 Args:
244 app: The base ASGI application.
246 Returns:
247 ASGIApp: The application wrapped with all middlewares.
248 """
249 for mdw in reversed(self.middlewares):
250 app = mdw(app)
251 return app
253 def mount_router(
254 self, app: Union["Router", "WSRouter"], path: Optional[str] = None
255 ) -> None:
256 """
257 Mount an ASGI application (e.g., another Router) under a specific path prefix.
259 Args:
260 path: The path prefix under which the app will be mounted.
261 app: The ASGI application (e.g., another Router) to mount.
262 """
263 if not path:
264 path = app.prefix
265 path = path.rstrip("/")
267 if path == "":
268 self.sub_routers[path] = app
269 return
270 if not path.startswith("/"):
271 path = f"/{path}"
273 self.sub_routers[path] = app
275 def __repr__(self) -> str:
276 return f"<BaseRouter prefix='{self.prefix}' routes={len(self.routes)}>"
279class Routes:
280 """
281 Encapsulates all routing information for an API endpoint, including path handling,
282 validation, OpenAPI documentation, and request processing.
284 Attributes:
285 raw_path: The original URL path string provided during initialization.
286 pattern: Compiled regex pattern for path matching.
287 handler: Callable that processes incoming requests.
288 methods: List of allowed HTTP methods for this endpoint.
289 validator: Request parameter validation rules.
290 request_schema: Schema for request body documentation.
291 response_schema: Schema for response documentation.
292 deprecated: Deprecation status indicator.
293 tags: OpenAPI documentation tags.
294 description: Endpoint functionality details.
295 summary: Concise endpoint purpose.
296 """
298 def __init__(
299 self,
300 path: Annotated[
301 str,
302 Doc(
303 """
304 URL path pattern for the endpoint. Supports dynamic parameters using curly brace syntax.
305 Examples:
306 - '/users' (static path)
307 - '/posts/{id}' (path parameter)
308 - '/files/{filepath:.*}' (regex-matched path parameter)
309 """
310 ),
311 ],
312 handler: Annotated[
313 Optional[HandlerType],
314 Doc(
315 """
316 Callable responsible for processing requests to this endpoint. Can be:
317 - A regular function
318 - An async function
319 - A class method
320 - Any object implementing __call__
322 The handler should accept a request object and return a response object.
323 Example: def user_handler(request: Request) -> Response: ...
324 """
325 ),
326 ],
327 methods: Annotated[
328 Optional[List[str]],
329 Doc(
330 """
331 HTTP methods allowed for this endpoint. Common methods include:
332 - GET: Retrieve resources
333 - POST: Create resources
334 - PUT: Update resources
335 - DELETE: Remove resources
336 - PATCH: Partial updatess
338 Defaults to ['GET'] if not specified. Use uppercase method names.
339 """
340 ),
341 ] = None,
342 name: Annotated[
343 Optional[str],
344 Doc(
345 """The unique identifier for the route. This name is used to generate
346 URLs dynamically with `url_for`. It should be a valid, unique string
347 that represents the route within the application."""
348 ),
349 ] = None,
350 summary: Annotated[
351 Optional[str],
352 Doc(
353 "A brief summary of the API endpoint. This should be a short, one-line description providing a high-level overview of its purpose."
354 ),
355 ] = None,
356 description: Annotated[
357 Optional[str],
358 Doc(
359 "A detailed explanation of the API endpoint, including functionality, expected behavior, and additional context."
360 ),
361 ] = None,
362 responses: Annotated[
363 Optional[Dict[int, Any]],
364 Doc(
365 "A dictionary mapping HTTP status codes to response schemas or descriptions. Keys are HTTP status codes (e.g., 200, 400), and values define the response format."
366 ),
367 ] = None,
368 request_model: Annotated[
369 Optional[Type[BaseModel]],
370 Doc(
371 "A Pydantic model representing the expected request payload. Defines the structure and validation rules for incoming request data."
372 ),
373 ] = None,
374 tags: Optional[List[str]] = None,
375 security: Optional[List[Dict[str, List[str]]]] = None,
376 operation_id: Optional[str] = None,
377 deprecated: bool = False,
378 parameters: List[Parameter] = [],
379 middlewares: List[Any] = [],
380 exclude_from_schema: bool = False,
381 **kwargs: Dict[str, Any],
382 ):
383 """
384 Initialize a route configuration with endpoint details.
386 Args:
387 path: URL path pattern with optional parameters.
388 handler: Request processing function/method.
389 methods: Allowed HTTP methods (default: ['GET']).
390 validator: Multi-layer request validation rules.
391 request_schema: Request body structure definition.
392 response_schema: Success response structure definition.
393 deprecated: Deprecation marker.
394 tags: Documentation categories.
395 description: Comprehensive endpoint documentation.
396 summary: Brief endpoint description.
398 Raises:
399 AssertionError: If handler is not callable.
400 """
401 assert callable(handler), "Route handler must be callable"
403 self.prefix: Optional[str] = None
404 if path == "":
405 path = "/"
406 self.raw_path = path
407 self.handler = inject_dependencies(handler)
408 self.methods = methods or allowed_methods_default
409 self.name = name
411 self.route_info = RouteBuilder.create_pattern(path)
412 self.pattern: Pattern[str] = self.route_info.pattern
413 self.param_names = self.route_info.param_names
414 self.route_type = self.route_info.route_type
415 self.middlewares: typing.List[MiddlewareType] = list(middlewares)
416 self.summary = summary
417 self.description = description
418 self.responses = responses
419 self.request_model = request_model
420 self.kwargs = kwargs
421 self.tags = tags
422 self.security = security
423 self.operation_id = operation_id
424 self.deprecated = deprecated
425 self.parameters = parameters
426 self.exlude_from_schema = exclude_from_schema
428 def match(self, path: str, method: str) -> typing.Tuple[Any, Any, Any]:
429 """
430 Match a path against this route's pattern and return captured parameters.
432 Args:
433 path: The URL path to match.
435 Returns:
436 Optional[Dict[str, Any]]: A dictionary of captured parameters if the path matches,
437 otherwise None.
438 """
439 match = self.pattern.match(path)
440 if match:
441 matched_params = match.groupdict()
442 for key, value in matched_params.items():
443 matched_params[key] = self.route_info.convertor[ # type: ignore
444 key
445 ].convert(value)
446 is_method_allowed = method.lower() in [m.lower() for m in self.methods]
447 return match, matched_params, is_method_allowed
448 return None, None, False
450 def url_path_for(self, _name: str, **path_params: Any) -> URLPath:
451 """
452 Generate a URL path for the route with the given name and parameters.
454 Args:
455 name: The name of the route.
456 path_params: A dictionary of path parameters to substitute into the route's path.
458 Returns:
459 str: The generated URL path.
461 Raises:
462 ValueError: If the route name does not match or if required parameters are missing.
463 """
464 if _name != self.name:
465 raise ValueError(
466 f"Route name '{_name}' does not match the current route name '{self.name}'."
467 )
469 required_params = set(self.param_names)
470 provided_params = set(path_params.keys())
471 if required_params != provided_params:
472 missing_params = required_params - provided_params
473 extra_params = provided_params - required_params
474 raise ValueError(
475 f"Missing parameters: {missing_params}. Extra parameters: {extra_params}."
476 )
478 path = self.raw_path
479 for param_name, param_value in path_params.items():
480 param_value = str(param_value)
482 path = re.sub(rf"\{ {param_name}(:[^} ]+)?} ", param_value, path)
484 return URLPath(path=path, protocol="http")
486 async def handle(self, scope: Scope, receive: Receive, send: Send) -> Any:
487 """
488 Process an incoming request using the route's handler.
490 Args:
491 request: The incoming HTTP request object.
492 response: The outgoing HTTP response object.
494 Returns:
495 Response: The processed HTTP response object.
496 """
498 async def apply_middlewares(app: ASGIApp) -> ASGIApp:
499 middleware: typing.List[Middleware] = []
500 for mdw in self.middlewares:
501 middleware.append(wrap_middleware(mdw)) # type: ignore
502 for cls, args, kwargs in reversed(middleware):
503 app = cls(app, *args, **kwargs)
504 return app
506 app = await apply_middlewares(await request_response(self.handler))
508 await app(scope, receive, send)
510 def __call__(self) -> Tuple[Pattern[str], HandlerType]:
511 """
512 Return the route components for registration.
514 Returns:
515 Tuple[Pattern[str], HandlerType]: The compiled regex pattern and the handler.
517 """
518 return self.pattern, self.handler
520 def __repr__(self) -> str:
521 """
522 Return a string representation of the route.
524 Returns:
525 str: A string describing the route.
526 """
527 return f"<Route {self.raw_path} methods={self.methods}>"
530class Router(BaseRouter):
531 def __init__(
532 self,
533 prefix: Optional[str] = None,
534 routes: Optional[List[Routes]] = None,
535 tags: Optional[List[str]] = None,
536 exclude_from_schema: bool = False,
537 name: Optional[str] = None,
538 ):
539 self.prefix = prefix or ""
540 self.prefix.rstrip("/")
541 self.routes: List[Routes] = list(routes) if routes else []
542 self.middlewares: typing.List[Middleware] = []
543 self.sub_routers: Dict[str, Router] = {}
544 self.route_class = Routes
545 self.tags = tags or []
546 self.exclude_from_schema = exclude_from_schema
547 self.name = name
548 self.event = AsyncEventEmitter()
550 if self.prefix and not self.prefix.startswith("/"):
551 warnings.warn("Router prefix should start with '/'")
552 self.prefix = f"/{self.prefix}"
554 def build_middleware_stack(self, app: ASGIApp) -> ASGIApp:
555 """
556 Builds the middleware stack by applying all registered middlewares to the app.
558 Args:
559 app: The base ASGI application.
561 Returns:
562 ASGIApp: The application wrapped with all middlewares.
563 """
564 for cls, args, kwargs in reversed(self.middlewares):
565 app = cls(app, *args, **kwargs)
566 return app
568 def add_route(
569 self,
570 route: Annotated[
571 Routes, Doc("An instance of the Routes class representing an HTTP route.")
572 ],
573 ) -> None:
574 """
575 Adds an HTTP route to the application.
577 This method registers an HTTP route, allowing the application to handle requests for a specific URL path.
579 Args:
580 route (Routes): The HTTP route configuration.
582 Returns:
583 None
585 Example:
586 ```python
587 route = Routes("/home", home_handler, methods=["GET", "POST"])
588 app.add_route(route)
589 ```
590 """
591 route.tags = self.tags + route.tags if route.tags else self.tags
592 # original_handler = route.handler
594 if self.exclude_from_schema:
595 route.exlude_from_schema = True
596 original_handler = route.handler
598 async def wrapped_handler(
599 request: Request, response: Response, **kwargs: Dict[str, Any]
600 ):
601 sig = inspect.signature(original_handler)
602 params = list(sig.parameters.keys())
603 handler_args = [request, response]
604 handler_kwargs = {}
605 if len(params) > 2:
606 # Get path parameters from request
607 path_params = request.path_params
609 # For parameters after the first two (request/response)
610 for param in params[2:]:
611 if param in path_params:
612 handler_kwargs[param] = path_params[param]
614 return await original_handler(*handler_args, **handler_kwargs)
616 route.handler = inject_dependencies(wrapped_handler)
618 self.routes.append(route)
620 def add_middleware(self, middleware: MiddlewareType) -> None:
621 """Add middleware to the router"""
622 if callable(middleware):
623 mdw = Middleware(BaseMiddleware, dispatch=middleware) # type: ignore
624 self.middlewares.insert(0, mdw)
626 def get(
627 self,
628 path: Annotated[
629 str,
630 Doc(
631 """
632 URL path pattern for the GET endpoint.
633 Supports path parameters using {param} syntax.
634 Example: '/users/{user_id}'
635 """
636 ),
637 ],
638 handler: Annotated[
639 Optional[HandlerType],
640 Doc(
641 """
642 Async handler function for GET requests.
643 Receives (request, response) and returns response or raw data.
645 Example:
646 async def get_user(request, response):
647 user = await get_user_from_db(request.path_params['user_id'])
648 return response.json(user)
649 """
650 ),
651 ] = None,
652 name: Annotated[
653 Optional[str],
654 Doc(
655 """
656 Unique route identifier for URL generation.
657 Example: 'get-user-by-id'
658 """
659 ),
660 ] = None,
661 summary: Annotated[
662 Optional[str],
663 Doc(
664 """
665 Brief summary for OpenAPI documentation.
666 Example: 'Retrieves a user by ID'
667 """
668 ),
669 ] = None,
670 description: Annotated[
671 Optional[str],
672 Doc(
673 """
674 Detailed description for OpenAPI documentation.
675 Example: 'Returns full user details including profile information'
676 """
677 ),
678 ] = None,
679 responses: Annotated[
680 Optional[Dict[int, Any]],
681 Doc(
682 """
683 Response models by status code.
684 Example:
685 {
686 200: UserSchema,
687 404: {"description": "User not found"},
688 500: {"description": "Server error"}
689 }
690 """
691 ),
692 ] = None,
693 request_model: Annotated[
694 Optional[Type[BaseModel]],
695 Doc(
696 """
697 Pydantic model for request validation (query params).
698 Example:
699 class UserQuery(BaseModel):
700 active_only: bool = True
701 limit: int = 100
702 """
703 ),
704 ] = None,
705 middlewares: Annotated[
706 List[Any],
707 Doc(
708 """
709 List of route-specific middleware functions.
710 Example: [auth_required, rate_limit]
711 """
712 ),
713 ] = [],
714 tags: Annotated[
715 Optional[List[str]],
716 Doc(
717 """
718 OpenAPI tags for grouping related endpoints.
719 Example: ["Users", "Public"]
720 """
721 ),
722 ] = None,
723 security: Annotated[
724 Optional[List[Dict[str, List[str]]]],
725 Doc(
726 """
727 Security requirements for OpenAPI docs.
728 Example: [{"BearerAuth": []}]
729 """
730 ),
731 ] = None,
732 operation_id: Annotated[
733 Optional[str],
734 Doc(
735 """
736 Unique operation identifier for OpenAPI.
737 Example: 'users.get_by_id'
738 """
739 ),
740 ] = None,
741 deprecated: Annotated[
742 bool,
743 Doc(
744 """
745 Mark endpoint as deprecated in docs.
746 Example: True
747 """
748 ),
749 ] = False,
750 parameters: Annotated[
751 List[Parameter],
752 Doc(
753 """
754 Additional OpenAPI parameter definitions.
755 Example: [Parameter(name="fields", in_="query", description="Fields to include")]
756 """
757 ),
758 ] = [],
759 exclude_from_schema: Annotated[
760 bool,
761 Doc(
762 """
763 Exclude this route from OpenAPI docs.
764 Example: True for internal endpoints
765 """
766 ),
767 ] = False,
768 **kwargs: Annotated[
769 Dict[str, Any],
770 Doc(
771 """
772 Additional route metadata.
773 Example: {"x-internal": True}
774 """
775 ),
776 ],
777 ) -> Callable[..., Any]:
778 """
779 Register a GET endpoint with comprehensive OpenAPI support.
781 Examples:
782 1. Basic GET endpoint:
783 @router.get("/users")
784 async def get_users(request: Request, response: Response):
785 users = await get_all_users()
786 return response.json(users)
788 2. GET with path parameter and response model:
789 @router.get(
790 "/users/{user_id}",
791 responses={
792 200: UserResponse,
793 404: {"description": "User not found"}
794 }
795 )
796 async def get_user(request: Request, response: Response):
797 user_id = request.path_params['user_id']
798 user = await get_user_by_id(user_id)
799 if not user:
800 return response.status(404).json({"error": "User not found"})
801 return response.json(user)
803 3. GET with query parameters:
804 class UserQuery(BaseModel):
805 active: bool = True
806 limit: int = 100
808 @router.get("/users/search", request_model=UserQuery)
809 async def search_users(request: Request, response: Response):
810 query = request.query_params
811 users = await search_users(
812 active=query['active'],
813 limit=query['limit']
814 )
815 return response.json(users)
816 """
818 def decorator(handler: HandlerType) -> HandlerType:
819 route = self.route_class(
820 path=path,
821 handler=handler,
822 methods=["GET"],
823 name=name,
824 summary=summary,
825 description=description,
826 responses=responses,
827 request_model=request_model,
828 middlewares=middlewares,
829 tags=tags,
830 security=security,
831 operation_id=operation_id,
832 deprecated=deprecated,
833 parameters=parameters,
834 exclude_from_schema=exclude_from_schema,
835 **kwargs,
836 )
837 self.add_route(route)
838 return handler
840 if handler is None:
841 return decorator
842 return decorator(handler)
844 def post(
845 self,
846 path: Annotated[
847 str,
848 Doc(
849 """
850 URL path pattern for the POST endpoint.
851 Example: '/api/v1/users'
852 """
853 ),
854 ],
855 handler: Annotated[
856 Optional[HandlerType],
857 Doc(
858 """
859 Async handler function for POST requests.
860 Example:
861 async def create_user(request, response):
862 user_data = request.json()
863 return response.json(user_data, status=201)
864 """
865 ),
866 ] = None,
867 name: Annotated[
868 Optional[str],
869 Doc(
870 """
871 Unique route name for URL generation.
872 Example: 'api-v1-create-user'
873 """
874 ),
875 ] = None,
876 summary: Annotated[
877 Optional[str],
878 Doc(
879 """
880 Brief endpoint summary.
881 Example: 'Create new user'
882 """
883 ),
884 ] = None,
885 description: Annotated[
886 Optional[str],
887 Doc(
888 """
889 Detailed endpoint description.
890 Example: 'Creates new user with provided data'
891 """
892 ),
893 ] = None,
894 responses: Annotated[
895 Optional[Dict[int, Any]],
896 Doc(
897 """
898 Response schemas by status code.
899 Example: {
900 201: UserSchema,
901 400: {"description": "Invalid input"},
902 409: {"description": "User already exists"}
903 }
904 """
905 ),
906 ] = None,
907 request_model: Annotated[
908 Optional[Type[BaseModel]],
909 Doc(
910 """
911 Model for request body validation.
912 Example:
913 class UserCreate(BaseModel):
914 username: str
915 email: EmailStr
916 password: str
917 """
918 ),
919 ] = None,
920 middlewares: Annotated[
921 List[Any],
922 Doc(
923 """
924 Route-specific middleware.
925 Example: [rate_limit(10), validate_content_type('json')]
926 """
927 ),
928 ] = [],
929 tags: Annotated[
930 Optional[List[str]],
931 Doc(
932 """
933 OpenAPI tags for grouping.
934 Example: ["User Management"]
935 """
936 ),
937 ] = None,
938 security: Annotated[
939 Optional[List[Dict[str, List[str]]]],
940 Doc(
941 """
942 Security requirements.
943 Example: [{"BearerAuth": []}]
944 """
945 ),
946 ] = None,
947 operation_id: Annotated[
948 Optional[str],
949 Doc(
950 """
951 Unique operation ID.
952 Example: 'createUser'
953 """
954 ),
955 ] = None,
956 deprecated: Annotated[
957 bool,
958 Doc(
959 """
960 Mark as deprecated.
961 Example: False
962 """
963 ),
964 ] = False,
965 parameters: Annotated[
966 List[Parameter],
967 Doc(
968 """
969 Additional parameters.
970 Example: [Parameter(name="X-Request-ID", in_="header")]
971 """
972 ),
973 ] = [],
974 exclude_from_schema: Annotated[
975 bool,
976 Doc(
977 """
978 Hide from OpenAPI docs.
979 Example: False
980 """
981 ),
982 ] = False,
983 **kwargs: Annotated[
984 Dict[str, Any],
985 Doc(
986 """
987 Additional metadata.
988 Example: {"x-audit-log": True}
989 """
990 ),
991 ],
992 ) -> Callable[..., Any]:
993 """
994 Register a POST endpoint with the application.
996 Examples:
997 1. Simple POST endpoint:
998 @router.post("/messages")
999 async def create_message(request, response):
1000 message = await Message.create(**request.json())
1001 return response.json(message, status=201)
1003 2. POST with request validation:
1004 class ProductCreate(BaseModel):
1005 name: str
1006 price: float
1007 category: str
1009 @router.post(
1010 "/products",
1011 request_model=ProductCreate,
1012 responses={201: ProductSchema}
1013 )
1014 async def create_product(request, response):
1015 product = await Product.create(**request.validated_data)
1016 return response.json(product, status=201)
1018 3. POST with file upload:
1019 @router.post("/upload")
1020 async def upload_file(request, response):
1021 file = request.files.get('file')
1022 # Process file upload
1023 return response.json({"filename": file.filename})
1024 """
1025 return self.route(
1026 path=path,
1027 methods=["POST"],
1028 handler=handler,
1029 name=name,
1030 summary=summary,
1031 description=description,
1032 responses=responses,
1033 request_model=request_model,
1034 middlewares=middlewares,
1035 tags=tags,
1036 security=security,
1037 operation_id=operation_id,
1038 deprecated=deprecated,
1039 parameters=parameters,
1040 exclude_from_schema=exclude_from_schema,
1041 **kwargs,
1042 )
1044 def delete(
1045 self,
1046 path: Annotated[
1047 str,
1048 Doc(
1049 """
1050 URL path pattern for the DELETE endpoint.
1051 Example: '/api/v1/users/{id}'
1052 """
1053 ),
1054 ],
1055 handler: Annotated[
1056 Optional[HandlerType],
1057 Doc(
1058 """
1059 Async handler function for DELETE requests.
1060 Example:
1061 async def delete_user(request, response):
1062 user_id = request.path_params['id']
1063 return response.json({"deleted": user_id})
1064 """
1065 ),
1066 ] = None,
1067 name: Annotated[
1068 Optional[str],
1069 Doc(
1070 """
1071 Unique route name for URL generation.
1072 Example: 'api-v1-delete-user'
1073 """
1074 ),
1075 ] = None,
1076 summary: Annotated[
1077 Optional[str],
1078 Doc(
1079 """
1080 Brief endpoint summary.
1081 Example: 'Delete user account'
1082 """
1083 ),
1084 ] = None,
1085 description: Annotated[
1086 Optional[str],
1087 Doc(
1088 """
1089 Detailed endpoint description.
1090 Example: 'Permanently deletes user account and all associated data'
1091 """
1092 ),
1093 ] = None,
1094 responses: Annotated[
1095 Optional[Dict[int, Any]],
1096 Doc(
1097 """
1098 Response schemas by status code.
1099 Example: {
1100 204: None,
1101 404: {"description": "User not found"},
1102 403: {"description": "Forbidden"}
1103 }
1104 """
1105 ),
1106 ] = None,
1107 request_model: Annotated[
1108 Optional[Type[BaseModel]],
1109 Doc(
1110 """
1111 Model for request validation.
1112 Example:
1113 class DeleteConfirmation(BaseModel):
1114 confirm: bool
1115 """
1116 ),
1117 ] = None,
1118 middlewares: Annotated[
1119 List[Any],
1120 Doc(
1121 """
1122 Route-specific middleware.
1123 Example: [admin_required, confirm_action]
1124 """
1125 ),
1126 ] = [],
1127 tags: Annotated[
1128 Optional[List[str]],
1129 Doc(
1130 """
1131 OpenAPI tags for grouping.
1132 Example: ["User Management"]
1133 """
1134 ),
1135 ] = None,
1136 security: Annotated[
1137 Optional[List[Dict[str, List[str]]]],
1138 Doc(
1139 """
1140 Security requirements.
1141 Example: [{"BearerAuth": []}]
1142 """
1143 ),
1144 ] = None,
1145 operation_id: Annotated[
1146 Optional[str],
1147 Doc(
1148 """
1149 Unique operation ID.
1150 Example: 'deleteUser'
1151 """
1152 ),
1153 ] = None,
1154 deprecated: Annotated[
1155 bool,
1156 Doc(
1157 """
1158 Mark as deprecated.
1159 Example: False
1160 """
1161 ),
1162 ] = False,
1163 parameters: Annotated[
1164 List[Parameter],
1165 Doc(
1166 """
1167 Additional parameters.
1168 Example: [Parameter(name="confirm", in_="query")]
1169 """
1170 ),
1171 ] = [],
1172 exclude_from_schema: Annotated[
1173 bool,
1174 Doc(
1175 """
1176 Hide from OpenAPI docs.
1177 Example: False
1178 """
1179 ),
1180 ] = False,
1181 **kwargs: Annotated[
1182 Dict[str, Any],
1183 Doc(
1184 """
1185 Additional metadata.
1186 Example: {"x-destructive": True}
1187 """
1188 ),
1189 ],
1190 ) -> Callable[..., Any]:
1191 """
1192 Register a DELETE endpoint with the application.
1194 Examples:
1195 1. Simple DELETE endpoint:
1196 @router.delete("/users/{id}")
1197 async def delete_user(request, response):
1198 await User.delete(request.path_params['id'])
1199 return response.status(204)
1201 2. DELETE with confirmation:
1202 @router.delete(
1203 "/account",
1204 responses={
1205 204: None,
1206 400: {"description": "Confirmation required"}
1207 }
1208 )
1209 async def delete_account(request, response):
1210 if not request.query_params.get('confirm'):
1211 return response.status(400)
1212 await request.user.delete()
1213 return response.status(204)
1215 3. Soft DELETE:
1216 @router.delete("/posts/{id}")
1217 async def soft_delete_post(request, response):
1218 await Post.soft_delete(request.path_params['id'])
1219 return response.json({"status": "archived"})
1220 """
1221 return self.route(
1222 path=path,
1223 methods=["DELETE"],
1224 handler=handler,
1225 name=name,
1226 summary=summary,
1227 description=description,
1228 responses=responses,
1229 request_model=request_model,
1230 middlewares=middlewares,
1231 tags=tags,
1232 security=security,
1233 operation_id=operation_id,
1234 deprecated=deprecated,
1235 parameters=parameters,
1236 exclude_from_schema=exclude_from_schema,
1237 **kwargs,
1238 )
1240 def put(
1241 self,
1242 path: Annotated[
1243 str,
1244 Doc(
1245 """
1246 URL path pattern for the PUT endpoint.
1247 Example: '/api/v1/users/{id}'
1248 """
1249 ),
1250 ],
1251 handler: Annotated[
1252 Optional[HandlerType],
1253 Doc(
1254 """
1255 Async handler function for PUT requests.
1256 Example:
1257 async def update_user(request, response):
1258 user_id = request.path_params['id']
1259 return response.json({"updated": user_id})
1260 """
1261 ),
1262 ] = None,
1263 name: Annotated[
1264 Optional[str],
1265 Doc(
1266 """
1267 Unique route name for URL generation.
1268 Example: 'api-v1-update-user'
1269 """
1270 ),
1271 ] = None,
1272 summary: Annotated[
1273 Optional[str],
1274 Doc(
1275 """
1276 Brief endpoint summary.
1277 Example: 'Update user details'
1278 """
1279 ),
1280 ] = None,
1281 description: Annotated[
1282 Optional[str],
1283 Doc(
1284 """
1285 Detailed endpoint description.
1286 Example: 'Full update of user resource'
1287 """
1288 ),
1289 ] = None,
1290 responses: Annotated[
1291 Optional[Dict[int, Any]],
1292 Doc(
1293 """
1294 Response schemas by status code.
1295 Example: {
1296 200: UserSchema,
1297 400: {"description": "Invalid input"},
1298 404: {"description": "User not found"}
1299 }
1300 """
1301 ),
1302 ] = None,
1303 request_model: Annotated[
1304 Optional[Type[BaseModel]],
1305 Doc(
1306 """
1307 Model for request body validation.
1308 Example:
1309 class UserUpdate(BaseModel):
1310 email: Optional[EmailStr]
1311 password: Optional[str]
1312 """
1313 ),
1314 ] = None,
1315 middlewares: Annotated[
1316 List[Any],
1317 Doc(
1318 """
1319 Route-specific middleware.
1320 Example: [owner_required, validate_etag]
1321 """
1322 ),
1323 ] = [],
1324 tags: Annotated[
1325 Optional[List[str]],
1326 Doc(
1327 """
1328 OpenAPI tags for grouping.
1329 Example: ["User Management"]
1330 """
1331 ),
1332 ] = None,
1333 security: Annotated[
1334 Optional[List[Dict[str, List[str]]]],
1335 Doc(
1336 """
1337 Security requirements.
1338 Example: [{"BearerAuth": []}]
1339 """
1340 ),
1341 ] = None,
1342 operation_id: Annotated[
1343 Optional[str],
1344 Doc(
1345 """
1346 Unique operation ID.
1347 Example: 'updateUser'
1348 """
1349 ),
1350 ] = None,
1351 deprecated: Annotated[
1352 bool,
1353 Doc(
1354 """
1355 Mark as deprecated.
1356 Example: False
1357 """
1358 ),
1359 ] = False,
1360 parameters: Annotated[
1361 List[Parameter],
1362 Doc(
1363 """
1364 Additional parameters.
1365 Example: [Parameter(name="If-Match", in_="header")]
1366 """
1367 ),
1368 ] = [],
1369 exclude_from_schema: Annotated[
1370 bool,
1371 Doc(
1372 """
1373 Hide from OpenAPI docs.
1374 Example: False
1375 """
1376 ),
1377 ] = False,
1378 **kwargs: Annotated[
1379 Dict[str, Any],
1380 Doc(
1381 """
1382 Additional metadata.
1383 Example: {"x-idempotent": True}
1384 """
1385 ),
1386 ],
1387 ) -> Callable[..., Any]:
1388 """
1389 Register a PUT endpoint with the application.
1391 Examples:
1392 1. Simple PUT endpoint:
1393 @router.put("/users/{id}")
1394 async def update_user(request, response):
1395 user_id = request.path_params['id']
1396 await User.update(user_id, **request.json())
1397 return response.json({"status": "updated"})
1399 2. PUT with full resource replacement:
1400 @router.put(
1401 "/articles/{slug}",
1402 request_model=ArticleUpdate,
1403 responses={
1404 200: ArticleSchema,
1405 404: {"description": "Article not found"}
1406 }
1407 )
1408 async def replace_article(request, response):
1409 article = await Article.replace(
1410 request.path_params['slug'],
1411 request.validated_data
1412 )
1413 return response.json(article)
1415 3. PUT with conditional update:
1416 @router.put("/resources/{id}")
1417 async def update_resource(request, response):
1418 if request.headers.get('If-Match') != expected_etag:
1419 return response.status(412)
1420 # Process update
1421 return response.json({"status": "success"})
1422 """
1423 return self.route(
1424 path=path,
1425 methods=["PUT"],
1426 handler=handler,
1427 name=name,
1428 summary=summary,
1429 description=description,
1430 responses=responses,
1431 request_model=request_model,
1432 middlewares=middlewares,
1433 tags=tags,
1434 security=security,
1435 operation_id=operation_id,
1436 deprecated=deprecated,
1437 parameters=parameters,
1438 exclude_from_schema=exclude_from_schema,
1439 **kwargs,
1440 )
1442 def patch(
1443 self,
1444 path: Annotated[
1445 str,
1446 Doc(
1447 """
1448 URL path pattern for the PATCH endpoint.
1449 Example: '/api/v1/users/{id}'
1450 """
1451 ),
1452 ],
1453 handler: Annotated[
1454 Optional[HandlerType],
1455 Doc(
1456 """
1457 Async handler function for PATCH requests.
1458 Example:
1459 async def partial_update_user(request, response):
1460 user_id = request.path_params['id']
1461 return response.json({"updated": user_id})
1462 """
1463 ),
1464 ] = None,
1465 name: Annotated[
1466 Optional[str],
1467 Doc(
1468 """
1469 Unique route name for URL generation.
1470 Example: 'api-v1-partial-update-user'
1471 """
1472 ),
1473 ] = None,
1474 summary: Annotated[
1475 Optional[str],
1476 Doc(
1477 """
1478 Brief endpoint summary.
1479 Example: 'Partially update user details'
1480 """
1481 ),
1482 ] = None,
1483 description: Annotated[
1484 Optional[str],
1485 Doc(
1486 """
1487 Detailed endpoint description.
1488 Example: 'Partial update of user resource'
1489 """
1490 ),
1491 ] = None,
1492 responses: Annotated[
1493 Optional[Dict[int, Any]],
1494 Doc(
1495 """
1496 Response schemas by status code.
1497 Example: {
1498 200: UserSchema,
1499 400: {"description": "Invalid input"},
1500 404: {"description": "User not found"}
1501 }
1502 """
1503 ),
1504 ] = None,
1505 request_model: Annotated[
1506 Optional[Type[BaseModel]],
1507 Doc(
1508 """
1509 Model for request body validation.
1510 Example:
1511 class UserPatch(BaseModel):
1512 email: Optional[EmailStr] = None
1513 password: Optional[str] = None
1514 """
1515 ),
1516 ] = None,
1517 middlewares: Annotated[
1518 List[Any],
1519 Doc(
1520 """
1521 Route-specific middleware.
1522 Example: [owner_required, validate_patch]
1523 """
1524 ),
1525 ] = [],
1526 tags: Annotated[
1527 Optional[List[str]],
1528 Doc(
1529 """
1530 OpenAPI tags for grouping.
1531 Example: ["User Management"]
1532 """
1533 ),
1534 ] = None,
1535 security: Annotated[
1536 Optional[List[Dict[str, List[str]]]],
1537 Doc(
1538 """
1539 Security requirements.
1540 Example: [{"BearerAuth": []}]
1541 """
1542 ),
1543 ] = None,
1544 operation_id: Annotated[
1545 Optional[str],
1546 Doc(
1547 """
1548 Unique operation ID.
1549 Example: 'partialUpdateUser'
1550 """
1551 ),
1552 ] = None,
1553 deprecated: Annotated[
1554 bool,
1555 Doc(
1556 """
1557 Mark as deprecated.
1558 Example: False
1559 """
1560 ),
1561 ] = False,
1562 parameters: Annotated[
1563 List[Parameter],
1564 Doc(
1565 """
1566 Additional parameters.
1567 Example: [Parameter(name="fields", in_="query")]
1568 """
1569 ),
1570 ] = [],
1571 exclude_from_schema: Annotated[
1572 bool,
1573 Doc(
1574 """
1575 Hide from OpenAPI docs.
1576 Example: False
1577 """
1578 ),
1579 ] = False,
1580 **kwargs: Annotated[
1581 Dict[str, Any],
1582 Doc(
1583 """
1584 Additional metadata.
1585 Example: {"x-partial-update": True}
1586 """
1587 ),
1588 ],
1589 ) -> Callable[..., Any]:
1590 """
1591 Register a PATCH endpoint with the application.
1593 Examples:
1594 1. Simple PATCH endpoint:
1595 @router.patch("/users/{id}")
1596 async def update_user(request, response):
1597 user_id = request.path_params['id']
1598 await User.partial_update(user_id, **request.json())
1599 return response.json({"status": "updated"})
1601 2. PATCH with JSON Merge Patch:
1602 @router.patch(
1603 "/articles/{id}",
1604 request_model=ArticlePatch,
1605 responses={200: ArticleSchema}
1606 )
1607 async def patch_article(request, response):
1608 article = await Article.patch(
1609 request.path_params['id'],
1610 request.validated_data
1611 )
1612 return response.json(article)
1614 3. PATCH with selective fields:
1615 @router.patch("/profile")
1616 async def update_profile(request, response):
1617 allowed_fields = {'bio', 'avatar_url'}
1618 updates = {k: v for k, v in request.json().items()
1619 if k in allowed_fields}
1620 await Profile.update(request.user.id, **updates)
1621 return response.json(updates)
1622 """
1623 return self.route(
1624 path=path,
1625 methods=["PATCH"],
1626 handler=handler,
1627 name=name,
1628 summary=summary,
1629 description=description,
1630 responses=responses,
1631 request_model=request_model,
1632 middlewares=middlewares,
1633 tags=tags,
1634 security=security,
1635 operation_id=operation_id,
1636 deprecated=deprecated,
1637 parameters=parameters,
1638 exclude_from_schema=exclude_from_schema,
1639 **kwargs,
1640 )
1642 def options(
1643 self,
1644 path: Annotated[
1645 str,
1646 Doc(
1647 """
1648 URL path pattern for the OPTIONS endpoint.
1649 Example: '/api/v1/users'
1650 """
1651 ),
1652 ],
1653 handler: Annotated[
1654 Optional[HandlerType],
1655 Doc(
1656 """
1657 Async handler function for OPTIONS requests.
1658 Example:
1659 async def user_options(request, response):
1660 response.headers['Allow'] = 'GET, POST, OPTIONS'
1661 return response
1662 """
1663 ),
1664 ] = None,
1665 name: Annotated[
1666 Optional[str],
1667 Doc(
1668 """
1669 Unique route name for URL generation.
1670 Example: 'api-v1-user-options'
1671 """
1672 ),
1673 ] = None,
1674 summary: Annotated[
1675 Optional[str],
1676 Doc(
1677 """
1678 Brief endpoint summary.
1679 Example: 'Get supported operations'
1680 """
1681 ),
1682 ] = None,
1683 description: Annotated[
1684 Optional[str],
1685 Doc(
1686 """
1687 Detailed endpoint description.
1688 Example: 'Returns supported HTTP methods and CORS headers'
1689 """
1690 ),
1691 ] = None,
1692 responses: Annotated[
1693 Optional[Dict[int, Any]],
1694 Doc(
1695 """
1696 Response schemas by status code.
1697 Example: {
1698 200: None,
1699 204: None
1700 }
1701 """
1702 ),
1703 ] = None,
1704 request_model: Annotated[
1705 Optional[Type[BaseModel]],
1706 Doc(
1707 """
1708 Model for request validation.
1709 Example:
1710 class OptionsQuery(BaseModel):
1711 detailed: bool = False
1712 """
1713 ),
1714 ] = None,
1715 middlewares: Annotated[
1716 List[Any],
1717 Doc(
1718 """
1719 Route-specific middleware.
1720 Example: [cors_middleware]
1721 """
1722 ),
1723 ] = [],
1724 tags: Annotated[
1725 Optional[List[str]],
1726 Doc(
1727 """
1728 OpenAPI tags for grouping.
1729 Example: ["CORS"]
1730 """
1731 ),
1732 ] = None,
1733 security: Annotated[
1734 Optional[List[Dict[str, List[str]]]],
1735 Doc(
1736 """
1737 Security requirements.
1738 Example: []
1739 """
1740 ),
1741 ] = None,
1742 operation_id: Annotated[
1743 Optional[str],
1744 Doc(
1745 """
1746 Unique operation ID.
1747 Example: 'userOptions'
1748 """
1749 ),
1750 ] = None,
1751 deprecated: Annotated[
1752 bool,
1753 Doc(
1754 """
1755 Mark as deprecated.
1756 Example: False
1757 """
1758 ),
1759 ] = False,
1760 parameters: Annotated[
1761 List[Parameter],
1762 Doc(
1763 """
1764 Additional parameters.
1765 Example: [Parameter(name="Origin", in_="header")]
1766 """
1767 ),
1768 ] = [],
1769 exclude_from_schema: Annotated[
1770 bool,
1771 Doc(
1772 """
1773 Hide from OpenAPI docs.
1774 Example: True
1775 """
1776 ),
1777 ] = False,
1778 **kwargs: Annotated[
1779 Dict[str, Any],
1780 Doc(
1781 """
1782 Additional metadata.
1783 Example: {"x-cors": True}
1784 """
1785 ),
1786 ],
1787 ) -> Callable[..., Any]:
1788 """
1789 Register an OPTIONS endpoint with the application.
1791 Examples:
1792 1. Simple OPTIONS endpoint:
1793 @router.options("/users")
1794 async def user_options(request, response):
1795 response.headers['Allow'] = 'GET, POST, OPTIONS'
1796 return response
1798 2. CORS OPTIONS handler:
1799 @router.options("/{path:path}")
1800 async def cors_options(request, response):
1801 response.headers.update({
1802 'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE',
1803 'Access-Control-Allow-Headers': 'Content-Type',
1804 'Access-Control-Max-Age': '86400'
1805 })
1806 return response.status(204)
1808 3. Detailed OPTIONS response:
1809 @router.options("/resources")
1810 async def resource_options(request, response):
1811 return response.json({
1812 "methods": ["GET", "POST"],
1813 "formats": ["application/json"],
1814 "limits": {"max_size": "10MB"}
1815 })
1816 """
1817 return self.route(
1818 path=path,
1819 methods=["OPTIONS"],
1820 handler=handler,
1821 name=name,
1822 summary=summary,
1823 description=description,
1824 responses=responses,
1825 request_model=request_model,
1826 middlewares=middlewares,
1827 tags=tags,
1828 security=security,
1829 operation_id=operation_id,
1830 deprecated=deprecated,
1831 parameters=parameters,
1832 exclude_from_schema=exclude_from_schema,
1833 **kwargs,
1834 )
1836 def head(
1837 self,
1838 path: Annotated[
1839 str,
1840 Doc(
1841 """
1842 URL path pattern for the HEAD endpoint.
1843 Example: '/api/v1/resources/{id}'
1844 """
1845 ),
1846 ],
1847 handler: Annotated[
1848 Optional[HandlerType],
1849 Doc(
1850 """
1851 Async handler function for HEAD requests.
1852 Example:
1853 async def check_resource(request, response):
1854 exists = await Resource.exists(request.path_params['id'])
1855 return response.status(200 if exists else 404)
1856 """
1857 ),
1858 ] = None,
1859 name: Annotated[
1860 Optional[str],
1861 Doc(
1862 """
1863 Unique route name for URL generation.
1864 Example: 'api-v1-check-resource'
1865 """
1866 ),
1867 ] = None,
1868 summary: Annotated[
1869 Optional[str],
1870 Doc(
1871 """
1872 Brief endpoint summary.
1873 Example: 'Check resource existence'
1874 """
1875 ),
1876 ] = None,
1877 description: Annotated[
1878 Optional[str],
1879 Doc(
1880 """
1881 Detailed endpoint description.
1882 Example: 'Returns headers only to check if resource exists'
1883 """
1884 ),
1885 ] = None,
1886 responses: Annotated[
1887 Optional[Dict[int, Any]],
1888 Doc(
1889 """
1890 Response schemas by status code.
1891 Example: {
1892 200: None,
1893 404: None
1894 }
1895 """
1896 ),
1897 ] = None,
1898 request_model: Annotated[
1899 Optional[Type[BaseModel]],
1900 Doc(
1901 """
1902 Model for request validation.
1903 Example:
1904 class ResourceCheck(BaseModel):
1905 check_children: bool = False
1906 """
1907 ),
1908 ] = None,
1909 middlewares: Annotated[
1910 List[Any],
1911 Doc(
1912 """
1913 Route-specific middleware.
1914 Example: [cache_control('public')]
1915 """
1916 ),
1917 ] = [],
1918 tags: Annotated[
1919 Optional[List[str]],
1920 Doc(
1921 """
1922 OpenAPI tags for grouping.
1923 Example: ["Resource Management"]
1924 """
1925 ),
1926 ] = None,
1927 security: Annotated[
1928 Optional[List[Dict[str, List[str]]]],
1929 Doc(
1930 """
1931 Security requirements.
1932 Example: [{"ApiKeyAuth": []}]
1933 """
1934 ),
1935 ] = None,
1936 operation_id: Annotated[
1937 Optional[str],
1938 Doc(
1939 """
1940 Unique operation ID.
1941 Example: 'checkResource'
1942 """
1943 ),
1944 ] = None,
1945 deprecated: Annotated[
1946 bool,
1947 Doc(
1948 """
1949 Mark as deprecated.
1950 Example: False
1951 """
1952 ),
1953 ] = False,
1954 parameters: Annotated[
1955 List[Parameter],
1956 Doc(
1957 """
1958 Additional parameters.
1959 Example: [Parameter(name="X-Check-Type", in_="header")]
1960 """
1961 ),
1962 ] = [],
1963 exclude_from_schema: Annotated[
1964 bool,
1965 Doc(
1966 """
1967 Hide from OpenAPI docs.
1968 Example: False
1969 """
1970 ),
1971 ] = False,
1972 **kwargs: Annotated[
1973 Dict[str, Any],
1974 Doc(
1975 """
1976 Additional metadata.
1977 Example: {"x-head-only": True}
1978 """
1979 ),
1980 ],
1981 ) -> Callable[..., Any]:
1982 """
1983 Register a HEAD endpoint with the application.
1985 Examples:
1986 1. Simple HEAD endpoint:
1987 @router.head("/resources/{id}")
1988 async def check_resource(request, response):
1989 exists = await Resource.exists(request.path_params['id'])
1990 return response.status(200 if exists else 404)
1992 2. HEAD with cache headers:
1993 @router.head("/static/{path:path}")
1994 async def check_static(request, response):
1995 path = request.path_params['path']
1996 if not static_file_exists(path):
1997 return response.status(404)
1998 response.headers['Last-Modified'] = get_last_modified(path)
1999 return response.status(200)
2001 3. HEAD with metadata:
2002 @router.head("/documents/{id}")
2003 async def document_metadata(request, response):
2004 doc = await Document.metadata(request.path_params['id'])
2005 if not doc:
2006 return response.status(404)
2007 response.headers['X-Document-Size'] = str(doc.size)
2008 return response.status(200)
2009 """
2010 return self.route(
2011 path=path,
2012 methods=["HEAD"],
2013 handler=handler,
2014 name=name,
2015 summary=summary,
2016 description=description,
2017 responses=responses,
2018 request_model=request_model,
2019 middlewares=middlewares,
2020 tags=tags,
2021 security=security,
2022 operation_id=operation_id,
2023 deprecated=deprecated,
2024 parameters=parameters,
2025 exclude_from_schema=exclude_from_schema,
2026 **kwargs,
2027 )
2029 def route(
2030 self,
2031 path: Annotated[
2032 str,
2033 Doc(
2034 """
2035 The URL path pattern for the route. Supports path parameters using curly braces:
2036 - '/users/{user_id}' - Simple path parameter
2037 - '/files/{filepath:path}' - Matches any path (including slashes)
2038 - '/items/{id:int}' - Type-constrained parameter
2039 """
2040 ),
2041 ],
2042 methods: Annotated[
2043 List[str],
2044 Doc(
2045 """
2046 List of HTTP methods this route should handle.
2047 Common methods: ['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS', 'HEAD']
2048 Defaults to all standard methods if not specified.
2049 """
2050 ),
2051 ] = allowed_methods_default,
2052 handler: Annotated[
2053 Optional[HandlerType],
2054 Doc(
2055 """
2056 The async handler function for this route. Must accept:
2057 - request: Request object
2058 - response: Response object
2059 And return either a Response object or raw data (dict, list, str)
2060 """
2061 ),
2062 ] = None,
2063 name: Annotated[
2064 Optional[str],
2065 Doc(
2066 """
2067 Unique name for this route, used for URL generation with url_for().
2068 If not provided, will be generated from the path and methods.
2069 """
2070 ),
2071 ] = None,
2072 summary: Annotated[
2073 Optional[str],
2074 Doc("Brief one-line description of the route for OpenAPI docs"),
2075 ] = None,
2076 description: Annotated[
2077 Optional[str], Doc("Detailed description of the route for OpenAPI docs")
2078 ] = None,
2079 responses: Annotated[
2080 Optional[Dict[int, Union[Type[BaseModel], Dict[str, Any]]]],
2081 Doc(
2082 """
2083 Response models by status code for OpenAPI docs.
2084 Example: {200: UserModel, 404: ErrorModel}
2085 """
2086 ),
2087 ] = None,
2088 request_model: Annotated[
2089 Optional[Type[BaseModel]],
2090 Doc("Pydantic model for request body validation and OpenAPI docs"),
2091 ] = None,
2092 middlewares: Annotated[
2093 List[MiddlewareType],
2094 Doc(
2095 """
2096 List of middleware specific to this route.
2097 These will be executed in order before the route handler.
2098 """
2099 ),
2100 ] = [],
2101 tags: Annotated[
2102 Optional[List[str]],
2103 Doc(
2104 """
2105 OpenAPI tags for grouping related routes in documentation.
2106 Inherits parent router tags if not specified.
2107 """
2108 ),
2109 ] = None,
2110 security: Annotated[
2111 Optional[List[Dict[str, List[str]]]],
2112 Doc(
2113 """
2114 Security requirements for this route.
2115 Example: [{"bearerAuth": []}] for JWT auth.
2116 """
2117 ),
2118 ] = None,
2119 operation_id: Annotated[
2120 Optional[str],
2121 Doc(
2122 """
2123 Unique identifier for this operation in OpenAPI docs.
2124 Auto-generated if not provided.
2125 """
2126 ),
2127 ] = None,
2128 deprecated: Annotated[
2129 bool, Doc("Mark route as deprecated in OpenAPI docs")
2130 ] = False,
2131 parameters: Annotated[
2132 List[Parameter],
2133 Doc(
2134 """
2135 Additional OpenAPI parameter definitions.
2136 Path parameters are automatically included from the path pattern.
2137 """
2138 ),
2139 ] = [],
2140 exclude_from_schema: Annotated[
2141 bool,
2142 Doc(
2143 """
2144 If True, excludes this route from OpenAPI documentation.
2145 Useful for internal or admin routes.
2146 """
2147 ),
2148 ] = False,
2149 **kwargs: Annotated[
2150 Dict[str, Any],
2151 Doc(
2152 """
2153 Additional route metadata that will be available in the request scope.
2154 Useful for custom extensions or plugin-specific data.
2155 """
2156 ),
2157 ],
2158 ) -> Union[HandlerType, Callable[..., HandlerType]]:
2159 """
2160 Register a route with configurable HTTP methods and OpenAPI documentation.
2162 This is the most flexible way to register routes, allowing full control over
2163 HTTP methods, request/response validation, and OpenAPI documentation.
2165 Can be used as a decorator:
2166 @router.route("/users", methods=["GET", "POST"])
2167 async def user_handler(request, response):
2168 ...
2170 Or directly:
2171 router.route("/users", methods=["GET", "POST"], handler=user_handler)
2173 Args:
2174 path: URL path pattern with optional parameters
2175 methods: HTTP methods this route accepts
2176 handler: Async function to handle requests
2177 name: Unique route identifier
2178 summary: Brief route description
2179 description: Detailed route description
2180 responses: Response models by status code
2181 request_model: Request body validation model
2182 middlewares: Route-specific middleware
2183 tags: OpenAPI tags for grouping
2184 security: Security requirements
2185 operation_id: OpenAPI operation ID
2186 deprecated: Mark as deprecated
2187 parameters: Additional OpenAPI parameters
2188 exclude_from_schema: Hide from docs
2189 **kwargs: Additional route metadata
2191 Returns:
2192 The route handler function (when used as decorator)
2193 """
2195 def decorator(handler: HandlerType) -> HandlerType:
2196 route = self.route_class(
2197 path=path,
2198 handler=handler,
2199 methods=methods,
2200 name=name,
2201 summary=summary,
2202 description=description,
2203 responses=responses,
2204 request_model=request_model,
2205 middlewares=middlewares,
2206 tags=tags,
2207 security=security,
2208 operation_id=operation_id,
2209 deprecated=deprecated,
2210 parameters=parameters,
2211 exclude_from_schema=exclude_from_schema,
2212 **kwargs,
2213 )
2214 self.add_route(route)
2215 return handler
2217 if handler is None:
2218 return decorator
2219 return decorator(handler)
2221 def url_for(self, _name: str, **path_params: Any) -> URLPath:
2222 """
2223 Generate a URL path including all router prefixes for nested routes.
2225 Args:
2226 _name: Route name in format 'router1.router2.route_name'
2227 **path_params: Path parameters to substitute
2229 Returns:
2230 URLPath: Complete path including all router prefixes
2231 """
2232 name_parts = _name.split(".")
2233 current_router = self
2234 path_segments = []
2236 # First collect all router prefixes
2237 for part in name_parts[:-1]:
2238 found = False
2239 for mount_path, sub_router in current_router.sub_routers.items():
2240 if getattr(sub_router, "name", None) == part:
2241 path_segments.append(mount_path.strip("/"))
2242 current_router = sub_router
2243 found = True
2244 break
2245 if not found:
2246 raise ValueError(
2247 f"Router '{part}' not found while building URL for '{_name}'"
2248 )
2250 route_name = name_parts[-1]
2251 for route in current_router.routes:
2252 if route.name == route_name:
2253 route_path = route.url_path_for(route_name, **path_params)
2254 path_segments.append(route_path.strip("/"))
2256 full_path = "/" + "/".join(filter(None, path_segments))
2257 return URLPath(path=full_path, protocol="http")
2259 raise ValueError(f"Route '{route_name}' not found in router")
2261 def __repr__(self) -> str:
2262 return f"<Router prefix='{self.prefix}' routes={len(self.routes)}>"
2264 async def __call__(
2265 self,
2266 scope: Scope,
2267 receive: Receive,
2268 send: Send,
2269 ) -> Any:
2270 app = self.build_middleware_stack(self.app)
2271 await app(scope, receive, send)
2273 async def app(self, scope: Scope, receive: Receive, send: Send):
2274 scope["app"] = self
2275 url = get_route_path(scope)
2277 for mount_path, sub_app in self.sub_routers.items():
2278 if url.startswith(mount_path):
2279 scope["path"] = url[len(mount_path) :]
2280 await sub_app(scope, receive, send)
2281 return
2283 path_matched = False
2284 allowed_methods_: typing.Set[str] = set()
2285 for route in self.routes:
2286 match, matched_params, is_allowed = route.match(url, scope["method"])
2288 if match:
2289 path_matched = True
2290 if is_allowed:
2291 route.handler = allowed_methods(route.methods)(route.handler)
2292 scope["route_params"] = RouteParam(matched_params)
2293 await route.handle(scope, receive, send)
2294 return
2295 else:
2296 allowed_methods_.update(route.methods)
2297 if path_matched:
2298 response = JSONResponse(
2299 content="Method not allowed",
2300 status_code=405,
2301 headers={"Allow": ", ".join(allowed_methods_)},
2302 )
2303 await response(scope, receive, send)
2304 return
2306 raise NotFoundException
2308 def mount_router(self, app: "Router", path: typing.Optional[str] = None) -> None:
2309 """
2310 Mount an ASGI application (e.g., another Router) under a specific path prefix.
2312 Args:
2313 path: The path prefix under which the app will be mounted.
2314 app: The ASGI application (e.g., another Router) to mount.
2315 """
2317 if not path:
2318 path = app.prefix
2319 path = path.rstrip("/")
2321 if path == "":
2322 self.sub_routers[path] = app
2323 return
2324 if not path.startswith("/"):
2325 path = f"/{path}"
2327 if path in self.sub_routers.keys():
2328 raise ValueError("Router with prefix exists !")
2330 self.sub_routers[path] = app
2332 def get_all_routes(self) -> List[Routes]:
2333 """Returns a flat list of all HTTP routes in this router and all nested sub-routers"""
2334 all_routes: List[Routes] = []
2335 routers_to_process: List[Any] = [(self, "")] # (router, current_prefix)
2337 while routers_to_process:
2338 current_router, current_prefix = routers_to_process.pop(0)
2340 for route in current_router.routes:
2341 # Create a copy of the route with updated path
2342 new_route = copy.copy(route)
2343 new_route.raw_path = current_prefix + route.raw_path
2344 new_route.prefix = current_prefix
2345 all_routes.append(new_route)
2347 for mount_path, sub_router in current_router.sub_routers.items():
2348 if isinstance(sub_router, Router):
2349 new_prefix = current_prefix + mount_path
2350 routers_to_process.append((sub_router, new_prefix))
2352 return all_routes
2355class WebsocketRoutes:
2356 def __init__(
2357 self,
2358 path: str,
2359 handler: WsHandlerType,
2360 middlewares: typing.List[WsMiddlewareType] = [],
2361 ):
2362 assert callable(handler), "Route handler must be callable"
2363 assert asyncio.iscoroutinefunction(handler), "Route handler must be async"
2364 self.raw_path = path
2365 self.handler: WsHandlerType = inject_dependencies(handler)
2366 self.middlewares = middlewares
2367 self.route_info = RouteBuilder.create_pattern(path)
2368 self.pattern = self.route_info.pattern
2369 self.param_names = self.route_info.param_names
2370 self.route_type = self.route_info.route_type
2371 self.router_middleware = None
2373 def match(self, path: str) -> typing.Tuple[Any, Any]:
2374 """
2375 Match a path against this route's pattern and return captured parameters.
2377 Args:
2378 path: The URL path to match.
2380 Returns:
2381 Optional[Dict[str, Any]]: A dictionary of captured parameters if the path matches,
2382 otherwise None.
2383 """
2384 match = self.pattern.match(path)
2385 if match:
2386 matched_params = match.groupdict()
2387 for key, value in matched_params.items():
2388 matched_params[key] = self.route_info.convertor[ # type:ignore
2389 key
2390 ].convert(value)
2391 return match, matched_params
2392 return None, None
2394 async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
2395 """
2396 Handles the WebSocket connection by calling the route's handler.
2398 Args:
2399 websocket: The WebSocket connection.
2400 params: The extracted route parameters.
2401 """
2402 websocket_session = WebSocket(scope, receive=receive, send=send)
2403 await self.handler(websocket_session)
2405 def __repr__(self) -> str:
2406 return f"<WSRoute {self.raw_path}>"
2409class WSRouter(BaseRouter):
2410 def __init__(
2411 self,
2412 prefix: Optional[str] = None,
2413 middleware: Optional[List[Any]] = [],
2414 routes: Optional[List[WebsocketRoutes]] = [],
2415 ):
2416 self.prefix = prefix or ""
2417 self.routes: List[WebsocketRoutes] = routes or []
2418 self.middlewares: List[Callable[[ASGIApp], ASGIApp]] = []
2419 self.sub_routers: Dict[str, ASGIApp] = {}
2420 if self.prefix and not self.prefix.startswith("/"):
2421 warnings.warn("WSRouter prefix should start with '/'")
2422 self.prefix = f"/{self.prefix}"
2424 def add_ws_route(
2425 self,
2426 route: Annotated[
2427 WebsocketRoutes,
2428 Doc("An instance of the Routes class representing a WebSocket route."),
2429 ],
2430 ) -> None:
2431 """
2432 Adds a WebSocket route to the application.
2434 This method registers a WebSocket route, allowing the application to handle WebSocket connections.
2436 Args:
2437 route (Routes): The WebSocket route configuration.
2439 Returns:
2440 None
2442 Example:
2443 ```python
2444 route = Routes("/ws/chat", chat_handler)
2445 app.add_ws_route(route)
2446 ```
2447 """
2448 self.routes.append(route)
2450 def add_ws_middleware(self, middleware: type[ASGIApp]) -> None: # type: ignore[override]
2451 """Add middleware to the WebSocket router"""
2452 self.middlewares.insert(0, middleware) # type: ignore
2454 def ws_route(
2455 self,
2456 path: Annotated[
2457 str, Doc("The WebSocket route path. Must be a valid URL pattern.")
2458 ],
2459 handler: Annotated[
2460 Optional[WsHandlerType],
2461 Doc("The WebSocket handler function. Must be an async function."),
2462 ] = None,
2463 middlewares: Annotated[
2464 List[WsMiddlewareType],
2465 Doc("List of middleware to be executes before the router handler"),
2466 ] = [],
2467 ) -> Any:
2468 """
2469 Registers a WebSocket route.
2471 This decorator is used to define WebSocket routes in the application, allowing handlers
2472 to manage WebSocket connections. When a WebSocket client connects to the given path,
2473 the specified handler function will be executed.
2475 Returns:
2476 Callable: The original WebSocket handler function.
2478 Example:
2479 ```python
2481 @app.ws_route("/ws/chat")
2482 async def chat_handler(websocket):
2483 await websocket.accept()
2484 while True:
2485 message = await websocket.receive_text()
2486 await websocket.send_text(f"Echo: {message}")
2487 ```
2488 """
2489 if handler:
2490 return self.add_ws_route(
2491 WebsocketRoutes(path, handler, middlewares=middlewares)
2492 )
2494 def decorator(handler: WsHandlerType) -> WsHandlerType:
2495 self.add_ws_route(WebsocketRoutes(path, handler, middlewares=middlewares))
2496 return handler
2498 return decorator
2500 def build_middleware_stack( # type:ignore
2501 self, scope: Scope, receive: Receive, send: Send
2502 ) -> ASGIApp: # type:ignore
2503 app = self.app
2504 for mdw in reversed(self.middlewares):
2505 app = mdw(app) # type:ignore[assignment]
2506 return app
2508 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
2509 if scope["type"] != "websocket":
2510 return
2511 app = self.build_middleware_stack(scope, receive, send)
2512 app = WebSocketErrorMiddleware(app)
2513 await app(scope, receive, send)
2515 async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
2517 url = get_route_path(scope)
2518 for mount_path, sub_app in self.sub_routers.items():
2519 if url.startswith(mount_path):
2520 scope["path"] = url[len(mount_path) :]
2521 await sub_app(scope, receive, send)
2522 return
2523 for route in self.routes:
2524 match, params = route.match(url)
2525 if match:
2527 scope["route_params"] = params
2528 await route.handle(scope, receive, send)
2529 return
2530 await send({"type": "websocket.close", "code": 404})
2532 def wrap_asgi(
2533 self,
2534 middleware_cls: Annotated[
2535 Callable[[ASGIApp], Any],
2536 Doc(
2537 "An ASGI middleware class or callable that takes an app as its first argument and returns an ASGI app"
2538 ),
2539 ],
2540 ) -> None:
2541 """
2542 Wraps the entire application with an ASGI middleware.
2544 This method allows adding middleware at the ASGI level, which intercepts all requests
2545 (HTTP, WebSocket, and Lifespan) before they reach the application.
2547 Args:
2548 middleware_cls: An ASGI middleware class or callable that follows the ASGI interface
2549 *args: Additional positional arguments to pass to the middleware
2550 **kwargs: Additional keyword arguments to pass to the middleware
2552 Returns:
2553 NexiosApp: The application instance for method chaining
2556 """
2557 self.app = middleware_cls(self.app)
2559 def mount_router( # type:ignore
2560 self, app: "WSRouter", path: typing.Optional[str] = None
2561 ) -> None: # type:ignore
2562 """
2563 Mount an ASGI application (e.g., another Router) under a specific path prefix.
2565 Args:
2566 path: The path prefix under which the app will be mounted.
2567 app: The ASGI application (e.g., another Router) to mount.
2568 """
2570 if not path:
2571 path = app.prefix
2572 path = path.rstrip("/")
2574 if path == "":
2575 self.sub_routers[path] = app
2576 return
2577 if not path.startswith("/"):
2578 path = f"/{path}"
2580 self.sub_routers[path] = app
2582 def __repr__(self) -> str:
2583 return f"<WSRouter prefix='{self.prefix}' routes={len(self.routes)}>"