`Gateway` class¶

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

How to import¶

from ravyn import Gateway

ravyn.Gateway ¶

Gateway(
    path=None,
    *,
    handler,
    name=None,
    include_in_schema=True,
    parent=None,
    dependencies=None,
    middleware=None,
    interceptors=None,
    permissions=None,
    exception_handlers=None,
    before_request=None,
    after_request=None,
    deprecated=None,
    is_from_router=False,
    security=None,
    tags=None,
    operation_id=None,
)

Bases: Path, Dispatcher, _GatewayCommon

Gateway object class used by Ravyn routes.

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

Read more about Gateway and how to use it.

Example

from ravyn import Ravyn. get

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

Gateway(path="/home", handler=home)

PARAMETER	DESCRIPTION
`path`	Relative path of the `Gateway`. The path can contain parameters in a dictionary like format and if the path is not provided, it will default to `/`. Example `Gateway()` Example with parameters `Gateway(path="/{age: int}")` TYPE: `Optional[str]` DEFAULT: `None`
`handler`	An instance of handler. TYPE: `Union[HTTPHandler, BaseController, Type[BaseController], Type[HTTPHandler]]`
`name`	The name for the Gateway. The name can be reversed by `url_path_for()`. TYPE: `Optional[str]` DEFAULT: `None`
`include_in_schema`	Boolean flag indicating if it should be added to the OpenAPI docs. TYPE: `bool` DEFAULT: `True`
`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. Example `from edgy import Database, Registry from ravyn import Ravyn, Request, Gateway, get database = Database("postgresql+asyncpg://user:password@host:port/database") registry = Registry(database=database) async def create_user(request: Request): # Logic to create the user data = await request.json() ... app = Ravyn( routes=[Gateway("/create", handler=create_user)], after_request=[database.disconnect], )` TYPE: `Union[Sequence[Callable[[], Any]], None]` DEFAULT: `None`
`after_request`	A `list` of events that are trigger after the application processes the request. Read more about the events. Example `from edgy import Database, Registry from ravyn import Ravyn, Request, Gateway, get database = Database("postgresql+asyncpg://user:password@host:port/database") registry = Registry(database=database) async def create_user(request: Request): # Logic to create the user data = await request.json() ... app = Ravyn( routes=[Gateway("/create", handler=create_user)], after_request=[database.disconnect], )` TYPE: `Union[Sequence[Callable[[], Any]], None]` DEFAULT: `None`
`deprecated`	Boolean flag for indicating the deprecation of the Gateway and to display it in the OpenAPI documentation.. TYPE: `Optional[bool]` 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`
`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`
`operation_id`	Unique operation id that allows distinguishing the same handler in different Gateways. Used for OpenAPI purposes. TYPE: `Optional[str]` DEFAULT: `None`

Source code in ravyn/routing/gateways.py

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

            **Example**

            ```python
            Gateway()
            ```

            **Example with parameters**

            ```python
            Gateway(path="/{age: int}")
            ```
            """
        ),
    ] = None,
    *,
    handler: Annotated[
        Union["HTTPHandler", BaseController, Type[BaseController], Type["HTTPHandler"]],
        Doc(
            """
        An instance of [handler](https://ravyn.dev/routing/handlers/#http-handlers).
        """
        ),
    ],
    name: Annotated[
        Optional[str],
        Doc(
            """
            The name for the Gateway. The name can be reversed by `url_path_for()`.
            """
        ),
    ] = None,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if it should be added to the OpenAPI docs.
            """
        ),
    ] = True,
    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/).

            **Example**

            ```python
            from edgy import Database, Registry

            from ravyn import Ravyn, Request, Gateway, get

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

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


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

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

            **Example**

            ```python
            from edgy import Database, Registry

            from ravyn import Ravyn, Request, Gateway, get

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


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


            app = Ravyn(
                routes=[Gateway("/create", handler=create_user)],
                after_request=[database.disconnect],
            )
            ```
            """
        ),
    ] = None,
    deprecated: Annotated[
        Optional[bool],
        Doc(
            """
            Boolean flag for indicating the deprecation of the Gateway and to display it
            in the OpenAPI documentation..
            """
        ),
    ] = 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,
    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,
    operation_id: Annotated[
        Optional[str],
        Doc(
            """
            Unique operation id that allows distinguishing the same handler in different Gateways.

            Used for OpenAPI purposes.
            """
        ),
    ] = None,
) -> 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
    )
    self.methods = getattr(handler, "http_methods", None)
    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),
        include_in_schema=include_in_schema,
        name=resolved_name,
        methods=self.methods,
        middleware=prepared_middleware,
        exception_handlers=exception_handlers,
        permissions=lilya_permissions or {},  # type: ignore
    )

    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.name = resolved_name
    self.handler = cast("Callable", handler)
    self.dependencies = dependencies or {}  # type: ignore
    self.interceptors = list(interceptors or [])
    self.deprecated = deprecated
    self.parent = parent
    self.security = security
    self.tags = list(tags or [])
    self.response_class = None
    self.response_cookies = None
    self.response_headers = None
    self.operation_id = operation_id
    self.lilya_permissions = lilya_permissions or {}
    self.include_in_schema = include_in_schema

    self._compile(handler, self.path)

    if self.is_handler(self.handler):
        if self.operation_id or getattr(handler, "operation_id", None) is not None:
            generated = self.generate_operation_id(self.name or "", self.handler)  # type: ignore
            self.operation_id = f"{operation_id}_{generated}" if operation_id else generated
        elif not getattr(handler, "operation_id", None):
            handler.operation_id = self.generate_operation_id(self.name or "", self.handler)  # type: ignore

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_response(handler)

middleware `instance-attribute` ¶

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

permissions `instance-attribute` ¶

permissions = permissions if permissions is not None else []

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

methods `instance-attribute` ¶

methods = getattr(handler, 'http_methods', None)

before_request `instance-attribute` ¶

before_request = list(before_request or [])

after_request `instance-attribute` ¶

after_request = list(after_request or [])

name `instance-attribute` ¶

name = resolved_name

handler `instance-attribute` ¶

handler = cast('Callable', handler)

dependencies `instance-attribute` ¶

dependencies = dependencies or {}

interceptors `instance-attribute` ¶

interceptors = list(interceptors or [])

deprecated `instance-attribute` ¶

deprecated = deprecated

parent `instance-attribute` ¶

parent = parent

security `instance-attribute` ¶

security = security

tags `instance-attribute` ¶

tags = list(tags or [])

response_class `instance-attribute` ¶

response_class = None

response_cookies `instance-attribute` ¶

response_cookies = None

response_headers `instance-attribute` ¶

response_headers = None

operation_id `instance-attribute` ¶

operation_id = operation_id

lilya_permissions `instance-attribute` ¶

lilya_permissions = lilya_permissions or {}

include_in_schema `instance-attribute` ¶

include_in_schema = include_in_schema

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.HTTP:
        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.HTTP.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

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_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],
    }

    if self.methods and scope["method"] not in self.methods:
        return Match.PARTIAL, child_scope
    else:
        return Match.FULL, child_scope

handle_controller `async` ¶

handle_controller(scope, receive, send)

Instantiates the Controller object and executes the call.

Source code in lilya/routing.py

async def handle_controller(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Instantiates the Controller object and executes
    the call.
    """
    app = self.app()  # type: ignore[call-arg]
    await app(scope, receive, send)  # type: ignore

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
        - include_in_schema
        - deprecated
        - security
        - tags

Gateway class¶

How to import¶

ravyn.Gateway ¶

handler_signature property ¶

path_parameters property ¶

stringify_parameters property ¶

parent_levels property ¶

dependency_names property ¶

signature property ¶

app instance-attribute ¶

middleware instance-attribute ¶

permissions instance-attribute ¶

exception_handlers instance-attribute ¶

wrapped_permissions instance-attribute ¶

path instance-attribute ¶

methods instance-attribute ¶

before_request instance-attribute ¶

after_request instance-attribute ¶

name instance-attribute ¶

handler instance-attribute ¶

dependencies instance-attribute ¶

interceptors instance-attribute ¶

deprecated instance-attribute ¶

parent instance-attribute ¶

security instance-attribute ¶

tags instance-attribute ¶

response_class instance-attribute ¶

response_cookies instance-attribute ¶

response_headers instance-attribute ¶

operation_id instance-attribute ¶

lilya_permissions instance-attribute ¶

include_in_schema instance-attribute ¶

is_class_based staticmethod ¶

is_handler staticmethod ¶

generate_operation_id ¶

handle_middleware staticmethod ¶

parse_path ¶

get_response_for_handler ¶

get_response_for_request async ¶

create_signature_model ¶

create_handler_transformer_model ¶

get_lookup_path ¶

get_dependencies ¶

is_unique_dependency staticmethod ¶

get_cookies ¶

get_headers ¶

get_response_data async ¶

allow_connection async ¶

dispatch_allow_connection async ¶

get_security_schemes ¶

get_handler_tags ¶

handle_signature ¶

search ¶

path_for ¶

url_path_for ¶

dispatch async ¶

handle_not_found async ¶

handle_not_found_fallthrough async staticmethod ¶

handle_exception_handlers async ¶

extract_request_information ¶

handle_response ¶

extract_request_params_information ¶

is_explicitly_bound ¶

handle_websocket_session ¶

validate_params ¶

handle_match ¶

handle_controller async ¶

handle_dispatch async ¶

`Gateway` class¶

handler_signature `property` ¶

path_parameters `property` ¶

stringify_parameters `property` ¶

parent_levels `property` ¶

dependency_names `property` ¶

signature `property` ¶

app `instance-attribute` ¶

middleware `instance-attribute` ¶

permissions `instance-attribute` ¶

exception_handlers `instance-attribute` ¶

wrapped_permissions `instance-attribute` ¶

path `instance-attribute` ¶

methods `instance-attribute` ¶

before_request `instance-attribute` ¶

after_request `instance-attribute` ¶

name `instance-attribute` ¶

handler `instance-attribute` ¶

dependencies `instance-attribute` ¶

interceptors `instance-attribute` ¶

deprecated `instance-attribute` ¶

parent `instance-attribute` ¶

security `instance-attribute` ¶

tags `instance-attribute` ¶

response_class `instance-attribute` ¶

response_cookies `instance-attribute` ¶

response_headers `instance-attribute` ¶

operation_id `instance-attribute` ¶

lilya_permissions `instance-attribute` ¶

include_in_schema `instance-attribute` ¶

is_class_based `staticmethod` ¶

is_handler `staticmethod` ¶

handle_middleware `staticmethod` ¶

get_response_for_request `async` ¶

is_unique_dependency `staticmethod` ¶

get_response_data `async` ¶

allow_connection `async` ¶

dispatch_allow_connection `async` ¶

dispatch `async` ¶

handle_not_found `async` ¶

handle_not_found_fallthrough `async` `staticmethod` ¶

handle_exception_handlers `async` ¶

handle_controller `async` ¶

handle_dispatch `async` ¶