WebSocketGateway class

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

How to import

from ravyn import WebSocketGateway

ravyn.WebSocketGateway

WebSocketGateway(
    path=None,
    *,
    handler,
    name=None,
    parent=None,
    dependencies=None,
    middleware=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    before_request=None,
    after_request=None,
    is_from_router=False,
)

Bases: WebSocketPath, Dispatcher, _GatewayCommon

WebSocketGateway object class used by Ravyn routes.

The WebSocketGateway act as a brigde between the router handlers and the main Ravyn routing system.

Read more about WebSocketGateway and how to use it.

Example

from ravyn import Ravyn. websocket

@websocket()
async def world_socket(socket: Websocket) -> None:
    await socket.accept()
    msg = await socket.receive_json()
    assert msg
    assert socket
    await socket.close()

WebSocketGateway(path="/ws", handler=home)

PARAMETER	DESCRIPTION
path	Relative path of the WebSocketGateway. The path can contain parameters in a dictionary like format and if the path is not provided, it will default to /. Example WebSocketGateway() Example with parameters WebSocketGateway(path="/{age: int}") TYPE: Optional[str] DEFAULT: None
handler	An instance of handler. TYPE: Union[WebSocketHandler, BaseController, Type[BaseController], Type[WebSocketHandler]]
name	The name for the Gateway. The name can be reversed by url_path_for(). 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
middleware	A list of middleware to run for every request. The middlewares of a Gateway 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
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
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
is_from_router	Used by the .add_router() function of the Ravyn class indicating if the Gateway is coming from a router. TYPE: bool DEFAULT: False

Source code in ravyn/routing/gateways.py

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

            **Example**

            ```python
            WebSocketGateway()
            ```

            **Example with parameters**

            ```python
            WebSocketGateway(path="/{age: int}")
            ```
            """
        ),
    ] = None,
    *,
    handler: Annotated[
        Union[
            "WebSocketHandler", BaseController, Type[BaseController], Type["WebSocketHandler"]
        ],
        Doc(
            """
        An instance of [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,
    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,
    middleware: Annotated[
        Optional[list["Middleware"]],
        Doc(
            """
            A list of middleware to run for every request. The middlewares of a Gateway 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,
    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,
    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,
    is_from_router: Annotated[
        bool,
        Doc(
            """
            Used by the `.add_router()` function of the `Ravyn` class indicating if the
            Gateway is coming from a router.
            """
        ),
    ] = False,
) -> None:
    raw_handler = handler
    handler = self._instantiate_if_controller(raw_handler, self)  # type: ignore

    self.path = self._resolve_path(
        path, getattr(handler, "path", "/"), is_from_router=is_from_router
    )
    resolved_name = self._resolve_name(name, handler)

    prepared_middleware = self._prepare_middleware(handler, middleware)

    lilya_permissions, _ = self._prepare_permissions(handler, permissions)

    super().__init__(
        path=self.path,
        handler=cast(Callable, handler),
        name=resolved_name,
        middleware=prepared_middleware,
        exception_handlers=exception_handlers,
        permissions=lilya_permissions or {},  # type: ignore
        before_request=before_request,
        after_request=after_request,
    )

    self._apply_events(handler, before_request, after_request)
    self._apply_interceptors(handler, interceptors)

    self.before_request = list(before_request or [])
    self.after_request = list(after_request or [])
    self.handler = cast("Callable", handler)
    self.dependencies = dependencies or {}  # type: ignore
    self.interceptors = list(interceptors or [])
    self.parent = parent
    self.name = resolved_name
    self.lilya_permissions = lilya_permissions or {}
    self.permissions = getattr(handler, "permissions", {}) or {}  # type: ignore
    self.include_in_schema = False  # websockets are excluded by default

    self._compile(handler, self.path)

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 lilya show-urls.

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.

signature property

signature

app instance-attribute

app = handle_websocket_session(handler)

middleware instance-attribute

middleware = [(wrap_middleware(mid)) for mid in middleware]

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

path instance-attribute

path = _resolve_path(
    path,
    getattr(handler, "path", "/"),
    is_from_router=is_from_router,
)

before_request instance-attribute

before_request = list(before_request or [])

after_request instance-attribute

after_request = list(after_request or [])

handler instance-attribute

handler = cast('Callable', handler)

dependencies instance-attribute

dependencies = dependencies or {}

interceptors instance-attribute

interceptors = list(interceptors or [])

parent instance-attribute

parent = parent

name instance-attribute

name = resolved_name

lilya_permissions instance-attribute

lilya_permissions = lilya_permissions or {}

permissions instance-attribute

permissions = getattr(handler, 'permissions', {}) or {}

include_in_schema instance-attribute

include_in_schema = False

is_class_based staticmethod

is_class_based(handler)

Checks if the handler object or class is a subclass or instance of BaseController.

Source code in ravyn/routing/gateways.py

@staticmethod
def is_class_based(
    handler: "HTTPHandler | WebSocketHandler | ParentType | BaseController",
) -> bool:
    """Checks if the handler object or class is a subclass or instance of BaseController."""
    return bool(
        is_class_and_subclass(handler, BaseController) or isinstance(handler, BaseController)
    )

is_handler staticmethod

is_handler(handler)

Checks if the callable is a standalone function/method and NOT a BaseController instance/subclass.

Source code in ravyn/routing/gateways.py

@staticmethod
def is_handler(handler: Callable[..., Any]) -> bool:
    """Checks if the callable is a standalone function/method and NOT a BaseController instance/subclass."""
    return bool(
        not is_class_and_subclass(handler, BaseController)
        and not isinstance(handler, BaseController)
    )

generate_operation_id

generate_operation_id(name, handler)

Generates a unique, normalized operation ID suitable for OpenAPI specification.

The ID is constructed from the handler's base name/class name and the route path, often appended with the primary HTTP method.

PARAMETER	DESCRIPTION
name	The explicit name given to the route (if any). TYPE: str | None
handler	The handler object (HTTPHandler, WebSocketHandler, or BaseController instance). TYPE: HTTPHandler | WebSocketHandler | BaseController

RETURNS	DESCRIPTION
str	A cleaned string suitable for use as an OpenAPI operationId.

Source code in ravyn/routing/gateways.py

def generate_operation_id(
    self,
    name: str | None,
    handler: "HTTPHandler | WebSocketHandler | BaseController",
) -> str:
    """
    Generates a unique, normalized operation ID suitable for OpenAPI specification.

    The ID is constructed from the handler's base name/class name and the route path,
    often appended with the primary HTTP method.

    Args:
        name: The explicit name given to the route (if any).
        handler: The handler object (HTTPHandler, WebSocketHandler, or BaseController instance).

    Returns:
        A cleaned string suitable for use as an OpenAPI operationId.
    """
    if self.is_class_based(getattr(handler, "parent", None) or handler):
        base: str = handler.parent.__class__.__name__.lower()
    else:
        base = name or getattr(handler, "name", "") or ""

    path_fmt: str = getattr(handler, "path_format", "") or ""

    # Remove non-word characters and combine base and path format
    operation_id: str = re.sub(r"\W", "_", f"{base}{path_fmt}")

    # Append the primary method (if available)
    methods: list[str] = list(
        getattr(handler, "methods", []) or getattr(handler, "http_methods", []) or []
    )
    if methods:
        operation_id = f"{operation_id}_{methods[0].lower()}"

    return operation_id

handle_middleware staticmethod

handle_middleware(handler, base_middleware)

Normalizes a list of middleware classes/instances into a list of DefineMiddleware instances.

Merges handler-defined middleware with Gateway-level middleware.

PARAMETER	DESCRIPTION
handler	The route handler object. TYPE: Any
base_middleware	The list of middleware defined at the Gateway level. TYPE: list[Middleware]

RETURNS	DESCRIPTION
list[Middleware]	A list of Middleware objects, all wrapped in DefineMiddleware if necessary.

Source code in ravyn/routing/gateways.py

@staticmethod
def handle_middleware(handler: Any, base_middleware: list["Middleware"]) -> list["Middleware"]:
    """
    Normalizes a list of middleware classes/instances into a list of `DefineMiddleware` instances.

    Merges `handler`-defined middleware with `Gateway`-level middleware.

    Args:
        handler: The route handler object.
        base_middleware: The list of middleware defined at the Gateway level.

    Returns:
        A list of `Middleware` objects, all wrapped in `DefineMiddleware` if necessary.
    """
    _middleware: list["Middleware"] = []

    # Merge handler middleware if handler is not a Controller
    if not is_class_and_subclass(handler, BaseController) and not isinstance(
        handler, BaseController
    ):
        base_middleware += handler.middleware or []

    for middleware in base_middleware or []:
        if isinstance(middleware, DefineMiddleware):
            _middleware.append(middleware)
        else:
            _middleware.append(DefineMiddleware(middleware))  # type: ignore
    return _middleware

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

handle_signature

handle_signature()

Validates the return annotation of a handler if enforce_return_annotation is set to True.

Source code in lilya/routing.py

def handle_signature(self) -> None:
    """
    Validates the return annotation of a handler
    if `enforce_return_annotation` is set to True.
    """
    if not _monkay.settings.enforce_return_annotation:
        return None

    if self.signature.return_annotation is inspect._empty:
        raise ImproperlyConfigured(
            "A return value of a route handler function should be type annotated. "
            "If your function doesn't return a value or returns None, annotate it as returning 'NoReturn' or 'None' respectively."
        )

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"] == ScopeType.WEBSOCKET:
        route_path = get_route_path(scope)
        match = self.path_regex.match(route_path)

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

    return Match.NONE, {}

path_for

path_for(name, **path_params)

Generates a URL path for the specified route name and parameters.

PARAMETER	DESCRIPTION
name	The name of the route. TYPE: str
path_params	The path parameters. TYPE: dict DEFAULT: {}

RETURNS	DESCRIPTION
URLPath	The generated URL path. TYPE: URLPath

RAISES	DESCRIPTION
NoMatchFound	If there is no match for the given name and parameters.

Source code in lilya/routing.py

def path_for(self, name: str, **path_params: Any) -> URLPath:
    """
    Generates a URL path for the specified route name and parameters.

    Args:
        name (str): The name of the route.
        path_params (dict): The path parameters.

    Returns:
        URLPath: The generated URL path.

    Raises:
        NoMatchFound: If there is no match for the given name and parameters.

    """
    return self.url_path_for(name, **path_params)

url_path_for

url_path_for(name, /, **path_params)

Generates a URL path for the specified route name and parameters.

PARAMETER	DESCRIPTION
name	The name of the route. TYPE: str
path_params	The path parameters. TYPE: dict DEFAULT: {}

RETURNS	DESCRIPTION
URLPath	The generated URL path. TYPE: URLPath

RAISES	DESCRIPTION
NoMatchFound	If there is no match for the given name and parameters.

Source code in lilya/routing.py

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

    Args:
        name (str): The name of the route.
        path_params (dict): The path parameters.

    Returns:
        URLPath: The generated URL path.

    Raises:
        NoMatchFound: If there is no match for the given name and parameters.

    """
    self.validate_params(name, path_params)

    path, remaining_params = replace_params(
        self.path_format, self.param_convertors, path_params
    )
    assert not remaining_params
    return URLPath(path=path, protocol=ScopeType.WEBSOCKET.value)

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

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)

extract_request_information

extract_request_information(request, signature)

Extracts the information and flattens the request dictionaries in the handler.

Source code in lilya/_internal/_responses.py

def extract_request_information(self, request: Request, signature: inspect.Signature) -> None:
    """
    Extracts the information and flattens the request dictionaries in the handler.
    """
    self.__query_params__ = dict(request.query_params.items())
    self.__path_params__ = dict(request.path_params.items())
    self.__header_params__ = dict(request.headers.items())
    self.__cookie_params__ = dict(request.cookies.items())

    reserved_keys = set(self.__path_params__.keys())
    reserved_keys.update(self.__query_params__.keys())
    reserved_keys.update(self.__header_params__.keys())
    reserved_keys.update(self.__cookie_params__.keys())

    self.__reserved_data__ = {
        "path_params": self.__path_params__,
        "header_params": self.__header_params__,
        "cookie_params": self.__cookie_params__,
        "query_params": self.__query_params__,
    }

    # Store the body params in the handler variable
    self.__body_params__ = {
        k: v.annotation for k, v in signature.parameters.items() if k not in reserved_keys
    }

handle_response

handle_response(func, other_signature=None)

Decorator for creating a request-response ASGI application.

PARAMETER	DESCRIPTION
func	The function to be wrapped. TYPE: Callable
other_signature	Another passed signature TYPE: Signature DEFAULT: None

RETURNS	DESCRIPTION
ASGIApp	The ASGI application. TYPE: ASGIApp

Source code in lilya/_internal/_responses.py

def handle_response(
    self,
    func: Callable[[Request], Awaitable[Response] | Response]
    | Callable[[], Coroutine[Any, Any, Response]],
    other_signature: inspect.Signature | None = None,
) -> ASGIApp:
    """
    Decorator for creating a request-response ASGI application.

    Args:
        func (Callable): The function to be wrapped.
        other_signature (inspect.Signature): Another passed signature

    Returns:
        ASGIApp: The ASGI application.
    """

    # Precompute a static signature and a lightweight plan for this handler.
    # For function views, the signature is stable. For controllers the other_signature
    # will be provided by the caller. We default to `inspect.signature(func)` only once.
    static_signature: inspect.Signature | None = other_signature or inspect.signature(func)
    sig_params = tuple(static_signature.parameters.items()) if static_signature else ()
    has_provider_markers = any(
        isinstance(p.default, (Provides, Resolve, Security)) for _, p in sig_params
    )
    has_body_candidates = any(
        (p.default is inspect._empty)
        and (n not in (SignatureDefault.REQUEST, "context"))
        and not isinstance(p.default, (Query, Header, Cookie))
        for n, p in sig_params
    )
    pre_bound_fields = [
        (n, p.default) for n, p in sig_params if isinstance(p.default, (Query, Header, Cookie))
    ]

    async def app(scope: Scope, receive: Receive, send: Send) -> None:
        """
        ASGI application handling request-response.

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

        Returns:
            None
        """
        request = Request(scope=scope, receive=receive, send=send)

        async def inner_app(scope: Scope, receive: Receive, send: Send) -> None:
            """
            Inner ASGI application handling request-response.

            Sometimes the handler does not need the request to be passed
            in the handler and we can avoid it by ignoring the request
            object in the arguments.

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

            Returns:
                None
            """
            signature: inspect.Signature = (
                other_signature or self.signature or static_signature
            )
            self.extract_request_information(request=request, signature=signature)

            # Ultra-fast short-circuits already exist below (no params, request-only, context-only).
            # Here we add a common-case fast path for handlers that:
            #  - have no Provides/Resolve/Security markers
            #  - have no body-bound candidates
            # In that case we skip `_extract_params_from_request` entirely and only pull
            # path/query/header/cookie/context.
            if not has_provider_markers:
                # Re-evaluate body candidates only if the signature at call time differs.
                _has_body_candidates = has_body_candidates
                if signature is not static_signature:
                    sp = tuple(signature.parameters.items())
                    _has_body_candidates = any(
                        (p.default is inspect._empty)
                        and (n not in (SignatureDefault.REQUEST, "context"))
                        and not isinstance(p.default, (Query, Header, Cookie))
                        for n, p in sp
                    )
                if not _has_body_candidates:
                    # Build params without dependency/body resolution.
                    # 1) Path params that match the function signature
                    fast_params = {
                        name: val
                        for name, val in request.path_params.items()
                        if name in signature.parameters
                    }
                    # 2) Query/Header/Cookie bound fields
                    #    Use the pre-bound list if signature matches; otherwise compute once.
                    bound_fields = (
                        pre_bound_fields
                        if signature is static_signature
                        else [
                            (n, p.default)
                            for n, p in signature.parameters.items()
                            if isinstance(p.default, (Query, Header, Cookie))
                        ]
                    )
                    for n, field in bound_fields:
                        if isinstance(field, Query):
                            source = request.query_params
                            key = field.alias or n
                        elif isinstance(field, Header):
                            source = request.headers  # type: ignore
                            key = field.value
                        else:  # Cookie
                            source = request.cookies  # type: ignore
                            key = field.value
                        try:
                            if not isinstance(field, Cookie):
                                values = source.getall(key, None)
                                if values is None:
                                    raw_value = None
                                elif isinstance(values, list):
                                    raw_value = values[0] if len(values) == 1 else values
                                else:
                                    raw_value = values  # type: ignore
                            else:
                                raw_value = source.get(key)
                        except (KeyError, TypeError):
                            raw_value = None

                        if field.required and raw_value is None:
                            raise UnprocessableEntity(
                                f"Missing mandatory query parameter '{key}'"
                            )
                        if raw_value is None:
                            fast_params[n] = getattr(field, "default", None)
                        else:
                            try:
                                if field.cast and isinstance(raw_value, list):
                                    fast_params[n] = [raw_value]
                                elif field.cast:
                                    fast_params[n] = field.resolve(raw_value, field.cast)
                                else:
                                    fast_params[n] = (
                                        raw_value[0]
                                        if isinstance(raw_value, list) and len(raw_value) == 1
                                        else raw_value
                                    )
                            except (TypeError, ValueError):
                                raise UnprocessableEntity(
                                    f"Invalid value for query parameter '{key}': expected {field.cast.__name__}"
                                ) from None

                    # 3) Context (if requested)
                    if "context" in signature.parameters:
                        fast_params["context"] = Context(
                            __handler__=cast("BasePath", self), __request__=request
                        )

                    # 4) Request (if requested)
                    if SignatureDefault.REQUEST in signature.parameters:
                        fast_params["request"] = request

                    response = await self._execute_function(func, **fast_params)
                    await self._handle_response_content(response, scope, receive, send)
                    return

            params_from_request = await self._extract_params_from_request(
                request=request,
                signature=signature,
            )

            request_information = self.extract_request_params_information(
                request=request, signature=signature
            )

            func_params: dict[str, Any] = {
                **params_from_request,
                **self._extract_context(request=request, signature=signature),
                **request_information,
            }

            if signature.parameters:
                if SignatureDefault.REQUEST in signature.parameters:
                    func_params.update({"request": request})
                    response = await self._execute_function(func, **func_params)
                else:
                    response = await self._execute_function(func, **func_params)
            else:
                response = await self._execute_function(func, **func_params)

            await self._handle_response_content(response, scope, receive, send)

        await wrap_app_handling_exceptions(inner_app, request)(scope, receive, send)

    return app

extract_request_params_information

extract_request_params_information(request, signature)

Extracts the request information from the request and populates the request information.

Source code in lilya/_internal/_responses.py

def extract_request_params_information(
    self, request: Request, signature: inspect.Signature
) -> dict[str, Any]:
    """
    Extracts the request information from the request and populates the
    request information.
    """

    request_params: dict[str, Any] = {}
    parameters = signature.parameters

    for name, parameter in parameters.items():
        field = parameter.default
        if field is inspect._empty:
            continue

        if isinstance(field, Query):
            source = request.query_params
            key = field.alias or name
        elif isinstance(field, Header):
            source = request.headers  # type: ignore
            key = field.value
        elif isinstance(field, Cookie):
            source = request.cookies  # type: ignore
            key = field.value
        else:
            continue

        try:
            if not isinstance(field, Cookie):
                values = source.getall(key, None)
                if values is None:
                    raw_value = None
                elif isinstance(values, list):
                    raw_value = values[0] if len(values) == 1 else values
                else:
                    raw_value = values  # type: ignore
            else:
                raw_value = source.get(key)
        except (KeyError, TypeError):
            raw_value = None

        if field.required and raw_value is None:
            raise UnprocessableEntity(f"Missing mandatory query parameter '{key}'") from None

        # Fallback to default
        if raw_value is None:
            request_params[name] = field.default if hasattr(field, "default") else None
            continue

        # Apply casting if defined
        try:
            if field.cast and isinstance(raw_value, list):
                request_params[name] = [raw_value]
            elif field.cast:
                request_params[name] = field.resolve(raw_value, field.cast)
            else:
                if isinstance(raw_value, list) and len(raw_value) == 1:
                    request_params[name] = raw_value[0]
                else:
                    request_params[name] = raw_value
        except (TypeError, ValueError):
            raise UnprocessableEntity(
                f"Invalid value for query parameter '{key}': expected {field.cast.__name__}"
            ) from None

    return request_params

is_explicitly_bound

is_explicitly_bound(param)

Checks for explicitly bound parameter from the default.

Source code in lilya/_internal/_responses.py

def is_explicitly_bound(self, param: inspect.Parameter) -> bool:
    """
    Checks for explicitly bound parameter from the default.
    """
    default = None
    if hasattr(param, "default"):
        default = param.default
    if hasattr(param, "value"):
        default = param.value
    return isinstance(default, (Query, Header, Cookie))

handle_websocket_session

handle_websocket_session(func)

Decorator for creating a WebSocket session ASGI application.

PARAMETER	DESCRIPTION
func	The function to be wrapped. TYPE: Callable

RETURNS	DESCRIPTION
ASGIApp	The ASGI application. TYPE: ASGIApp

Source code in lilya/_internal/_responses.py

def handle_websocket_session(self, func: Callable[[WebSocket], Awaitable[None]]) -> ASGIApp:
    """
    Decorator for creating a WebSocket session ASGI application.

    Args:
        func (Callable): The function to be wrapped.

    Returns:
        ASGIApp: The ASGI application.
    """

    async def app(scope: Scope, receive: Receive, send: Send) -> None:
        """
        ASGI application handling WebSocket session.

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

        Returns:
            None
        """
        session = WebSocket(scope=scope, receive=receive, send=send)
        existing = list(scope.get("dependencies", []))
        scope["dependencies"] = existing + [getattr(self, "dependencies", {})]

        async def inner_app(scope: Scope, receive: Receive, send: Send) -> None:
            """
            Inner ASGI application handling WebSocket session.

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

            Returns:
                None
            """
            signature = inspect.signature(func)
            kwargs: dict[str, Any] = {}

            # merge app/include/route deps exactly like HTTP does:
            merged: dict[str, Any] = {}

            # app-level
            app = scope.get("app", None)
            app_obj = getattr(app, "dependencies", {}) or {}
            merged.update(app_obj)

            # include-level
            for inc in scope.get("dependencies", []):
                merged.update(inc)

            # route-level
            route_map = getattr(func, "_lilya_dependencies", {}) or {}
            merged.update(route_map)

            websocket = WebSocket(scope, receive, send)

            overrides = websocket.scope.get("dependency_overrides", {})
            merged.update(overrides)

            # now for each Provides() param, resolve it
            for name, param in signature.parameters.items():
                if isinstance(param.default, Provides):
                    if name not in merged:
                        raise WebSocketException(
                            code=1011,
                            reason=f"Missing dependency '{name}' for websocket handler '{func.__name__}'",
                        )
                    provider = merged[name]

                    if isinstance(provider, (Resolve, Security)):
                        kwargs[name] = await async_resolve_dependencies(
                            request=websocket,
                            func=provider.dependency,
                        )
                        continue
                    data = await provider.resolve(websocket, merged)
                    kwargs[name] = data

                elif isinstance(param.default, (Resolve, Security)):
                    kwargs[name] = await async_resolve_dependencies(
                        request=websocket,
                        func=param.default.dependency,
                    )

                elif param.name in merged and param.default is inspect.Parameter.empty:
                    provider = merged[name]
                    kwargs[name] = await provider.resolve(websocket, merged)

            await self._execute_function(func, session, **kwargs)

        await wrap_app_handling_exceptions(inner_app, session)(scope, receive, send)

    return app

handle_match

handle_match(scope, match)

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) -> 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()
    }
    path_params = {**scope.get("path_params", {}), **matched_params}
    upstream = list(scope.get("dependencies", []))
    child_scope = {
        "handler": self.handler,
        "path_params": path_params,
        "dependencies": upstream + [self.dependencies],
    }

    return Match.FULL, child_scope

validate_params

validate_params(name, path_params)

Validates the route name and path parameters.

PARAMETER	DESCRIPTION
name	The name of the route. TYPE: str
path_params	The path parameters. TYPE: dict

RAISES	DESCRIPTION
NoMatchFound	If there is a mismatch in route name or parameters.

Source code in lilya/routing.py

def validate_params(self, name: str, path_params: dict) -> None:
    """
    Validates the route name and path parameters.

    Args:
        name (str): The name of the route.
        path_params (dict): The path parameters.

    Raises:
        NoMatchFound: If there is a mismatch in route name or parameters.
    """
    seen_params = set(path_params.keys())
    expected_params = set(self.param_convertors.keys())

    if name != self.name or seen_params != expected_params:
        raise NoMatchFound(name, path_params)

handle_dispatch async

handle_dispatch(scope, receive, send)

Handles the interception of messages and calls from the API.

Source code in ravyn/routing/gateways.py

async def handle_dispatch(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
    """
    Handles the interception of messages and calls from the API.
    """
    await self.app(scope, receive, send)

        - path
        - handler
        - parent
        - dependencies
        - middleware
        - interceptors
        - permissions
        - exception_handlers
        - deprecated
        - is_from_router
        - security
        - tags