Include class

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

How to import

from ravyn import Include

ravyn.Include

Include(
    path=None,
    app=None,
    name=None,
    routes=None,
    namespace=None,
    pattern=None,
    parent=None,
    dependencies=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    middleware=None,
    before_request=None,
    after_request=None,
    include_in_schema=True,
    deprecated=None,
    security=None,
    tags=None,
    redirect_slashes=True,
)

Bases: Dispatcher, Include

Include object class that allows scalability and modularity to happen with elegance.

Read more about the Include to understand what can be done.

Include manages routes as a list or as a namespace but not both or a ImproperlyConfigured is raised.

PARAMETER	DESCRIPTION
path	Relative path of the Include. 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	An application can be anything that is treated as an ASGI application. For example, it can be a ChildRavyn, another Ravyn, a Router or even an external WSGI application (Django, Flask...) The app is a parameter that makes the Include extremely powerful when it comes to integrate with ease with whatever Python stack you want and need. Example from ravyn import Ravyn, ChildRavyn, Include Ravyn( routes=[ Include('/child', app=ChildRavyn(...)) ] ) Example with a WSGI framework from flask import Flask, escape, request from ravyn import Ravyn, Gateway, Include, Request, get from ravyn.middleware.wsgi import WSGIMiddleware flask_app = Flask(__name__) @flask_app.route("/") def flask_main(): name = request.args.get("name", "Ravyn") return f"Hello, {escape(name)} from your Flask integrated!" @get("/home/{name:str}") async def home(request: Request) -> dict: name = request.path_params["name"] return {"name": escape(name)} app = Ravyn( routes=[ Gateway(handler=home), Include("/flask", WSGIMiddleware(flask_app)), ] ) TYPE: ASGIApp | str DEFAULT: None
name	The name for the Gateway. The name can be reversed by url_path_for(). TYPE: Optional[str] DEFAULT: None
routes	A global list of ravyn routes. Those routes may vary and those can be Gateway, WebSocketGateWay or even another Include. This is also an entry-point for the routes of the Include 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
namespace	A string with a qualified namespace from where the URLs should be loaded. The namespace is an alternative to routes parameter. When a namespace is specified and a routes as well, an ImproperlyCOnfigured exception is raised as it can only be one or another. The namespace can be extremely useful as it avoids the imports from the top of the file that can lead to partially imported errors. When using a namespace, the Include will look for the default route_patterns list in the imported namespace (object) unless a different pattern is specified. Example Assuming there is a file with some routes located at myapp/auth/urls.py. myapp/auth/urls.py from ravyn import Gateway from .view import welcome, create_user route_patterns = [ Gateway(handler=welcome, name="welcome"), Gateway(handler=create_user, name="create-user"), ] Using the namespace to import the URLs. from ravyn import Include Include("/auth", namespace="myapp.auth.urls") TYPE: Optional[str] DEFAULT: None
pattern	A string pattern information from where the urls shall be read from. By default, the when using the namespace it will lookup for a route_patterns but somethimes you might want to opt for a different name and this is where the pattern comes along. Example Assuming there is a file with some routes located at myapp/auth/urls.py. The urls will be placed inside a urls list. myapp/auth/urls.py from ravyn import Gateway from .view import welcome, create_user urls = [ Gateway(handler=welcome, name="welcome"), Gateway(handler=create_user, name="create-user"), ] Using the namespace to import the URLs. from ravyn import Include Include("/auth", namespace="myapp.auth.urls", pattern="urls") TYPE: Optional[str] 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
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] | Any] 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 trigger after the application processes the request. Read more about the events. 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. TYPE: Union[Sequence[Callable[..., Any]], None] DEFAULT: None
include_in_schema	Boolean flag indicating if it should be added to the OpenAPI docs. This will add all the routes of the Include even those nested (Include containing more Includes.) TYPE: bool DEFAULT: True
deprecated	Boolean flag for indicating the deprecation of the Include and all of its routes and to display it in the OpenAPI documentation.. TYPE: Optional[bool] 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
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
redirect_slashes	Boolean flag indicating if the redirect slashes are enabled for the routes or not. TYPE: Optional[bool] DEFAULT: True

Source code in ravyn/routing/router.py

def __init__(
    self,
    path: Annotated[
        Optional[str],
        Doc(
            """
            Relative path of the `Include`.
            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[
        ASGIApp | str,
        Doc(
            """
            An application can be anything that is treated as an ASGI application.
            For example, it can be a [ChildRavyn](https://ravyn.dev/routing/router/#child-ravyn-application), another `Ravyn`, a [Router](https://ravyn.dev/routing/router/#router-class) or even an external [WSGI application](https://ravyn.dev/wsgi/) (Django, Flask...)

            The app is a parameter that makes the Include extremely powerful when it comes
            to integrate with ease with whatever Python stack you want and need.

            **Example**

            ```python
            from ravyn import Ravyn, ChildRavyn, Include

            Ravyn(
                routes=[
                    Include('/child', app=ChildRavyn(...))
                ]
            )
            ```

            **Example with a WSGI framework**

            ```python
            from flask import Flask, escape, request

            from ravyn import Ravyn, Gateway, Include, Request, get
            from ravyn.middleware.wsgi import WSGIMiddleware

            flask_app = Flask(__name__)


            @flask_app.route("/")
            def flask_main():
                name = request.args.get("name", "Ravyn")
                return f"Hello, {escape(name)} from your Flask integrated!"


            @get("/home/{name:str}")
            async def home(request: Request) -> dict:
                name = request.path_params["name"]
                return {"name": escape(name)}


            app = Ravyn(
                routes=[
                    Gateway(handler=home),
                    Include("/flask", WSGIMiddleware(flask_app)),
                ]
            )
            ```

            """
        ),
    ] = None,
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    routes: Annotated[
        Optional[Sequence[Union[APIGateHandler, Include, HTTPHandler, WebSocketHandler]]],
        Doc(
            """
            A global `list` of ravyn routes. Those routes may vary and those can
            be `Gateway`, `WebSocketGateWay` or even another `Include`.

            This is also an entry-point for the routes of the Include
            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,
    namespace: Annotated[
        Optional[str],
        Doc(
            """
            A string with a qualified namespace from where the URLs should be loaded.

            The namespace is an alternative to `routes` parameter. When a `namespace` is
            specified and a routes as well, an `ImproperlyCOnfigured` exception is raised as
            it can only be one or another.

            The `namespace` can be extremely useful as it avoids the imports from the top
            of the file that can lead to `partially imported` errors.

            When using a `namespace`, the `Include` will look for the default `route_patterns` list in the imported namespace (object) unless a different `pattern` is specified.

            **Example**

            Assuming there is a file with some routes located at `myapp/auth/urls.py`.

            ```python title="myapp/auth/urls.py"
            from ravyn import Gateway
            from .view import welcome, create_user

            route_patterns = [
                Gateway(handler=welcome, name="welcome"),
                Gateway(handler=create_user, name="create-user"),
            ]
            ```

            Using the `namespace` to import the URLs.

            ```python
            from ravyn import Include

            Include("/auth", namespace="myapp.auth.urls")
            ```
            """
        ),
    ] = None,
    pattern: Annotated[
        Optional[str],
        Doc(
            """
            A string `pattern` information from where the urls shall be read from.

            By default, the when using the `namespace` it will lookup for a `route_patterns`
            but somethimes you might want to opt for a different name and this is where the
            `pattern` comes along.

            **Example**

            Assuming there is a file with some routes located at `myapp/auth/urls.py`.
            The urls will be placed inside a `urls` list.

            ```python title="myapp/auth/urls.py"
            from ravyn import Gateway
            from .view import welcome, create_user

            urls = [
                Gateway(handler=welcome, name="welcome"),
                Gateway(handler=create_user, name="create-user"),
            ]
            ```

            Using the `namespace` to import the URLs.

            ```python
            from ravyn import Include

            Include("/auth", namespace="myapp.auth.urls", pattern="urls")
            ```
            """
        ),
    ] = 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,
    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] | Any],
        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[
        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/).
            """
        ),
    ] = 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/).
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.

            This will add all the routes of the Include even those nested (Include containing more Includes.)
            """
        ),
    ] = True,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Include and all of its routes and to display it in the OpenAPI documentation..
            """
        ),
    ] = 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,
    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,
    redirect_slashes: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag indicating if the redirect slashes are enabled for the
            routes or not.
            """
        ),
    ] = True,
) -> None:
    self.path = path
    if not path:
        self.path = "/"

    if namespace and routes:
        raise ImproperlyConfigured("It can only be namespace or routes, not both.")

    if namespace and not isinstance(namespace, str):
        raise ImproperlyConfigured("Namespace must be a string. Example: 'myapp.routes'.")

    if pattern and not isinstance(pattern, str):
        raise ImproperlyConfigured("Pattern must be a string. Example: 'route_patterns'.")

    if pattern and routes:
        raise ImproperlyConfigured("Pattern must be used only with namespace.")

    if namespace:
        routes = include(namespace, pattern)

    # Add the middleware to the include
    self.middleware = middleware or []
    include_middleware: Sequence[Middleware] = []

    for _middleware in self.middleware:
        if isinstance(_middleware, DefineMiddleware):  # pragma: no cover
            include_middleware.append(_middleware)
        else:
            include_middleware.append(  # type: ignore
                DefineMiddleware(cast("Type[DefineMiddleware]", _middleware))
            )

    if isinstance(app, str):
        app = load(app)

    self.app = self.resolve_app_parent(app=app)

    self.dependencies = dependencies or {}  # type: ignore
    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
    self.response_class = None
    self.response_cookies = None
    self.response_headers = None
    self.parent = parent
    self.security = security or []
    self.tags = tags or []

    if namespace:
        routes = include(namespace, pattern)

    if routes:
        routes = self.resolve_route_path_handler(routes)

    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__(
        path=self.path,
        app=self.app,
        routes=routes,
        name=name,
        middleware=include_middleware,
        exception_handlers=exception_handlers,
        deprecated=deprecated,
        include_in_schema=include_in_schema,
        redirect_slashes=redirect_slashes,
        permissions=self.__lilya_permissions__,  # type: ignore
        before_request=before_request,
        after_request=after_request,
    )

    # 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 = {}

    # Making sure Ravyn uses the Ravyn permission system and not Lilya's.
    # 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."
    )

handle_not_found instance-attribute

handle_not_found = handle_not_found_fallthrough

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.

router_class class-attribute

router_class = Router

exception_handlers instance-attribute

exception_handlers = (
    {}
    if exception_handlers is None
    else dict(exception_handlers)
)

wrapped_permissions instance-attribute

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

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 []
)

name instance-attribute

name = name

include_in_schema instance-attribute

include_in_schema = include_in_schema

deprecated instance-attribute

deprecated = deprecated

routes property

routes

Returns a list of declared path objects.

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.

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

middleware instance-attribute

middleware = middleware or []

app instance-attribute

app = resolve_app_parent(app=app)

dependencies instance-attribute

dependencies = dependencies or {}

interceptors instance-attribute

interceptors = interceptors or []

response_class instance-attribute

response_class = None

response_cookies instance-attribute

response_cookies = None

response_headers instance-attribute

response_headers = None

parent instance-attribute

parent = parent

security instance-attribute

security = security or []

tags instance-attribute

tags = tags or []

lilya_permissions instance-attribute

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

permissions instance-attribute

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

handle_signature

handle_signature()

Source code in lilya/routing.py

def handle_signature(self) -> None:
    raise NotImplementedError()  # pragma: no cover

search

search(scope)

Searches within the route patterns and matches against the regex.

If found, then dispatches the request to the handler of the object.

PARAMETER	DESCRIPTION
scope	The request scope. TYPE: Scope

RETURNS	DESCRIPTION
tuple[Match, Scope]	Tuple[Match, Scope]: The match result and child scope.

Source code in lilya/routing.py

def search(self, scope: Scope) -> tuple[Match, Scope]:
    """
    Searches within the route patterns and matches against the regex.

    If found, then dispatches the request to the handler of the object.

    Args:
        scope (Scope): The request scope.

    Returns:
        Tuple[Match, Scope]: The match result and child scope.
    """
    if scope["type"] in {ScopeType.HTTP, ScopeType.WEBSOCKET}:
        root_path = scope.get("root_path", "")
        route_path = get_route_path(scope)
        match = self.path_regex.match(route_path)

        if match:
            return self.handle_match(scope, match, route_path, root_path)

    return Match.NONE, {}

path_for

path_for(name, /, **path_params)

Generate a URLPath for a given route name and path parameters.

PARAMETER	DESCRIPTION
name	The name of the route. TYPE: str
path_params	Path parameters for route substitution. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
URLPath	The generated URLPath. TYPE: URLPath

RAISES	DESCRIPTION
NoMatchFound	If no matching route is found for the given name and parameters.

Source code in lilya/routing.py

def path_for(self, name: str, /, **path_params: Any) -> URLPath:
    """
    Generate a URLPath for a given route name and path parameters.

    Args:
        name (str): The name of the route.
        path_params: Path parameters for route substitution.

    Returns:
        URLPath: The generated URLPath.

    Raises:
        NoMatchFound: If no matching route is found for the given name and parameters.
    """

    return self.url_path_for(name, **path_params)

url_path_for

url_path_for(name, /, **path_params)

Generate a URLPath for a given route name and path parameters.

PARAMETER	DESCRIPTION
name	The name of the route. TYPE: str
path_params	Path parameters for route substitution. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
URLPath	The generated URLPath. TYPE: URLPath

RAISES	DESCRIPTION
NoMatchFound	If no matching route is found for the given name and parameters.

Source code in lilya/routing.py

def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
    """
    Generate a URLPath for a given route name and path parameters.

    Args:
        name (str): The name of the route.
        path_params: Path parameters for route substitution.

    Returns:
        URLPath: The generated URLPath.

    Raises:
        NoMatchFound: If no matching route is found for the given name and parameters.
    """
    if self.name is not None and name == self.name and "path" in path_params:
        path_params["path"] = path_params["path"].lstrip("/")
        path, remaining_params = replace_params(
            self.path_format, self.param_convertors, path_params
        )
        if not remaining_params:
            return URLPath(path=path)
    elif self.name is None or name.startswith(self.name + ":"):
        return self._path_for_without_name(name, path_params)
    raise NoMatchFound(name, path_params)

dispatch async

dispatch(scope, receive, send)

Dispatches the request to the appropriate handler.

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 dispatch(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Dispatches the request to the appropriate handler.

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

    Returns:
        None
    """
    match, child_scope = self.search(scope)

    if match == Match.NONE:
        await self.handle_not_found(scope, receive, send)
        return

    scope.update(child_scope)
    await self.handle_dispatch(scope, receive, send)  # type: ignore

handle_not_found_fallthrough async staticmethod

handle_not_found_fallthrough(scope, receive, send)

Source code in lilya/routing.py

@staticmethod
async def handle_not_found_fallthrough(scope: Scope, receive: Receive, send: Send) -> None:
    raise ContinueRouting()

handle_exception_handlers async

handle_exception_handlers(scope, receive, send, exc)

Manages exception handlers for HTTP and WebSocket scopes.

PARAMETER	DESCRIPTION
scope	The ASGI scope. TYPE: dict
receive	The receive function. TYPE: callable
send	The send function. TYPE: callable
exc	The exception to handle. TYPE: Exception

Source code in lilya/routing.py

async def handle_exception_handlers(
    self, scope: Scope, receive: Receive, send: Send, exc: Exception
) -> None:
    """
    Manages exception handlers for HTTP and WebSocket scopes.

    Args:
        scope (dict): The ASGI scope.
        receive (callable): The receive function.
        send (callable): The send function.
        exc (Exception): The exception to handle.
    """
    status_code = self._get_status_code(exc)

    if scope["type"] == ScopeType.HTTP:
        await self._handle_http_exception(scope, receive, send, exc, status_code)
    elif scope["type"] == ScopeType.WEBSOCKET:
        await self._handle_websocket_exception(send, exc, status_code)

handle_match

handle_match(scope, match, route_path, root_path)

Handles the case when a match is found in the route patterns.

PARAMETER	DESCRIPTION
scope	The request scope. TYPE: Scope
match	The match object from the regex. TYPE: Match

RETURNS	DESCRIPTION
tuple[Match, Scope]	Tuple[Match, Scope]: The match result and child scope.

Source code in lilya/routing.py

def handle_match(
    self, scope: Scope, match: re.Match, route_path: str, root_path: str
) -> tuple[Match, Scope]:
    """
    Handles the case when a match is found in the route patterns.

    Args:
        scope (Scope): The request scope.
        match: The match object from the regex.

    Returns:
        Tuple[Match, Scope]: The match result and child scope.
    """
    matched_params = {
        key: self.param_convertors[key].transform(value)
        for key, value in match.groupdict().items()
    }
    remaining_path = f"/{matched_params.pop('path', '')}"
    matched_path = route_path[: -len(remaining_path)]

    path_params = {**scope.get("path_params", {}), **matched_params}
    existing = list(scope.get("dependencies", []))
    child_scope = {
        "path_params": path_params,
        "app_root_path": scope.get("app_root_path", root_path),
        "root_path": root_path + matched_path,
        "handler": self.app,
        "dependencies": existing + [self.dependencies],
    }
    return Match.FULL, child_scope

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

resolve_app_parent

resolve_app_parent(app)

Resolves the owner of ChildRavyn or Ravyn iself.

Source code in ravyn/routing/router.py

def resolve_app_parent(self, app: Optional[Any]) -> Optional[Any]:
    """
    Resolves the owner of ChildRavyn or Ravyn iself.
    """
    from ravyn import ChildRavyn, Ravyn

    if app is not None and isinstance(app, (Ravyn, ChildRavyn)):
        app.parent = self
    return app

resolve_route_path_handler

resolve_route_path_handler(routes)

Make sure the paths are properly configured from the handler handler. The handler can be a Lilya function, an View or a HTTPHandler.

LilyaBasePath() has a limitation of not allowing nested LilyaBasePath().

Example:

route_patterns = [
    LilyaBasePath(path='/my path', routes=[
        LilyaBasePath(path='/another mount')
    ])
]

Include() extends the LilyaBasePath and adds extras on the top. Allowing nested Include() also allows importing in different ways. Via qualified namespace or via list() but not both.

Example:

1. Root of the application, example, `urls.py`.

route_patterns = [
    Include(path='/api/v1/', namespace='myapp.v1.urls'),
    Gateway(path='/example', handler=example_endpoint, name='example')
    ...
]

2. Inside `myapp.v1.urls`

from mysecondapp.urls import route_patterns

route_patterns = [
    Include(path='/api/v2/', routes=route_patterns),
    Gateway(path='/another-example', handler=another_endpoint, name='example')
    ...
]

3. Inside `mysecondapp.v1.urls`:

route_patterns = [
    Gateway(path='/last-example', handler=another_endpoint, name='example')
    ...
]

Source code in ravyn/routing/router.py

def resolve_route_path_handler(
    self,
    routes: Sequence[Union[APIGateHandler, Include, HTTPHandler, WebSocketHandler]],
) -> list[Union[Gateway, WebSocketGateway, Include]]:
    """
    Make sure the paths are properly configured from the handler handler.
    The handler can be a Lilya function, an View or a HTTPHandler.

    LilyaBasePath() has a limitation of not allowing nested LilyaBasePath().

    Example:

        route_patterns = [
            LilyaBasePath(path='/my path', routes=[
                LilyaBasePath(path='/another mount')
            ])
        ]


    Include() extends the LilyaBasePath and adds extras on the top. Allowing nested
    Include() also allows importing in different ways. Via qualified namespace
    or via list() but not both.

    Example:

        1. Root of the application, example, `urls.py`.

        route_patterns = [
            Include(path='/api/v1/', namespace='myapp.v1.urls'),
            Gateway(path='/example', handler=example_endpoint, name='example')
            ...
        ]

        2. Inside `myapp.v1.urls`

        from mysecondapp.urls import route_patterns

        route_patterns = [
            Include(path='/api/v2/', routes=route_patterns),
            Gateway(path='/another-example', handler=another_endpoint, name='example')
            ...
        ]

        3. Inside `mysecondapp.v1.urls`:

        route_patterns = [
            Gateway(path='/last-example', handler=another_endpoint, name='example')
            ...
        ]

    """
    routing: list[Union[Gateway, WebSocketGateway, Include]] = []

    for route in routes:  # pragma: no cover
        if isinstance(route, WebhookHandler):
            # fail later, a WebhookHandler is subclass of HTTPHandler, so pass down
            ...
        elif isinstance(route, HTTPHandler):
            route = Gateway(handler=route)
        elif isinstance(route, WebSocketHandler):
            route = WebSocketGateway(handler=route)

        if not isinstance(route, (Include, Gateway, WebSocketGateway)):
            raise ImproperlyConfigured("The route must be of type Gateway or Include")

        route.parent = self
        if isinstance(route, Include):
            routing.append(route)
            continue

        if isinstance(route.handler, (HTTPHandler, WebSocketHandler)):
            route.handler.parent = route
            routing.append(route)
            continue

        if is_class_and_subclass(route.handler, BaseController) or isinstance(
            route.handler, BaseController
        ):
            if not route.handler.parent:
                route.handler = route.handler(parent=self)  # type: ignore

            route_handlers: list[Union[Gateway, WebhookGateway, Include]] = (
                route.handler.get_routes(  # type: ignore
                    path=route.path,
                    middleware=route.middleware,
                    interceptors=self.interceptors,
                    permissions=route.permissions,
                    exception_handlers=route.exception_handlers,
                    before_request=route.before_request,
                    after_request=route.after_request,
                )
            )
            if route_handlers:
                routing.extend(
                    cast(
                        list[Union[Gateway, WebSocketGateway, Include]],
                        route_handlers,
                    )
                )
    return routing

handle_interceptors async

handle_interceptors(scope, receive, send)

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

Source code in ravyn/routing/router.py

async def handle_interceptors(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
    """
    Handles the interceptors for the Include.
    This method ensures that the interceptors are set correctly
    and that they are compatible with the Lilya interceptors system.
    """
    # 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:
        effective_permissions = self.permissions.copy() if self.permissions else {}
        effective_lilya_permissions = (
            self.lilya_permissions.copy() if self.lilya_permissions else {}
        )
        # 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:
        dispatch_call = super().handle_dispatch
        await self.dispatch_allow_connection(
            effective_permissions,
            connection,
            scope,
            receive,
            send,
            dispatch_call=dispatch_call,
        )

handle_dispatch async

handle_dispatch(scope, receive, send)

In the call we need to make sure we call the permissions and validations first to assure it won't even reach the lower app and run the validation against.

Source code in ravyn/routing/router.py

async def handle_dispatch(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
    """
    In the call we need to make sure we call the permissions and validations first to
    assure it won't even reach the lower app and run the validation against.
    """
    await self.handle_interceptors(scope, receive, send)
    await self.handle_permissions(scope, receive, send)
    await super().handle_dispatch(scope, receive, send)

        - path
        - app
        - name
        - routes
        - namespace
        - pattern
        - parent
        - dependencies
        - exception_handlers
        - interceptors
        - permissions
        - middleware
        - include_in_schema
        - deprecated
        - security
        - tags