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)

path instance-attribute

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

handler instance-attribute

handler = cast('Callable', handler)

parent instance-attribute

parent = parent

dependencies instance-attribute

dependencies = dependencies or {}

middleware instance-attribute

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

interceptors instance-attribute

interceptors = list(interceptors or [])

permissions instance-attribute

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

exception_handlers instance-attribute

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