`Router` class¶

This is the reference for the Router that contains all the parameters, attributes and functions.

How to import¶

from ravyn import Router

ravyn.Router ¶

Router(
    path=None,
    app=None,
    parent=None,
    on_startup=None,
    on_shutdown=None,
    redirect_slashes=None,
    default=None,
    routes=None,
    name=None,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    response_class=None,
    response_cookies=None,
    response_headers=None,
    lifespan=None,
    before_request=None,
    after_request=None,
    tags=None,
    deprecated=None,
    security=None,
)

Bases: RoutingMethodsMixin, BaseRouter

The Router object used by Ravyn upon instantiation.

The router is what is created by default when the routes parameter is defined.

This object is complex and very powerful. Read more in detail about the Router and how to use it.

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format and if the path is not provided, it will default to `/`. Example `Include()` Example with parameters `Include(path="/{age: int}")` TYPE: `Optional[str]` DEFAULT: `None`
`app`	A `Router` instance always expects an `Ravyn` instance as an app or any subclass of Ravyn, like a `ChildRavyn`. Example `from ravyn import ChildRavyn, Router Router('/child', app=ChildRavyn(...))` TYPE: `Optional[Application]` DEFAULT: `None`
`parent`	Who owns the Gateway. If not specified, the application automatically it assign it. This is directly related with the application levels. TYPE: `Optional[ParentType]` DEFAULT: `None`
`on_startup`	A `list` of events that are trigger upon the application starts. Read more about the events. Example from pydantic import BaseModel from saffier import Database, Registry from ravyn import Router, Gateway, post database = Database("postgresql+asyncpg://user:password@host:port/database") registry = Registry(database=database) class User(BaseModel): name: str email: str password: str retype_password: str @post("/create", tags=["user"], description="Creates a new user in the database") async def create_user(data: User) -> None: # Logic to create the user ... app = Router( routes=[Gateway(handler=create_user)], on_startup=[database.connect], ) TYPE: `Optional[list[LifeSpanHandler]]` DEFAULT: `None`
`on_shutdown`	A `list` of events that are trigger upon the application shuts down. Read more about the events. Example from pydantic import BaseModel from saffier import Database, Registry from ravyn import Router, Gateway, post database = Database("postgresql+asyncpg://user:password@host:port/database") registry = Registry(database=database) class User(BaseModel): name: str email: str password: str retype_password: str @post("/create", tags=["user"], description="Creates a new user in the database") async def create_user(data: User) -> None: # Logic to create the user ... app = Router( routes=[Gateway(handler=create_user)], on_shutdown=[database.disconnect], ) TYPE: `Optional[list[LifeSpanHandler]]` DEFAULT: `None`
`redirect_slashes`	Boolean flag indicating if the redirect slashes are enabled for the routes or not. TYPE: `Optional[bool]` DEFAULT: `None`
`default`	A `default` ASGI callable. TYPE: `Optional[ASGIApp]` DEFAULT: `None`
`routes`	A `list` of ravyn routes. Those routes may vary and those can be `Gateway`, `WebSocketGateWay` or even an `Include`. This is also an entry-point for the routes of the `Router` but it does not rely on only one level. Read more about how to use and leverage the Ravyn routing system. Example `from ravyn import Ravyn, Gateway, Request, get, Include @get() async def homepage(request: Request) -> str: return "Hello, home!" @get() async def another(request: Request) -> str: return "Hello, another!" app = Ravyn( routes=[ Gateway(handler=homepage) Include("/nested", routes=[ Gateway(handler=another) ]) ] )` Note The Include is very powerful and this example is not enough to understand what more things you can do. Read in more detail about this. TYPE: `Optional[Sequence[Union[APIGateHandler, Include, HTTPHandler, WebSocketHandler]]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`response_class`	Default response class to be used within the Ravyn application. Read more about the Responses and how to use them. Example `from ravyn import Router, JSONResponse Router(response_class=JSONResponse)` TYPE: `Optional[ResponseType]` DEFAULT: `None`
`response_cookies`	A sequence of `ravyn.datastructures.Cookie` objects. Read more about the Cookies. Example `from ravyn import Router from ravyn.datastructures import Cookie response_cookies=[ Cookie( key="csrf", value="CIwNZNlR4XbisJF39I8yWnWX9wX4WFoz", max_age=3000, httponly=True, ) ] Router(response_cookies=response_cookies)` TYPE: `Optional[ResponseCookies]` DEFAULT: `None`
`response_headers`	A mapping of `ravyn.datastructures.ResponseHeader` objects. Read more about the ResponseHeader. Example `from ravyn import Router from ravyn.datastructures import ResponseHeader response_headers={ "authorize": ResponseHeader(value="granted") } Router(response_headers=response_headers)` TYPE: `Optional[ResponseHeaders]` DEFAULT: `None`
`lifespan`	A `lifespan` context manager handler. This is an alternative to `on_startup` and `on_shutdown` and you cannot used all combined. Read more about the lifespan. TYPE: `Optional[Lifespan[Any]]` DEFAULT: `None`
`before_request`	A `list` of events that are trigger after the application processes the request. Read more about the events. Example `from edgy import Database, Registry from ravyn import Ravyn, Request, Gateway, get database = Database("postgresql+asyncpg://user:password@host:port/database") registry = Registry(database=database) async def create_user(request: Request): # Logic to create the user data = await request.json() ... app = Ravyn( routes=[Gateway("/create", handler=create_user)], after_request=[database.disconnect], )` TYPE: `Union[Sequence[Callable[..., Any]], None]` DEFAULT: `None`
`after_request`	A `list` of events that are trigger after the application processes the request. Read more about the events. Example `from edgy import Database, Registry from ravyn import Ravyn, Request, Gateway, get database = Database("postgresql+asyncpg://user:password@host:port/database") registry = Registry(database=database) async def create_user(request: Request): # Logic to create the user data = await request.json() ... app = Ravyn( routes=[Gateway("/create", handler=create_user)], after_request=[database.disconnect], )` TYPE: `Union[Sequence[Callable[..., Any]], None]` DEFAULT: `None`
`tags`	A list of strings tags to be applied to the path operation. It will be added to the generated OpenAPI documentation. Note almost everything in Ravyn can be done in levels, which means these tags on a Ravyn instance, means it will be added to every route even if those routes also contain tags. TYPE: `Optional[Sequence[str]]` DEFAULT: `None`
`security`	Used by OpenAPI definition, the security must be compliant with the norms. Ravyn offers some out of the box solutions where this is implemented. The Ravyn security is available to automatically used. The security can be applied also on a level basis. For custom security objects, you must subclass `ravyn.openapi.security.base.HTTPBase` object. TYPE: `Optional[Sequence[SecurityScheme]]` DEFAULT: `None`

Source code in ravyn/routing/router.py

def __init__(
    self,
    path: Annotated[
        Optional[str],
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format
            and if the path is not provided, it will default to `/`.

            **Example**

            ```python
            Include()
            ```

            **Example with parameters**

            ```python
            Include(path="/{age: int}")
            ```
            """
        ),
    ] = None,
    app: Annotated[
        Optional[Application],
        Doc(
            """
            A `Router` instance always expects an `Ravyn` instance
            as an app or any subclass of Ravyn, like a `ChildRavyn`.

            **Example**

            ```python
            from ravyn import ChildRavyn, Router

            Router('/child', app=ChildRavyn(...))
            ```
            """
        ),
    ] = None,
    parent: Annotated[
        Optional[ParentType],
        Doc(
            """
            Who owns the Gateway. If not specified, the application automatically it assign it.

            This is directly related with the [application levels](https://ravyn.dev/application/levels/).
            """
        ),
    ] = None,
    on_startup: Annotated[
        Optional[list[LifeSpanHandler]],
        Doc(
            """
            A `list` of events that are trigger upon the application
            starts.

            Read more about the [events](https://ravyn.dev/lifespan-events/).

            **Example**

            ```python
            from pydantic import BaseModel
            from saffier import Database, Registry

            from ravyn import Router, Gateway, post

            database = Database("postgresql+asyncpg://user:password@host:port/database")
            registry = Registry(database=database)


            class User(BaseModel):
                name: str
                email: str
                password: str
                retype_password: str


            @post("/create", tags=["user"], description="Creates a new user in the database")
            async def create_user(data: User) -> None:
                # Logic to create the user
                ...


            app = Router(
                routes=[Gateway(handler=create_user)],
                on_startup=[database.connect],
            )
            ```
            """
        ),
    ] = None,
    on_shutdown: Annotated[
        Optional[list[LifeSpanHandler]],
        Doc(
            """
            A `list` of events that are trigger upon the application
            shuts down.

            Read more about the [events](https://ravyn.dev/lifespan-events/).

            **Example**

            ```python
            from pydantic import BaseModel
            from saffier import Database, Registry

            from ravyn import Router, Gateway, post

            database = Database("postgresql+asyncpg://user:password@host:port/database")
            registry = Registry(database=database)


            class User(BaseModel):
                name: str
                email: str
                password: str
                retype_password: str


            @post("/create", tags=["user"], description="Creates a new user in the database")
            async def create_user(data: User) -> None:
                # Logic to create the user
                ...


            app = Router(
                routes=[Gateway(handler=create_user)],
                on_shutdown=[database.disconnect],
            )
            ```
            """
        ),
    ] = None,
    redirect_slashes: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag indicating if the redirect slashes are enabled for the
            routes or not.
            """
        ),
    ] = None,
    default: Annotated[
        Optional[ASGIApp],
        Doc(
            """
            A `default` ASGI callable.
            """
        ),
    ] = None,
    routes: Annotated[
        Optional[Sequence[Union[APIGateHandler, Include, HTTPHandler, WebSocketHandler]]],
        Doc(
            """
            A `list` of ravyn routes. Those routes may vary and those can
            be `Gateway`, `WebSocketGateWay` or even an `Include`.

            This is also an entry-point for the routes of the `Router`
            but it **does not rely on only one [level](https://ravyn.dev/application/levels/)**.

            Read more about how to use and leverage
            the [Ravyn routing system](https://ravyn.dev/routing/routes/).

            **Example**

            ```python
            from ravyn import Ravyn, Gateway, Request, get, Include


            @get()
            async def homepage(request: Request) -> str:
                return "Hello, home!"


            @get()
            async def another(request: Request) -> str:
                return "Hello, another!"

            app = Ravyn(
                routes=[
                    Gateway(handler=homepage)
                    Include("/nested", routes=[
                        Gateway(handler=another)
                    ])
                ]
            )
            ```

            !!! Note
                The Include is very powerful and this example
                is not enough to understand what more things you can do.
                Read in [more detail](https://ravyn.dev/routing/routes/#include) about this.
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    response_class: Annotated[
        Optional[ResponseType],
        Doc(
            """
            Default response class to be used within the
            Ravyn application.

            Read more about the [Responses](https://ravyn.dev/responses/) and how
            to use them.

            **Example**

            ```python
            from ravyn import Router, JSONResponse

            Router(response_class=JSONResponse)
            ```
            """
        ),
    ] = None,
    response_cookies: Annotated[
        Optional[ResponseCookies],
        Doc(
            """
            A sequence of `ravyn.datastructures.Cookie` objects.

            Read more about the [Cookies](https://ravyn.dev/extras/cookie-fields/?h=responsecook#cookie-from-response-cookies).

            **Example**

            ```python
            from ravyn import Router
            from ravyn.datastructures import Cookie

            response_cookies=[
                Cookie(
                    key="csrf",
                    value="CIwNZNlR4XbisJF39I8yWnWX9wX4WFoz",
                    max_age=3000,
                    httponly=True,
                )
            ]

            Router(response_cookies=response_cookies)
            ```
            """
        ),
    ] = None,
    response_headers: Annotated[
        Optional[ResponseHeaders],
        Doc(
            """
            A mapping of `ravyn.datastructures.ResponseHeader` objects.

            Read more about the [ResponseHeader](https://ravyn.dev/extras/header-fields/#response-headers).

            **Example**

            ```python
            from ravyn import Router
            from ravyn.datastructures import ResponseHeader

            response_headers={
                "authorize": ResponseHeader(value="granted")
            }

            Router(response_headers=response_headers)
            ```
            """
        ),
    ] = None,
    lifespan: Annotated[
        Optional[Lifespan[Any]],
        Doc(
            """
            A `lifespan` context manager handler. This is an alternative
            to `on_startup` and `on_shutdown` and you **cannot used all combined**.

            Read more about the [lifespan](https://ravyn.dev/lifespan-events/).
            """
        ),
    ] = None,
    before_request: Annotated[
        Union[Sequence[Callable[..., Any]], None],
        Doc(
            """
            A `list` of events that are trigger after the application
            processes the request.

            Read more about the [events](https://lilya.dev/lifespan/).

            **Example**

            ```python
            from edgy import Database, Registry

            from ravyn import Ravyn, Request, Gateway, get

            database = Database("postgresql+asyncpg://user:password@host:port/database")
            registry = Registry(database=database)

            async def create_user(request: Request):
                # Logic to create the user
                data = await request.json()
                ...


            app = Ravyn(
                routes=[Gateway("/create", handler=create_user)],
                after_request=[database.disconnect],
            )
            ```
            """
        ),
    ] = None,
    after_request: Annotated[
        Union[Sequence[Callable[..., Any]], None],
        Doc(
            """
            A `list` of events that are trigger after the application
            processes the request.

            Read more about the [events](https://lilya.dev/lifespan/).

            **Example**

            ```python
            from edgy import Database, Registry

            from ravyn import Ravyn, Request, Gateway, get

            database = Database("postgresql+asyncpg://user:password@host:port/database")
            registry = Registry(database=database)


            async def create_user(request: Request):
                # Logic to create the user
                data = await request.json()
                ...


            app = Ravyn(
                routes=[Gateway("/create", handler=create_user)],
                after_request=[database.disconnect],
            )
            ```
            """
        ),
    ] = None,
    tags: Annotated[
        Optional[Sequence[str]],
        Doc(
            """
            A list of strings tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI documentation.

            **Note** almost everything in Ravyn can be done in [levels](https://ravyn.dev/application/levels/), which means
            these tags on a Ravyn instance, means it will be added to every route even
            if those routes also contain tags.
            """
        ),
    ] = None,
    deprecated: Optional[bool] = None,
    security: Annotated[
        Optional[Sequence[SecurityScheme]],
        Doc(
            """
            Used by OpenAPI definition, the security must be compliant with the norms.
            Ravyn offers some out of the box solutions where this is implemented.

            The [Ravyn security](https://ravyn.dev/openapi/) is available to automatically used.

            The security can be applied also on a [level basis](https://ravyn.dev/application/levels/).

            For custom security objects, you **must** subclass
            `ravyn.openapi.security.base.HTTPBase` object.
            """
        ),
    ] = None,
):
    self._app = app
    if not path:
        path = "/"
    else:
        assert path.startswith("/"), "A path prefix must start with '/'"
        assert not path.endswith("/"), (
            "A path must not end with '/', as the routes will start with '/'"
        )

    new_routes: list[Any] = []
    for route in routes or []:
        if isinstance(route, WebhookHandler):
            # WebhookHandler is a subclass of HTTPHandler, make sure to not upgrade it
            ...
        elif isinstance(route, HTTPHandler):
            route = Gateway(handler=route)
        elif is_class_and_subclass(route, BaseController):
            route = Gateway(
                handler=cast(BaseController, route),
                permissions=(
                    route.permissions
                    if not self.is_member_descriptor(route.permissions)
                    else []
                ),
                interceptors=(
                    route.interceptors
                    if not self.is_member_descriptor(route.interceptors)
                    else []
                ),
                exception_handlers=(
                    route.exception_handlers
                    if not self.is_member_descriptor(route.exception_handlers)
                    else {}
                ),
                dependencies=(
                    route.dependencies
                    if not self.is_member_descriptor(route.dependencies)
                    else {}
                ),
                middleware=(
                    route.middleware if not self.is_member_descriptor(route.middleware) else []
                ),
                before_request=(
                    route.before_request
                    if not self.is_member_descriptor(route.before_request)
                    else []
                ),
                after_request=(
                    route.after_request
                    if not self.is_member_descriptor(route.after_request)
                    else []
                ),
                security=(
                    route.security if not self.is_member_descriptor(route.security) else []
                ),
                tags=route.tags if not self.is_member_descriptor(route.tags) else [],
                include_in_schema=(
                    route.include_in_schema
                    if not self.is_member_descriptor(route.include_in_schema)
                    else True
                ),
                deprecated=(
                    route.deprecated
                    if not self.is_member_descriptor(route.deprecated)
                    else None
                ),
            )
        elif isinstance(route, WebSocketHandler):
            route = WebSocketGateway(handler=route)
        if not isinstance(
            route,
            (
                Include,
                Gateway,
                WebSocketGateway,
                LilyaBasePath,
                Host,
                Router,
            ),
        ) or isinstance(cast(Any, route), (WebhookGateway, WebhookHandler)):
            # WebhookGateway is subclass of LilyaBasePath
            raise ImproperlyConfigured(
                f"The route {route} must be of type HTTPHandler, WebSocketHandler, Gateway, WebSocketGateway or Include"
            )
        new_routes.append(route)

    assert lifespan is None or (on_startup is None and on_shutdown is None), (
        "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both."
    )

    self.__base_permissions__ = permissions or []
    self.__lilya_permissions__ = [
        wrap_permission(permission)
        for permission in self.__base_permissions__ or []
        if not is_ravyn_permission(permission)
    ]

    super().__init__(
        redirect_slashes=redirect_slashes,
        routes=new_routes,
        default=default,
        lifespan=lifespan,
        on_shutdown=on_shutdown,
        on_startup=on_startup,
        permissions=self.__lilya_permissions__,  # type: ignore
        before_request=before_request,
        after_request=after_request,
    )
    self.path = path
    self.on_startup = [] if on_startup is None else list(on_startup)
    self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
    self.parent: Optional[ParentType] = parent or app
    self.dependencies = dependencies or {}  # type: ignore
    self.exception_handlers = exception_handlers or {}
    self.interceptors: Sequence[Interceptor] = interceptors or []
    self._interceptors: Union[list["RavynInterceptor"], VoidType] = Void
    self._permissions_cache: dict[int, Any] | VoidType = Void
    self._lilya_permissions_cache: dict[int, Any] | VoidType = Void

    # Filter out the lilya unique permissions
    if self.__lilya_permissions__:
        self.lilya_permissions: Any = {
            index: permission in self.__lilya_permissions__
            for index, permission in enumerate(self.__lilya_permissions__)
        }
    else:
        self.lilya_permissions = {}

    # Filter out the ravyn unique permissions
    if self.__base_permissions__:
        self.permissions: Any = {
            index: wrap_permission(permission)
            for index, permission in enumerate(permissions)
            if is_ravyn_permission(permission)
        }
    else:
        self.permissions = {}

    assert not (self.permissions and self.lilya_permissions), (
        "Use either `Ravyn permissions` OR `Lilya permissions`, not both."
    )

    self.routes: Any = new_routes
    self.middleware = middleware or []
    self.tags = tags or []
    self.name = name
    self.response_class = response_class
    self.response_cookies = response_cookies or []
    self.response_headers = response_headers or {}
    self.deprecated = deprecated
    self.security = security or []

    # copy routes shallow, fixes bug with tests/dependencies/test_http_handler_dependency_injection.py
    # routes are modified by validate_root_route_parent
    for route in list(self.routes):
        self.validate_root_route_parent(route)

    for route in self.routes:
        self.create_signature_models(route)

    self.activate()

on_startup `instance-attribute` ¶

on_startup = [] if on_startup is None else list(on_startup)

on_shutdown `instance-attribute` ¶

on_shutdown = (
    [] if on_shutdown is None else list(on_shutdown)
)

dependencies `instance-attribute` ¶

dependencies = dependencies or {}

dependency_overrides `instance-attribute` ¶

dependency_overrides = {}

lifespan_context `instance-attribute` ¶

lifespan_context = handle_lifespan_events(
    on_startup=on_startup,
    on_shutdown=on_shutdown,
    lifespan=lifespan,
)

routes `instance-attribute` ¶

routes = new_routes

redirect_slashes `instance-attribute` ¶

redirect_slashes = redirect_slashes

default `instance-attribute` ¶

default = handle_not_found if default is None else default

include_in_schema `instance-attribute` ¶

include_in_schema = include_in_schema

deprecated `instance-attribute` ¶

deprecated = deprecated

middleware `instance-attribute` ¶

middleware = middleware or []

permissions `instance-attribute` ¶

permissions = {
    index: (wrap_permission(permission))
    for (index, permission) in (enumerate(permissions))
    if is_ravyn_permission(permission)
}

settings_module `instance-attribute` ¶

settings_module = settings_module

middleware_stack `instance-attribute` ¶

middleware_stack = app

permission_started `instance-attribute` ¶

permission_started = False

is_sub_router `instance-attribute` ¶

is_sub_router = is_sub_router

before_request `instance-attribute` ¶

before_request = (
    before_request if before_request is not None else []
)

after_request `instance-attribute` ¶

after_request = (
    after_request if after_request is not None else []
)

wrapped_permissions `instance-attribute` ¶

wrapped_permissions = [
    (wrap_permission(permission))
    for permission in (permissions or [])
]

handler_signature `property` ¶

handler_signature

Returns the Signature of the handler function.

This property returns the Signature object representing the signature of the handler function. The Signature object provides information about the parameters, return type, and annotations of the handler function.

Returns: - Signature: The Signature object representing the signature of the handler function.

Example:

handler = Dispatcher() signature = handler.handler_signature print(signature)

Note: - The Signature object is created using the from_callable method of the Signature class. - The from_callable method takes a callable object (in this case, the handler function) as input and returns a Signature object. - The Signature object can be used to inspect the parameters and return type of the handler function.

path_parameters `property` ¶

path_parameters

Gets the path parameters in a set format.

This property returns a set of path parameters used in the URL pattern of the handler. Each path parameter represents a dynamic value that is extracted from the URL during routing.

Returns: - Set[str]: A set of path parameters.

Example:

handler = Dispatcher() parameters = handler.path_parameters print(parameters)

Note: - The path parameters are extracted from the URL pattern defined in the handler's route. - The path parameters are represented as strings. - If no path parameters are defined in the URL pattern, an empty set will be returned.

stringify_parameters `property` ¶

stringify_parameters

Gets the param:type in string like list. Used for the directive ravyn show_urls.

This property returns a list of strings representing the parameter name and type in the format "param:type". It is used specifically for the ravyn show_urls directive.

The method first parses the path of the dispatcher object using the parse_path method. It then filters out any path components that are not dictionaries, leaving only the parameter components.

Next, it iterates over each parameter component and creates a string in the format "param:type". The parameter name is obtained from the 'name' key of the component dictionary, and the parameter type is obtained from the 'type' key of the component dictionary.

Finally, the method returns the list of stringified parameters.

Returns: - list[str]: A list of strings representing the parameter name and type in the format "param:type".

Example:

dispatcher = Dispatcher() parameters = dispatcher.stringify_parameters() print(parameters) ['param1:int', 'param2:str', 'param3:bool']

Note: - The parameter type is obtained using the __name__ attribute of the type object. - The parameter components are obtained by parsing the path of the dispatcher object. - If there are no parameter components in the path, an empty list will be returned.

parent_levels `property` ¶

parent_levels

Returns the handler from the app down to the route handler.

This property returns a list of all the parent levels of the current handler. Each parent level represents a higher level in the routing hierarchy.

Example: Consider the following routing hierarchy: app = Ravyn(routes=[ Include(path='/api/v1', routes=[ Gateway(path='/home', handler=home) ]) ])

In this example, the parent of the Gateway handler is the Include handler. The parent of the Include handler is the Ravyn router. The parent of the Ravyn router is the Ravyn app itself.

The parent_levels property uses a while loop to traverse the parent hierarchy. It starts with the current handler and iteratively adds each parent level to a list. Finally, it reverses the list to maintain the correct order of parent levels.

Returns: - list[Any]: A list of parent levels, starting from the current handler and going up to the app level.

Note: - The parent levels are determined based on the parent attribute of each handler. - If there are no parent levels (i.e., the current handler is the top-level handler), an empty list will be returned.

dependency_names `property` ¶

dependency_names

Returns a unique set of all dependency names provided in the handlers parent levels.

This property retrieves the dependencies from each parent level of the handler and collects all the dependency names in a set. It ensures that the set only contains unique dependency names.

Returns: - Set[str]: A set of unique dependency names.

Example:

handler = Dispatcher() dependency_names = handler.dependency_names print(dependency_names)

Note: - If no dependencies are defined in any of the parent levels, an empty set will be returned. - The dependencies are collected from all parent levels, ensuring that there are no duplicate dependency names in the final set.

path `instance-attribute` ¶

path = path

parent `instance-attribute` ¶

parent = parent or app

exception_handlers `instance-attribute` ¶

exception_handlers = exception_handlers or {}

interceptors `instance-attribute` ¶

interceptors = interceptors or []

lilya_permissions `instance-attribute` ¶

lilya_permissions = {
    index: (permission in __lilya_permissions__)
    for (index, permission) in (
        enumerate(__lilya_permissions__)
    )
}

tags `instance-attribute` ¶

tags = tags or []

name `instance-attribute` ¶

name = name

response_class `instance-attribute` ¶

response_class = response_class

response_cookies `instance-attribute` ¶

response_cookies = response_cookies or []

response_headers `instance-attribute` ¶

response_headers = response_headers or {}

security `instance-attribute` ¶

security = security or []

path_for ¶

path_for(name, **path_params)

Source code in ravyn/routing/router.py

def path_for(self, name: str, **path_params: Any) -> URLPath:
    warnings.warn(
        "Ravyn `path_for` is deprecated and will be removed in a future version. Use url_path_for instead.",
        UserWarning,
        stacklevel=2,
    )
    for route in self.routes or []:
        try:
            return cast("URLPath", route.path_for(name, **path_params))
        except NoMatchFound:
            ...

        if isinstance(route, (Gateway, WebSocketGateway)):
            handler = cast("Union[HTTPHandler, WebSocketHandler]", route.handler)
            try:
                return handler.path_for(name, **path_params)
            except NoMatchFound:
                ...

    raise NoMatchFound(name, path_params)

url_path_for ¶

url_path_for(name, **path_params)

Source code in ravyn/routing/router.py

def url_path_for(self, name: str, **path_params: Any) -> URLPath:
    for route in self.routes or []:
        try:
            return cast("URLPath", route.url_path_for(name, **path_params))
        except NoMatchFound:
            ...

        if isinstance(route, (Gateway, WebSocketGateway)):
            handler = cast("Union[HTTPHandler, WebSocketHandler]", route.handler)
            try:
                return handler.url_path_for(name, **path_params)
            except NoMatchFound:
                ...

    raise NoMatchFound(name, path_params)

startup `async` ¶

startup()

Runs the the events on startup.

Source code in lilya/routing.py

async def startup(self) -> None:
    """
    Runs the the events on startup.
    """
    for handler in self.on_startup:
        if is_async_callable(handler):
            await handler()
        else:
            handler()

shutdown `async` ¶

shutdown()

Runs the the events on startup.

Source code in lilya/routing.py

async def shutdown(self) -> None:
    """
    Runs the the events on startup.
    """
    for handler in self.on_shutdown:
        if is_async_callable(handler):
            await handler()
        else:
            handler()

handle_route `async` ¶

handle_route(route, path_handler)

Handle a route match.

PARAMETER	DESCRIPTION
`route`	The matched route. TYPE: `BasePath`
`path_handler`	The PathHandler. TYPE: `PathHandler`

Source code in lilya/routing.py

async def handle_route(self, route: BasePath, path_handler: PathHandler) -> None:
    """
    Handle a route match.

    Args:
        route: The matched route.
        path_handler (PathHandler): The PathHandler.
    """
    base_scope = path_handler.scope
    child = path_handler.child_scope or {}

    if "path_params" in child:
        base_scope.setdefault("path_params", {}).update(child["path_params"])
    # expose the matched route + template
    base_scope["route"] = route
    base_scope["route_path_template"] = getattr(route, "path", None)

    if self.dependency_overrides:
        base_scope["dependency_overrides"] = self.dependency_overrides

    new_scope = dict(base_scope)
    new_scope.update(child)
    await route.handle_dispatch(new_scope, path_handler.receive, path_handler.send)  # type: ignore

handle_partial `async` ¶

handle_partial(route, path_handler)

Handle a partial route match.

PARAMETER	DESCRIPTION
`route`	The matched route. TYPE: `BasePath`
`path_handler`	The PathHandler. TYPE: `PathHandler`

Source code in lilya/routing.py

async def handle_partial(self, route: BasePath, path_handler: PathHandler) -> None:
    """
    Handle a partial route match.

    Args:
        route: The matched route.
        path_handler (PathHandler): The PathHandler.
    """
    new_scope = dict(path_handler.scope)
    new_scope.update(path_handler.child_scope)
    await route.handle_dispatch(new_scope, path_handler.receive, path_handler.send)  # type: ignore

handle_default `async` ¶

handle_default(scope, receive, send)

Handle the default behavior.

PARAMETER	DESCRIPTION
`scope`	The ASGI scope. TYPE: `Scope`
`receive`	The ASGI receive channel. TYPE: `Receive`
`send`	The ASGI send channel. TYPE: `Send`

Source code in lilya/routing.py

async def handle_default(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Handle the default behavior.

    Args:
        scope (Scope): The ASGI scope.
        receive (Receive): The ASGI receive channel.
        send (Send): The ASGI send channel.
    """
    await self.default(scope, receive, send)

lifespan `async` ¶

lifespan(scope, receive, send)

Handle ASGI lifespan messages, managing application startup and shutdown events.

PARAMETER	DESCRIPTION
`scope`	The ASGI scope. TYPE: `Scope`
`receive`	The receive channel. TYPE: `Receive`
`send`	The send channel. TYPE: `Send`

Source code in lilya/routing.py

async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Handle ASGI lifespan messages, managing application startup and shutdown events.

    Args:
        scope (Scope): The ASGI scope.
        receive (Receive): The receive channel.
        send (Send): The send channel.
    """
    completed_startup = False

    async def startup_complete() -> None:
        await send({"type": "lifespan.startup.complete"})

    async def shutdown_complete() -> None:
        await send({"type": "lifespan.shutdown.complete"})

    async def lifespan_error(type: str, message: str) -> None:
        await send({"type": type, "message": message})

    try:
        app: Any = scope.get("app")
        await receive()

        # If it's a bound method (needs invoking to get the actual callable, example? From settings)
        if inspect.ismethod(self.lifespan_context):
            self.lifespan_context = (
                self.lifespan_context()
            )  # should return a (app)->AsyncContextManager callable

        async with self.lifespan_context(app) as state:
            if state is not None:
                if "state" not in scope:
                    raise RuntimeError(
                        'The server does not support "state" in the lifespan scope.'
                    )
                scope["state"].update(state)

            await startup_complete()
            completed_startup = True
            await receive()

    except BaseException:
        exc_text = traceback.format_exc()
        if completed_startup:
            await lifespan_error("lifespan.shutdown.failed", exc_text)
        else:
            await lifespan_error("lifespan.startup.failed", exc_text)
        raise

    else:
        await shutdown_complete()

app `async` ¶

app(scope, receive, send)

Handle ASGI messages, managing different scopes and routing.

PARAMETER	DESCRIPTION
`scope`	The ASGI scope. TYPE: `Scope`
`receive`	The ASGI receive channel. TYPE: `Receive`
`send`	The ASGI send channel. TYPE: `Send`

Source code in lilya/routing.py

async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Handle ASGI messages, managing different scopes and routing.

    Args:
        scope (Scope): The ASGI scope.
        receive (Receive): The ASGI receive channel.
        send (Send): The ASGI send channel.
    """
    assert scope["type"] in (
        ScopeType.HTTP,
        ScopeType.WEBSOCKET,
        ScopeType.LIFESPAN,
    )

    if "router" not in scope:
        scope["router"] = self

    sniffer = SendReceiveSniffer(receive, send)

    partial_matches: list[tuple[BaseRouter, BasePath, PathHandler]] = []
    had_match = False

    for route in self.routes:
        # we cannot continue when the sniffer detects a sent
        if sniffer.sent:
            return
        match, child_scope = route.search(scope)
        if match == Match.NONE:
            continue
        had_match = True
        path_handler = PathHandler(child_scope=child_scope, scope=scope, sniffer=sniffer)
        try:
            if match == Match.FULL:
                await self.handle_route(route, path_handler=path_handler)
                if sniffer.sent:
                    return  # type: ignore
                else:
                    sniffer.repeat_message = True
            elif match == Match.PARTIAL:
                partial_matches.append((self, route, path_handler))
        except PassPartialMatches as exc:
            # collect the partial matches from the sub-router
            partial_matches.extend(exc.partial_matches)
        except ContinueRouting:
            assert not sniffer.sent, "Cannot continue routing when an answer was already sent."
            sniffer.repeat_message = True
    # check if redirect would be possible, when no match was found
    if not had_match and self.redirect_slashes:
        route_path = get_route_path(scope)
        if scope["type"] == ScopeType.HTTP and route_path != "/":
            redirect_scope = dict(scope)
            if route_path.endswith("/"):
                redirect_scope["path"] = redirect_scope["path"].rstrip("/")
            else:
                redirect_scope["path"] = redirect_scope["path"] + "/"

            for route in self.routes:
                match, child_scope = route.search(redirect_scope)
                if match != Match.NONE:
                    redirect_url = URL.build_from_scope(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    await response(scope, receive, send)
                    return

    if self.is_sub_router:
        # now pass the partial matches to the upper router
        raise PassPartialMatches(partial_matches=partial_matches)
    # when in the main router, check partial matches (method does not match)
    # use for this the handler method of the sub-router to keep their modifications
    for router, route, path_handler in partial_matches:
        try:
            await router.handle_partial(route, path_handler)
            # should propagate back to this sniffer
            if sniffer.sent:
                return
            else:
                # rearm sniffer of the path handler (sub router)
                path_handler.sniffer.repeat_message = True
                sniffer.repeat_message = True
        except ContinueRouting:
            assert not sniffer.sent, "Cannot continue routing when an answer was already sent."
            path_handler.sniffer.repeat_message = True
            sniffer.repeat_message = True

    await self.handle_default(scope, receive, send)

handle_not_found `async` ¶

handle_not_found(scope, receive, send)

Handles the case when no match is found.

PARAMETER	DESCRIPTION
`scope`	The request scope. TYPE: `Scope`
`receive`	The receive channel. TYPE: `Receive`
`send`	The send channel. TYPE: `Send`

RETURNS	DESCRIPTION
`None`	None

Source code in lilya/routing.py

async def handle_not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Handles the case when no match is found.

    Args:
        scope (Scope): The request scope.
        receive (Receive): The receive channel.
        send (Send): The send channel.

    Returns:
        None
    """
    if "app" in scope:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

    if scope["type"] == ScopeType.HTTP:
        response = PlainText("Not Found", status_code=status.HTTP_404_NOT_FOUND)
        await response(scope, receive, send)

    elif scope["type"] == ScopeType.WEBSOCKET:
        websocket_close = WebSocketClose()
        await websocket_close(scope, receive, send)

include ¶

include(
    path,
    app,
    name=None,
    middleware=None,
    permissions=None,
    exception_handlers=None,
    dependencies=None,
    namespace=None,
    pattern=None,
    include_in_schema=True,
    before_request=None,
    after_request=None,
    redirect_slashes=True,
)

Adds an Include application into the routes.

Source code in lilya/routing.py

def include(
    self,
    path: str,
    app: ASGIApp,
    name: str | None = None,
    middleware: Sequence[DefineMiddleware] | None = None,
    permissions: Sequence[DefinePermission] | None = None,
    exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
    dependencies: Dependencies | None = None,
    namespace: str | None = None,
    pattern: str | None = None,
    include_in_schema: bool = True,
    before_request: Sequence[Callable[..., Any]] | None = None,
    after_request: Sequence[Callable[..., Any]] | None = None,
    redirect_slashes: bool = True,
) -> None:
    """
    Adds an Include application into the routes.
    """
    route = Include(
        path,
        app=app,
        name=name,
        middleware=middleware,
        permissions=permissions,
        exception_handlers=exception_handlers,
        dependencies=dependencies,
        namespace=namespace,
        pattern=pattern,
        include_in_schema=include_in_schema,
        before_request=before_request,
        after_request=after_request,
        redirect_slashes=redirect_slashes,
    )
    self.routes.append(route)

host ¶

host(host, app, name=None)

Adds a Host application into the routes.

Source code in lilya/routing.py

def host(self, host: str, app: ASGIApp, name: str | None = None) -> None:
    """
    Adds a Host application into the routes.
    """
    route = Host(host, app=app, name=name)
    self.routes.append(route)

add_event_handler ¶

add_event_handler(event_type, func)

Source code in ravyn/routing/router.py

def add_event_handler(self, event_type: str, func: Callable) -> None:  # pragma: no cover
    assert event_type in ("startup", "shutdown")

    if event_type == "startup":
        self.on_startup.append(func)

    else:
        self.on_shutdown.append(func)

on_event ¶

on_event(event_type)

Source code in ravyn/routing/router.py

def on_event(self, event_type: str) -> Callable:  # pragma: no cover
    def decorator(func: Callable) -> Callable:
        self.add_event_handler(event_type, func)
        return func

    return decorator

get ¶

get(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def get(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.GET.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

head ¶

head(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def head(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.HEAD.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

post ¶

post(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def post(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.POST.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

put ¶

put(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def put(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.PUT.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

patch ¶

patch(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def patch(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.PATCH.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

delete ¶

delete(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def delete(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.DELETE.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

trace ¶

trace(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def trace(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.TRACE.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

options ¶

options(
    path,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def options(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    return self.forward_single_method_route(
        path=path,
        method=HttpMethod.OPTIONS.value,
        dependencies=dependencies,
        interceptors=interceptors,
        permissions=permissions,
        exception_handlers=exception_handlers,
        middleware=middleware,
        name=name,
        include_in_schema=include_in_schema,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )

forward_single_method_route ¶

forward_single_method_route(path, method, **kwargs)

For customization, defaults to route.

Source code in ravyn/routing/router.py

def forward_single_method_route(self, path: str, method: str, **kwargs: Any) -> Callable:
    """For customization, defaults to route."""
    return self.route(path=path, methods=[method], **kwargs)

parse_path ¶

parse_path(path)

Using the Lilya TRANSFORMERS and the application registered convertors, transforms the path into a PathParameterSchema used for the OpenAPI definition.

Source code in ravyn/routing/core/base.py

def parse_path(self, path: str) -> list[Union[str, PathParameterSchema]]:
    """
    Using the Lilya TRANSFORMERS and the application registered convertors,
    transforms the path into a PathParameterSchema used for the OpenAPI definition.
    """
    _, path, variables, _ = compile_path(path)

    parsed_components: list[Union[str, PathParameterSchema]] = []

    for name, convertor in variables.items():
        _type = CONV2TYPE[convertor]
        parsed_components.append(
            PathParameterSchema(name=name, type=param_type_map[_type], full=name)
        )
    return parsed_components

get_response_for_handler ¶

get_response_for_handler()

Checks and validates the type of return response and maps to the corresponding handler with the given parameters.

RETURNS	DESCRIPTION
`Callable[[Any], Awaitable[Response]]`	Callable[[Any], Awaitable[LilyaResponse]]: The response handler function.

Source code in ravyn/routing/core/base.py

def get_response_for_handler(self) -> Callable[[Any], Awaitable[LilyaResponse]]:
    """
    Checks and validates the type of return response and maps to the corresponding
    handler with the given parameters.

    Returns:
        Callable[[Any], Awaitable[LilyaResponse]]: The response handler function.
    """
    if self._response_handler is not Void:
        return cast("Callable[[Any], Awaitable[LilyaResponse]]", self._response_handler)

    media_type = (
        self.media_type.value if isinstance(self.media_type, Enum) else self.media_type
    )

    response_class = self.get_response_class()
    headers = self.get_response_headers()
    cookies = self.get_response_cookies()

    if is_class_and_subclass(self.handler_signature.return_annotation, ResponseContainer):
        handler = self._get_response_container_handler(cookies, headers, media_type)
    elif is_class_and_subclass(self.handler_signature.return_annotation, JSONResponse):
        handler = self._get_json_response_handler(cookies, headers)  # type: ignore[assignment]
    elif is_class_and_subclass(self.handler_signature.return_annotation, Response):
        handler = self._get_response_handler(cookies, headers, media_type)  # type: ignore[assignment]
    elif is_class_and_subclass(self.handler_signature.return_annotation, LilyaResponse):
        handler = self._get_lilya_response_handler(cookies, headers)  # type: ignore[assignment]
    else:
        handler = self._get_default_handler(cookies, headers, media_type, response_class)  # type: ignore[assignment]

    self._response_handler = handler

    return cast(
        Callable[[Any], Awaitable[LilyaResponse]],
        self._response_handler,
    )

get_response_for_request `async` ¶

get_response_for_request(
    scope, request, route, parameter_model
)

Get response for the given request using the specified route and parameter model.

PARAMETER	DESCRIPTION
`scope`	The scope of the request. TYPE: `Scope`
`request`	The incoming request. TYPE: `Request`
`route`	The route handler for the request. TYPE: `HTTPHandler`
`parameter_model`	The parameter model for handling request parameters. TYPE: `TransformerModel`

RETURNS	DESCRIPTION
`LilyaResponse`	The response generated for the request. TYPE: `'LilyaResponse'`

Source code in ravyn/routing/core/base.py

async def get_response_for_request(
    self,
    scope: "Scope",
    request: Request,
    route: "HTTPHandler",
    parameter_model: "TransformerModel",
) -> "LilyaResponse":
    """
    Get response for the given request using the specified route and parameter model.

    Args:
        scope (Scope): The scope of the request.
        request (Request): The incoming request.
        route (HTTPHandler): The route handler for the request.
        parameter_model (TransformerModel): The parameter model for handling request parameters.

    Returns:
        LilyaResponse: The response generated for the request.
    """
    response_data = await self._get_response_data(
        route=route,
        parameter_model=parameter_model,
        request=request,
    )

    response = await self.to_response(
        app=scope["app"],
        data=response_data,
    )
    return cast("LilyaResponse", response)

create_signature_model ¶

create_signature_model(is_websocket=False)

Creates a signature model for the given route.

Websockets do not support methods.

Source code in ravyn/routing/core/base.py

def create_signature_model(self, is_websocket: bool = False) -> None:
    """
    Creates a signature model for the given route.

    Websockets do not support methods.
    """
    if not self.signature_model:
        self.signature_model = SignatureFactory(
            fn=cast(AnyCallable, self.fn),
            dependency_names=self.dependency_names,
        ).create_signature()

    for dependency in list(self.get_dependencies().values()):
        if not dependency.signature_model:
            dependency.signature_model = SignatureFactory(
                fn=dependency.dependency, dependency_names=self.dependency_names
            ).create_signature()

    transformer_model = self.create_handler_transformer_model()
    if not is_websocket:
        self.transformer = transformer_model
        for method in self.methods:
            self.route_map[method] = (self, transformer_model)
    else:
        self.websocket_parameter_model = transformer_model

create_handler_transformer_model ¶

create_handler_transformer_model()

Method to create a TransformerModel for a given handler.

Source code in ravyn/routing/core/base.py

def create_handler_transformer_model(self) -> TransformerModel:
    """Method to create a TransformerModel for a given handler."""
    dependencies = self.get_dependencies()
    signature_model = get_signature(self)

    return transformer_create_signature(
        signature_model=signature_model,
        dependencies=dependencies,
        path_parameters=self.path_parameters,
    )

get_lookup_path ¶

get_lookup_path(ignore_first=True)

Constructs and returns the lookup path for the current object by traversing its parent hierarchy.

The method collects the 'name' attribute of the current object and its ancestors, if they exist, and returns them as a list in reverse order (from the root ancestor to the current object).

RETURNS	DESCRIPTION
`list[str]`	list[str]: A list of names representing the lookup path from the root
`list[str]`	ancestor to the current object.

Source code in ravyn/routing/core/base.py

def get_lookup_path(self, ignore_first: bool = True) -> list[str]:
    """
    Constructs and returns the lookup path for the current object by traversing
    its parent hierarchy.

    The method collects the 'name' attribute of the current object and its
    ancestors, if they exist, and returns them as a list in reverse order
    (from the root ancestor to the current object).

    Returns:
        list[str]: A list of names representing the lookup path from the root
        ancestor to the current object.
    """

    names = []
    current: Any = self
    counter: int = 0 if ignore_first else 1

    while current:
        if getattr(current, "name", None) is not None:
            if counter >= 1:
                names.append(current.name)
        current = current.parent
        counter += 1
    return list(reversed(names))

get_dependencies ¶

get_dependencies()

Returns all dependencies of the handler function's starting from the parent levels.

This method retrieves all the dependencies of the handler function by iterating over each parent level. It collects the dependencies defined in each level and stores them in a dictionary.

Returns: - Dependencies: A dictionary containing all the dependencies of the handler function.

Raises: - RuntimeError: If get_dependencies is called before a signature model has been generated.

Example:

handler = Dispatcher() dependencies = handler.get_dependencies() print(dependencies)

Note: - If no dependencies are defined in any of the parent levels, an empty dictionary will be returned. - Each dependency is represented by a key-value pair in the dictionary, where the key is the dependency name and the value is the dependency object. - The dependencies are collected from all parent levels, ensuring that there are no duplicate dependencies in the final dictionary.

Source code in ravyn/routing/core/base.py

def get_dependencies(self) -> Dependencies:
    """
    Returns all dependencies of the handler function's starting from the parent levels.

    This method retrieves all the dependencies of the handler function by iterating over each parent level.
    It collects the dependencies defined in each level and stores them in a dictionary.

    Returns:
    - Dependencies: A dictionary containing all the dependencies of the handler function.

    Raises:
    - RuntimeError: If `get_dependencies` is called before a signature model has been generated.

    Example:
    >>> handler = Dispatcher()
    >>> dependencies = handler.get_dependencies()
    >>> print(dependencies)

    Note:
    - If no dependencies are defined in any of the parent levels, an empty dictionary will be returned.
    - Each dependency is represented by a key-value pair in the dictionary, where the key is the dependency name and the value is the dependency object.
    - The dependencies are collected from all parent levels, ensuring that there are no duplicate dependencies in the final dictionary.
    """
    if not self.signature_model:
        raise RuntimeError(
            "get_dependencies cannot be called before a signature model has been generated"
        )

    if not self._dependencies or self._dependencies is Void:
        self._dependencies: Dependencies = {}
        for level in self.parent_levels:
            for key, value in (level.dependencies or {}).items():
                if not isinstance(value, Inject):
                    value = Inject(value)
                self.is_unique_dependency(
                    dependencies=self._dependencies,
                    key=key,
                    injector=value,
                )
                self._dependencies[key] = value  # type: ignore[assignment]
    return self._dependencies

is_unique_dependency `staticmethod` ¶

is_unique_dependency(dependencies, key, injector)

Validates that a given inject has not been already defined under a different key in any of the levels.

This method takes in a dictionary of dependencies, a key, and an injector. It checks if the injector is already defined in the dependencies dictionary under a different key.

Parameters: - dependencies (Dependencies): A dictionary of dependencies. - key (str): The key to check for uniqueness. - injector (Inject): The injector to check.

Raises: - ImproperlyConfigured: If the injector is already defined under a different key in the dependencies dictionary.

Example:

dependencies = {"db": injector1, "logger": injector2} key = "db" injector = injector3 is_unique_dependency(dependencies, key, injector)

This method iterates over each key-value pair in the dependencies dictionary. If the value matches the given injector, it raises an ImproperlyConfigured exception with a detailed error message.

Note: - The dependencies dictionary is expected to have string keys and values of type Inject. - The key parameter should be a string representing the key to check for uniqueness. - The injector parameter should be an instance of the Inject class.

Source code in ravyn/routing/core/base.py

@staticmethod
def is_unique_dependency(dependencies: Dependencies, key: str, injector: Inject) -> None:
    """
    Validates that a given inject has not been already defined under a different key in any of the levels.

    This method takes in a dictionary of dependencies, a key, and an injector. It checks if the injector is already defined in the dependencies dictionary under a different key.

    Parameters:
    - dependencies (Dependencies): A dictionary of dependencies.
    - key (str): The key to check for uniqueness.
    - injector (Inject): The injector to check.

    Raises:
    - ImproperlyConfigured: If the injector is already defined under a different key in the dependencies dictionary.

    Example:
    >>> dependencies = {"db": injector1, "logger": injector2}
    >>> key = "db"
    >>> injector = injector3
    >>> is_unique_dependency(dependencies, key, injector)

    This method iterates over each key-value pair in the dependencies dictionary. If the value matches the given injector, it raises an ImproperlyConfigured exception with a detailed error message.

    Note:
    - The dependencies dictionary is expected to have string keys and values of type Inject.
    - The key parameter should be a string representing the key to check for uniqueness.
    - The injector parameter should be an instance of the Inject class.
    """
    for dependency_key, value in dependencies.items():
        if injector == value:
            raise ImproperlyConfigured(
                f"Injector for key {key} is already defined under the different key {dependency_key}. "
                f"If you wish to override a inject, it must have the same key."
            )

get_cookies ¶

get_cookies(local_cookies, other_cookies=None)

Returns a unique list of cookies.

This method takes two sets of cookies, local_cookies and other_cookies, and returns a list of dictionaries representing the normalized cookies.

Parameters: - local_cookies (ResponseCookies): The set of local cookies. - other_cookies (ResponseCookies): The set of other cookies.

Returns: - list[dict[str, Any]]: A list of dictionaries representing the normalized cookies.

The method first creates a filtered list of cookies by combining the local_cookies and other_cookies sets. It ensures that only unique cookies are included in the list.

Then, it normalizes each cookie by converting it into a dictionary representation, excluding the 'description' attribute. The normalized cookies are stored in a list.

Finally, the method returns the list of normalized cookies.

Note: - The 'description' attribute is excluded from the normalized cookies.

Example usage:

local_cookies = [...]
other_cookies = [...]
normalized_cookies = get_cookies(local_cookies, other_cookies)
print(normalized_cookies)

This will output the list of normalized cookies.

Source code in ravyn/routing/core/base.py

def get_cookies(
    self,
    local_cookies: ResponseCookies | None,
    other_cookies: ResponseCookies | None = None,
) -> list[dict[str, Any]]:  # pragma: no cover
    """
    Returns a unique list of cookies.

    This method takes two sets of cookies, `local_cookies` and `other_cookies`,
    and returns a list of dictionaries representing the normalized cookies.

    Parameters:
    - local_cookies (ResponseCookies): The set of local cookies.
    - other_cookies (ResponseCookies): The set of other cookies.

    Returns:
    - list[dict[str, Any]]: A list of dictionaries representing the normalized cookies.

    The method first creates a filtered list of cookies by combining the `local_cookies`
    and `other_cookies` sets. It ensures that only unique cookies are included in the list.

    Then, it normalizes each cookie by converting it into a dictionary representation,
    excluding the 'description' attribute. The normalized cookies are stored in a list.

    Finally, the method returns the list of normalized cookies.

    Note:
    - The 'description' attribute is excluded from the normalized cookies.

    Example usage:
    ```
    local_cookies = [...]
    other_cookies = [...]
    normalized_cookies = get_cookies(local_cookies, other_cookies)
    print(normalized_cookies)
    ```

    This will output the list of normalized cookies.
    """
    filtered_cookies: dict[str, Cookie] = {}
    for cookie in chain(local_cookies or _empty, other_cookies or _empty):
        filtered_cookies.setdefault(cookie.key, cookie)
    return [
        cookie.model_dump(exclude_none=True, exclude={"description"})
        for cookie in filtered_cookies.values()
    ]

get_headers ¶

get_headers(headers)

Returns a dictionary of response headers.

Parameters: - headers (ResponseHeaders): The response headers object.

Returns: - dict[str, Any]: A dictionary containing the response headers.

Example:

headers = {"Content-Type": "application/json", "Cache-Control": "no-cache"} response_headers = get_headers(headers) print(response_headers)

This method takes a ResponseHeaders object and converts it into a dictionary of response headers. Each key-value pair in the ResponseHeaders object is added to the dictionary.

Note: - The ResponseHeaders object is expected to have string keys and values. - If the ResponseHeaders object is empty, an empty dictionary will be returned.

Source code in ravyn/routing/core/base.py

def get_headers(self, headers: ResponseHeaders) -> dict[str, Any]:
    """
    Returns a dictionary of response headers.

    Parameters:
    - headers (ResponseHeaders): The response headers object.

    Returns:
    - dict[str, Any]: A dictionary containing the response headers.

    Example:
    >>> headers = {"Content-Type": "application/json", "Cache-Control": "no-cache"}
    >>> response_headers = get_headers(headers)
    >>> print(response_headers)
    {'Content-Type': 'application/json', 'Cache-Control': 'no-cache'}

    This method takes a `ResponseHeaders` object and converts it into a dictionary
    of response headers. Each key-value pair in the `ResponseHeaders` object is
    added to the dictionary.

    Note:
    - The `ResponseHeaders` object is expected to have string keys and values.
    - If the `ResponseHeaders` object is empty, an empty dictionary will be returned.
    """
    return {k: v.value for k, v in headers.items()}

get_response_data `async` ¶

get_response_data(data)

Retrieves the response data for synchronous and asynchronous operations.

This method takes in a data parameter, which can be either a regular value or an awaitable object. If data is an awaitable object, it will be awaited to retrieve the actual response data. If data is a regular value, it will be returned as is.

Parameters: - data (Any): The response data, which can be either a regular value or an awaitable object.

Returns: - Any: The actual response data.

Example usage:

response_data = await get_response_data(some_data)

Source code in ravyn/routing/core/base.py

async def get_response_data(self, data: Any) -> Any:  # pragma: no cover
    """
    Retrieves the response data for synchronous and asynchronous operations.

    This method takes in a `data` parameter, which can be either a regular value or an awaitable object.
    If `data` is an awaitable object, it will be awaited to retrieve the actual response data.
    If `data` is a regular value, it will be returned as is.

    Parameters:
    - data (Any): The response data, which can be either a regular value or an awaitable object.

    Returns:
    - Any: The actual response data.

    Example usage:
    ```
    response_data = await get_response_data(some_data)
    ```
    """
    if isawaitable(data):
        data = await data
    return data

allow_connection `async` ¶

allow_connection(connection, permission)

Asynchronously allows a connection based on the provided permission.

PARAMETER	DESCRIPTION
`permission`	The permission object to check. TYPE: `BasePermission`
`connection`	The connection object representing the request. TYPE: `Connection`

Returns: None Raises: PermissionException: If the permission check fails.

Source code in ravyn/routing/core/base.py

async def allow_connection(
    self, connection: "Connection", permission: AsyncCallable
) -> None:  # pragma: no cover
    """
    Asynchronously allows a connection based on the provided permission.

    Args:
        permission (BasePermission): The permission object to check.
        connection (Connection): The connection object representing the request.
    Returns:
        None
    Raises:
        PermissionException: If the permission check fails.
    """
    awaitable: BasePermission = cast("BasePermission", await permission())
    request: Request = cast("Request", connection)
    handler = cast("APIGateHandler", self)
    await continue_or_raise_permission_exception(request, handler, awaitable)

dispatch_allow_connection `async` ¶

dispatch_allow_connection(
    permissions,
    connection,
    scope,
    receive,
    send,
    dispatch_call,
)

Dispatches a connection based on the provided permissions.

PARAMETER	DESCRIPTION
`permissions`	A dictionary mapping permission levels to either an asynchronous callable or a DefinePermission instance. TYPE: `dict[int, Union[AsyncCallable, DefinePermission]]`
`connection`	The connection object to be dispatched. TYPE: `Connection`
`scope`	The scope of the connection. TYPE: `Scope`
`receive`	The receive channel for the connection. TYPE: `Receive`
`send`	The send channel for the connection. TYPE: `Send`

Returns: None

Source code in ravyn/routing/core/base.py

async def dispatch_allow_connection(
    self,
    permissions: dict[int, Union[AsyncCallable, DefinePermission]] | Any,
    connection: "Connection",
    scope: Scope,
    receive: Receive,
    send: Send,
    dispatch_call: Callable[..., Awaitable[None]],
) -> None:  # pragma: no cover
    """
    Dispatches a connection based on the provided permissions.

    Args:
        permissions (dict[int, Union[AsyncCallable, DefinePermission]]):
            A dictionary mapping permission levels to either an asynchronous
            callable or a DefinePermission instance.
        connection (Connection): The connection object to be dispatched.
        scope (Scope): The scope of the connection.
        receive (Receive): The receive channel for the connection.
        send (Send): The send channel for the connection.
    Returns:
        None
    """
    for _, permission in permissions.items():
        if isinstance(permission, AsyncCallable):
            await self.allow_connection(connection, permission)
        else:
            # Dispatches to lilya permissions
            await dispatch_call(scope, receive, send)

get_security_schemes ¶

get_security_schemes()

Returns a list of all security schemes associated with the handler.

This method iterates over each parent level of the handler and collects the security schemes defined in each level. The collected security schemes are stored in a list and returned.

Returns: - list[SecurityScheme]: A list of security schemes associated with the handler.

Example:

handler = Dispatcher() security_schemes = handler.get_security_schemes() print(security_schemes) [SecurityScheme(name='BearerAuth', type='http', scheme='bearer', bearer_format='JWT'), SecurityScheme(name='ApiKeyAuth', type='apiKey', in_='header', name='X-API-Key')]

Note: - If no security schemes are defined in any of the parent levels, an empty list will be returned. - Each security scheme is represented by an instance of the SecurityScheme class. - The SecurityScheme class has attributes such as name, type, scheme, bearer_format, in_, and name, which provide information about the security scheme.

Source code in ravyn/routing/core/base.py

def get_security_schemes(self) -> list[SecurityScheme]:
    """
    Returns a list of all security schemes associated with the handler.

    This method iterates over each parent level of the handler and collects the security schemes defined in each level.
    The collected security schemes are stored in a list and returned.

    Returns:
    - list[SecurityScheme]: A list of security schemes associated with the handler.

    Example:
    >>> handler = Dispatcher()
    >>> security_schemes = handler.get_security_schemes()
    >>> print(security_schemes)
    [SecurityScheme(name='BearerAuth', type='http', scheme='bearer', bearer_format='JWT'), SecurityScheme(name='ApiKeyAuth', type='apiKey', in_='header', name='X-API-Key')]

    Note:
    - If no security schemes are defined in any of the parent levels, an empty list will be returned.
    - Each security scheme is represented by an instance of the SecurityScheme class.
    - The SecurityScheme class has attributes such as name, type, scheme, bearer_format, in_, and name, which provide information about the security scheme.
    """
    security_schemes: list[SecurityScheme] = []
    for layer in self.parent_levels:
        security_schemes.extend(layer.security or [])
    return security_schemes

get_handler_tags ¶

get_handler_tags()

Returns all the tags associated with the handler by checking the parents as well.

This method retrieves all the tags associated with the handler by iterating over each parent level. It collects the tags defined in each level and stores them in a list.

Returns: - list[str]: A list of tags associated with the handler.

Example:

handler = Dispatcher() tags = handler.get_handler_tags() print(tags) ['api', 'user']

Note: - If no tags are defined in any of the parent levels, an empty list will be returned. - Each tag is represented as a string. - The tags are collected from all parent levels, ensuring that there are no duplicate tags in the final list.

Source code in ravyn/routing/core/base.py

def get_handler_tags(self) -> list[str]:
    """
    Returns all the tags associated with the handler by checking the parents as well.

    This method retrieves all the tags associated with the handler by iterating over each parent level.
    It collects the tags defined in each level and stores them in a list.

    Returns:
    - list[str]: A list of tags associated with the handler.

    Example:
    >>> handler = Dispatcher()
    >>> tags = handler.get_handler_tags()
    >>> print(tags)
    ['api', 'user']

    Note:
    - If no tags are defined in any of the parent levels, an empty list will be returned.
    - Each tag is represented as a string.
    - The tags are collected from all parent levels, ensuring that there are no duplicate tags in the final list.
    """
    tags: list[str] = []
    for layer in self.parent_levels:
        tags.extend(layer.tags or [])

    tags_clean: list[str] = []
    for tag in tags:
        if tag not in tags_clean:
            tags_clean.append(tag)

    return tags_clean if tags_clean else None

is_member_descriptor ¶

is_member_descriptor(value)

Checks if the value is a member descriptor type. This is used to determine if the value is a class attribute or not.

Source code in ravyn/routing/router.py

def is_member_descriptor(self, value: Any) -> bool:
    """
    Checks if the value is a member descriptor type.
    This is used to determine if the value is a class attribute or not.
    """
    return isinstance(value, types.MemberDescriptorType)

reorder_routes ¶

reorder_routes()

Source code in ravyn/routing/router.py

def reorder_routes(self) -> list[Sequence[Union[APIGateHandler, Include]]]:
    routes = sorted(
        self.routes,
        key=lambda router: router.path != "" and router.path != "/",
        reverse=True,
    )
    return routes

activate ¶

activate()

Source code in ravyn/routing/router.py

def activate(self) -> None:
    self.routes = self.reorder_routes()

handle_interceptors `async` ¶

handle_interceptors(scope, receive, send)

Source code in ravyn/routing/router.py

async def handle_interceptors(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
    # Inherit interceptors from parent if any
    if self.parent and self.parent.interceptors:
        for parent_interceptor in self.parent.interceptors:
            if parent_interceptor not in self.interceptors:
                self.interceptors.insert(0, parent_interceptor)

    # If no factories defined and no cache, nothing to do
    if (self._interceptors is Void) and not self.interceptors:
        return

    # Instantiate once and cache for reuse across requests
    if self._interceptors is Void:
        self._interceptors = [factory() for factory in self.interceptors]

    for interceptor in self._interceptors:
        if is_async_callable(interceptor.intercept):
            await interceptor.intercept(scope, receive, send)
        else:
            await run_in_threadpool(interceptor.intercept, scope, receive, send)

handle_permissions `async` ¶

handle_permissions(scope, receive, send)

Handles the permissions for the Include. This method ensures that the permissions are set correctly and that they are compatible with the Lilya permissions system.

Source code in ravyn/routing/router.py

async def handle_permissions(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Handles the permissions for the Include.
    This method ensures that the permissions are set correctly
    and that they are compatible with the Lilya permissions system.
    """
    # Use cached permissions if already computed
    if self._permissions_cache is not Void or self._lilya_permissions_cache is not Void:
        effective_permissions = (
            self._permissions_cache if self._permissions_cache is not Void else {}
        )
        effective_lilya_permissions = (
            self._lilya_permissions_cache if self._lilya_permissions_cache is not Void else {}
        )
    else:
        if self.parent and self.parent.permissions:
            # If the parent has permissions, we need to merge them with the current permissions
            self.permissions.update(
                {
                    index + len(self.permissions): wrap_permission(permission)
                    for index, permission in enumerate(self.parent.permissions)
                    if is_ravyn_permission(permission)
                }
            )

        effective_permissions = self.permissions
        effective_lilya_permissions = self.lilya_permissions
        # Cache the computed permissions
        self._permissions_cache = effective_permissions or Void
        self._lilya_permissions_cache = effective_lilya_permissions or Void

    if not effective_permissions and not effective_lilya_permissions:
        return

    connection = Connection(scope=scope, receive=receive)

    if effective_lilya_permissions:
        await self.dispatch_allow_connection(
            effective_lilya_permissions,
            connection,
            scope,
            receive,
            send,
            dispatch_call=self.app,
        )
    else:
        await self.dispatch_allow_connection(
            effective_permissions,
            connection,
            scope,
            receive,
            send,
            dispatch_call=self.app,
        )

not_found `async` ¶

not_found(scope, receive, send)

Ravyn version of a not found handler when a resource is called and cannot be dealt with properly.

Ravyn overrides the original Lilya not_found to return always JSONReponse.

For plain ASGI apps, just returns the response.

Source code in ravyn/routing/router.py

async def not_found(
    self, scope: "Scope", receive: "Receive", send: "Send"
) -> None:  # pragma: no cover
    """Ravyn version of a not found handler when a resource is
    called and cannot be dealt with properly.

    Ravyn overrides the original Lilya not_found to return always
    JSONReponse.

    For plain ASGI apps, just returns the response.
    """
    if scope["type"] == "websocket":
        websocket_close = WebSocketClose()
        await websocket_close(scope, receive, send)
        return

    if "app" in scope:
        raise NotFound(status_code=status.HTTP_404_NOT_FOUND)
    else:
        response = JSONResponse({"detail": "Not Found"}, status_code=status.HTTP_404_NOT_FOUND)
    await response(scope, receive, send)

create_signature_models ¶

create_signature_models(route)

Creates the signature models for the given routes.

PARAMETER	DESCRIPTION
`route`	The route for the signature model to be created. TYPE: `RouteParent`

Source code in ravyn/routing/router.py

def create_signature_models(self, route: RouteParent) -> None:
    """
    Creates the signature models for the given routes.

    Args:
        route: The route for the signature model to be created.
    """
    if isinstance(route, (Include, Host)):
        for _route in route.routes:
            self.create_signature_models(_route)

    if isinstance(route, (Gateway, WebhookGateway)):
        if not route.handler.parent:  # pragma: no cover
            route.handler.parent = route

        if not is_class_and_subclass(route.handler, BaseController) and not isinstance(
            route.handler, BaseController
        ):
            route.handler.create_signature_model()

    if isinstance(route, WebSocketGateway):
        route.handler.create_signature_model(is_websocket=True)

validate_root_route_parent ¶

validate_root_route_parent(value, override=False)

Handles everything parent from the root. When in the root, the parent must be setup. Appends the route path if exists.

Source code in ravyn/routing/router.py

def validate_root_route_parent(
    self,
    value: Union[Router, Include, Gateway, WebSocketGateway, WebhookGateway],
    override: bool = False,
) -> None:
    """
    Handles everything parent from the root. When in the root, the parent must be setup.
    Appends the route path if exists.
    """
    # Getting the value of the router for the path
    value.path = clean_path(self.path + getattr(value, "path", "/"))

    if isinstance(value, (Include, Gateway, WebSocketGateway, WebhookGateway)):
        if not value.parent and not override:
            value.parent = cast("Union[Router, Include, Gateway, WebSocketGateway]", self)

    if isinstance(value, (Gateway, WebSocketGateway, WebhookGateway)):
        if not is_class_and_subclass(value.handler, BaseController) and not isinstance(
            value.handler, BaseController
        ):
            if not value.handler.parent:
                value.handler.parent = value
        else:
            if not value.handler.parent:  # pragma: no cover
                value(parent=self)  # type: ignore

            handler: BaseController = cast("BaseController", value.handler)
            route_handlers = handler.get_routes(
                path=value.path,
                middleware=value.middleware,
                permissions=value.permissions,
                interceptors=value.interceptors,
                exception_handlers=value.exception_handlers,
            )
            if route_handlers:
                self.routes.extend(route_handlers)
            self.routes.pop(self.routes.index(value))

add_apiview ¶

add_apiview(value)

Adds an Controller or related to the application routing.

Example

from ravyn import Router, Controller, Gateway, get


class BaseController(Controller):
    path = "/"

    @get(status_code=status_code)
    async def hello(self) -> str:
        return "Hello, World!"


gateway = Gateway(handler=View)

app = Router()
app.add_apiview(value=gateway)

PARAMETER	DESCRIPTION
`value`	The `Controller` or similar to be added. TYPE: `Union[Gateway, WebSocketGateway]`

Source code in ravyn/routing/router.py

def add_apiview(
    self,
    value: Annotated[
        Union[Gateway, WebSocketGateway],
        Doc(
            """
            The `Controller` or similar to be added.
            """
        ),
    ],
) -> None:
    """
    Adds an [Controller](https://ravyn.dev/routing/controllers/) or related
    to the application routing.

    **Example**

    ```python
    from ravyn import Router, Controller, Gateway, get


    class BaseController(Controller):
        path = "/"

        @get(status_code=status_code)
        async def hello(self) -> str:
            return "Hello, World!"


    gateway = Gateway(handler=View)

    app = Router()
    app.add_apiview(value=gateway)
    ```
    """
    warnings.warn(
        "add_apiview is deprecated and will be removed in the release 0.4.0. "
        "Please use add_controller instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    self.add_controller(value=value)

add_controller ¶

add_controller(value)

Adds an Controller or related to the application routing.

Example

from ravyn import Router, Controller, Gateway, get


class View(Controller):
    path = "/"

    @get(status_code=status_code)
    async def hello(self) -> str:
        return "Hello, World!"


gateway = Gateway(handler=View)

app = Router()
app.add_controller(value=gateway)

PARAMETER	DESCRIPTION
`value`	The `Controller` or similar to be added. TYPE: `Union[Gateway, WebSocketGateway]`

Source code in ravyn/routing/router.py

def add_controller(
    self,
    value: Annotated[
        Union[Gateway, WebSocketGateway],
        Doc(
            """
            The `Controller` or similar to be added.
            """
        ),
    ],
) -> None:
    """
    Adds an [Controller](https://ravyn.dev/routing/controllers/) or related
    to the application routing.

    **Example**

    ```python
    from ravyn import Router, Controller, Gateway, get


    class View(Controller):
        path = "/"

        @get(status_code=status_code)
        async def hello(self) -> str:
            return "Hello, World!"


    gateway = Gateway(handler=View)

    app = Router()
    app.add_controller(value=gateway)
    ```
    """
    routes = []
    if not value.handler.parent:  # pragma: no cover
        value.handler(parent=self)

    route_handlers: list[Union[HTTPHandler, WebSocketHandler]] = value.handler.get_routes(
        path=value.path,
        middleware=value.middleware,
        interceptors=value.interceptors,
        permissions=value.permissions,
        exception_handlers=value.exception_handlers,
        include_in_schema=value.include_in_schema,
        before_request=value.before_request,
        after_request=value.after_request,
    )
    if route_handlers:
        self.routes.extend(route_handlers)
        routes.extend(route_handlers)

    for route in routes or []:
        self.create_signature_models(route)
    self.activate()

add_route ¶

add_route(
    path,
    handler,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

Adds a Route to the application routing.

This is a dynamic way of adding routes on the fly.

Example

from ravyn import get


@get(status_code=status_code)
async def hello(self) -> str:
    return "Hello, World!"


app = Ravyn()
app.add_route(path="/hello", handler=hello)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`handler`	An instance of handler. TYPE: `HTTPHandler`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def add_route(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    handler: Annotated[
        HTTPHandler,
        Doc(
            """
            An instance of [handler](https://ravyn.dev/routing/handlers/#http-handlers).
            """
        ),
    ],
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> None:
    """
    Adds a [Route](https://ravyn.dev/routing/routes/)
    to the application routing.

    This is a dynamic way of adding routes on the fly.

    **Example**

    ```python
    from ravyn import get


    @get(status_code=status_code)
    async def hello(self) -> str:
        return "Hello, World!"


    app = Ravyn()
    app.add_route(path="/hello", handler=hello)
    ```
    """
    if not isinstance(handler, HTTPHandler) or isinstance(handler, WebhookHandler):
        raise ImproperlyConfigured(
            f"handler must be a instance of HTTPHandler and not {handler.__class__.__name__}. "
            "Example: get(), put(), post(), delete(), patch(), route()."
        )
    gateway = Gateway(
        path=self.path + path,
        handler=handler,
        name=name,
        include_in_schema=include_in_schema,
        dependencies=dependencies,
        exception_handlers=exception_handlers,
        interceptors=interceptors,
        permissions=permissions,
        middleware=middleware,
        deprecated=deprecated,
        before_request=before_request,
        after_request=after_request,
    )
    self.validate_root_route_parent(gateway)
    self.create_signature_models(gateway)
    self.routes.append(gateway)

add_websocket_route ¶

add_websocket_route(
    path,
    handler,
    name=None,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    before_request=None,
    after_request=None,
)

Adds a websocket Route to the application routing.

This is a dynamic way of adding routes on the fly.

Example

from ravyn import websocket


@websocket()
async def websocket_route(socket: WebSocket) -> None:
    await socket.accept()
    data = await socket.receive_json()

    assert data
    await socket.send_json({"data": "ravyn"})
    await socket.close()


app = Ravyn()
app.add_websocket_route(path="/ws", handler=websocket_route)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`handler`	An instance of websocket handler. TYPE: `WebSocketHandler`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def add_websocket_route(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    handler: Annotated[
        WebSocketHandler,
        Doc(
            """
            An instance of [websocket handler](https://ravyn.dev/routing/handlers/#websocket-handler).
            """
        ),
    ],
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> None:
    """
    Adds a websocket [Route](https://ravyn.dev/routing/routes/)
    to the application routing.

    This is a dynamic way of adding routes on the fly.

    **Example**

    ```python
    from ravyn import websocket


    @websocket()
    async def websocket_route(socket: WebSocket) -> None:
        await socket.accept()
        data = await socket.receive_json()

        assert data
        await socket.send_json({"data": "ravyn"})
        await socket.close()


    app = Ravyn()
    app.add_websocket_route(path="/ws", handler=websocket_route)
    ```
    """
    if not isinstance(handler, WebSocketHandler):
        raise ImproperlyConfigured(
            f"handler must be a instance of WebSocketHandler and not {handler.__class__.__name__}. "
            "Example: websocket()."
        )
    websocket_gateway = WebSocketGateway(
        path=path,
        handler=handler,
        name=name,
        dependencies=dependencies,
        exception_handlers=exception_handlers,
        interceptors=interceptors,
        permissions=permissions,
        middleware=middleware,
        before_request=before_request,
        after_request=after_request,
    )
    self.validate_root_route_parent(websocket_gateway)
    self.create_signature_models(websocket_gateway)
    self.routes.append(websocket_gateway)

route ¶

route(
    path,
    methods=None,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    name=None,
    include_in_schema=True,
    deprecated=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`methods`	A list of HTTP methods to serve the Gateway. TYPE: `Optional[list[str]]` DEFAULT: `None`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def route(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    methods: Annotated[
        Optional[list[str]],
        Doc(
            """
            A list of HTTP methods to serve the Gateway.
            """
        ),
    ] = None,
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    if methods is None:
        methods = [HttpMethod.GET.value]

    def wrapper(func: Callable) -> Callable:
        handler = HTTPHandler(
            handler=func,
            methods=methods,
            dependencies=dependencies,
            permissions=cast(list["Permission"], permissions),
            exception_handlers=exception_handlers,
            middleware=middleware,
            include_in_schema=include_in_schema,
            deprecated=deprecated,
            before_request=before_request,
            after_request=after_request,
        )
        handler.fn = func
        self.add_route(path=path, handler=handler, name=name, interceptors=interceptors)
        return func

    return wrapper

websocket ¶

websocket(
    path,
    name=None,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    before_request=None,
    after_request=None,
)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format. TYPE: `str`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`dependencies`	A dictionary of string and Inject instances enable application level dependency injection. TYPE: `Optional[Dependencies]` DEFAULT: `None`
`interceptors`	A list of interceptors to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Interceptor]]` DEFAULT: `None`
`permissions`	A list of permissions to serve the application incoming requests (HTTP and Websockets). TYPE: `Optional[Sequence[Permission]]` DEFAULT: `None`
`exception_handlers`	A dictionary of exception types (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions. TYPE: `Optional[ExceptionHandlerMap]` DEFAULT: `None`
`middleware`	A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or Lilya Middleware as they are both converted internally. Read more about Python Protocols. TYPE: `Optional[list[Middleware]]` DEFAULT: `None`
`before_request`	A list of events that are triggered before the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`
`after_request`	A list of events that are triggered after the application processes the request. TYPE: `Sequence[Callable[..., Any]] \| None` DEFAULT: `None`

Source code in ravyn/routing/router.py

def websocket(
    self,
    path: Annotated[
        str,
        Doc(
            """
            Relative path of the `Gateway`.
            The path can contain parameters in a dictionary like format.
            """
        ),
    ],
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    dependencies: Annotated[
        Optional[Dependencies],
        Doc(
            """
            A dictionary of string and [Inject](https://ravyn.dev/dependencies/) instances enable application level dependency injection.
            """
        ),
    ] = None,
    interceptors: Annotated[
        Optional[Sequence[Interceptor]],
        Doc(
            """
            A list of [interceptors](https://ravyn.dev/interceptors/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    permissions: Annotated[
        Optional[Sequence[Permission]],
        Doc(
            """
            A list of [permissions](https://ravyn.dev/permissions/) to serve the application incoming requests (HTTP and Websockets).
            """
        ),
    ] = None,
    exception_handlers: Annotated[
        Optional[ExceptionHandlerMap],
        Doc(
            """
            A dictionary of [exception types](https://ravyn.dev/exceptions/) (or custom exceptions) and the handler functions on an application top level. Exception handler callables should be of the form of `handler(request, exc) -> response` and may be be either standard functions, or async functions.
            """
        ),
    ] = None,
    middleware: Annotated[
        Optional[list[Middleware]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of an Include will be checked from top-down or [Lilya Middleware](https://www.lilya.dev/middleware/) as they are both converted internally. Read more about [Python Protocols](https://peps.python.org/pep-0544/).
            """
        ),
    ] = None,
    before_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered before the application processes the request.
            """
        ),
    ] = None,
    after_request: Annotated[
        Sequence[Callable[..., Any]] | None,
        Doc(
            """
            A list of events that are triggered after the application processes the request.
            """
        ),
    ] = None,
) -> Callable:
    def wrapper(func: Callable) -> Callable:
        handler = WebSocketHandler(
            dependencies=dependencies,
            exception_handlers=exception_handlers,
            permissions=cast(list["Permission"], permissions),
            middleware=middleware,
            before_request=before_request,
            after_request=after_request,
        )
        handler.fn = func
        self.add_websocket_route(
            path=path, handler=handler, name=name, interceptors=interceptors
        )
        return func

    return wrapper

        - path
        - app
        - parent
        - on_shutdown
        - on_startup
        - lifespan
        - redirect_slashes
        - default
        - routes
        - dependencies
        - exception_handlers
        - interceptors
        - permissions
        - middleware
        - response_class
        - response_cookies
        - response_headers
        - deprecated
        - security
        - tags
        - add_controller
        - add_route
        - add_websocket_route

Router class¶

How to import¶

ravyn.Router ¶

on_startup instance-attribute ¶

on_shutdown instance-attribute ¶

dependencies instance-attribute ¶

dependency_overrides instance-attribute ¶

lifespan_context instance-attribute ¶

routes instance-attribute ¶

redirect_slashes instance-attribute ¶

default instance-attribute ¶

include_in_schema instance-attribute ¶

deprecated instance-attribute ¶

middleware instance-attribute ¶

permissions instance-attribute ¶

settings_module instance-attribute ¶

middleware_stack instance-attribute ¶

permission_started instance-attribute ¶

is_sub_router instance-attribute ¶

before_request instance-attribute ¶

after_request instance-attribute ¶

wrapped_permissions instance-attribute ¶

handler_signature property ¶

path_parameters property ¶

stringify_parameters property ¶

parent_levels property ¶

dependency_names property ¶

path instance-attribute ¶

parent instance-attribute ¶

exception_handlers instance-attribute ¶

interceptors instance-attribute ¶

lilya_permissions instance-attribute ¶

tags instance-attribute ¶

name instance-attribute ¶

response_class instance-attribute ¶

response_cookies instance-attribute ¶

response_headers instance-attribute ¶

security instance-attribute ¶

path_for ¶

url_path_for ¶

startup async ¶

shutdown async ¶

handle_route async ¶

handle_partial async ¶

handle_default async ¶

lifespan async ¶

app async ¶

handle_not_found async ¶

include ¶

host ¶

add_event_handler ¶

on_event ¶

get ¶

head ¶

post ¶

put ¶

patch ¶

delete ¶

trace ¶

options ¶

forward_single_method_route ¶

parse_path ¶

get_response_for_handler ¶

get_response_for_request async ¶

create_signature_model ¶

create_handler_transformer_model ¶

get_lookup_path ¶

get_dependencies ¶

is_unique_dependency staticmethod ¶

get_cookies ¶

get_headers ¶

get_response_data async ¶

allow_connection async ¶

dispatch_allow_connection async ¶

get_security_schemes ¶

get_handler_tags ¶

is_member_descriptor ¶

reorder_routes ¶

activate ¶

handle_interceptors async ¶

`Router` class¶

on_startup `instance-attribute` ¶

on_shutdown `instance-attribute` ¶

dependencies `instance-attribute` ¶

dependency_overrides `instance-attribute` ¶

lifespan_context `instance-attribute` ¶

routes `instance-attribute` ¶

redirect_slashes `instance-attribute` ¶

default `instance-attribute` ¶

include_in_schema `instance-attribute` ¶

deprecated `instance-attribute` ¶

middleware `instance-attribute` ¶

permissions `instance-attribute` ¶

settings_module `instance-attribute` ¶

middleware_stack `instance-attribute` ¶

permission_started `instance-attribute` ¶

is_sub_router `instance-attribute` ¶

before_request `instance-attribute` ¶

after_request `instance-attribute` ¶

wrapped_permissions `instance-attribute` ¶

handler_signature `property` ¶

path_parameters `property` ¶

stringify_parameters `property` ¶

parent_levels `property` ¶

dependency_names `property` ¶

path `instance-attribute` ¶

parent `instance-attribute` ¶

exception_handlers `instance-attribute` ¶

interceptors `instance-attribute` ¶

lilya_permissions `instance-attribute` ¶

tags `instance-attribute` ¶

name `instance-attribute` ¶

response_class `instance-attribute` ¶

response_cookies `instance-attribute` ¶

response_headers `instance-attribute` ¶

security `instance-attribute` ¶

startup `async` ¶

shutdown `async` ¶

handle_route `async` ¶

handle_partial `async` ¶

handle_default `async` ¶

lifespan `async` ¶

app `async` ¶

handle_not_found `async` ¶

get_response_for_request `async` ¶

is_unique_dependency `staticmethod` ¶

get_response_data `async` ¶

allow_connection `async` ¶

dispatch_allow_connection `async` ¶

handle_interceptors `async` ¶

handle_permissions `async` ¶

not_found `async` ¶