Router class

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

How to import

from esmerald import Router

esmerald.Router

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

Bases: RoutingMethodsMixin, BaseRouter

The Router object used by Esmerald upon instantiation.

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

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

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

Source code in esmerald/routing/router.py

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

            **Example**

            ```python
            Include()
            ```

            **Example with parameters**

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

            **Example**

            ```python
            from esmerald import ChildEsmerald, Router

            Router('/child', app=ChildEsmerald(...))
            ```
            """
        ),
    ] = 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://esmerald.dev/application/levels/).
            """
        ),
    ] = None,
    on_startup: Annotated[
        Optional[List[LifeSpanHandler]],
        Doc(
            """
            A `list` of events that are trigger upon the application
            starts.

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

            **Example**

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

            from esmerald import Router, Gateway, post

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


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


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


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

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

            **Example**

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

            from esmerald import Router, Gateway, post

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


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


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


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

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

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

            **Example**

            ```python
            from esmerald import Esmerald, Gateway, Request, get, Include


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


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

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

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

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

            **Example**

            ```python
            from esmerald import Router, JSONResponse

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

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

            **Example**

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

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

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

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

            **Example**

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

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

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

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

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

            **Example**

            ```python
            from edgy import Database, Registry

            from esmerald import Esmerald, 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 = Esmerald(
                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 esmerald import Esmerald, 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 = Esmerald(
                routes=[Gateway("/create", handler=create_user)],
                after_request=[database.disconnect],
            )
            ```
            """
        ),
    ] = None,
    tags: Annotated[
        Optional[Sequence[str]],
        Doc(
            """
            A list of strings tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI documentation.

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

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

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

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

    new_routes: list[Any] = []
    for route in routes or []:
        if isinstance(route, WebhookHandler):
            # WebhookHandler is a subclass of HTTPHandler, make sure to not upgrade it
            ...
        elif isinstance(route, HTTPHandler):
            route = Gateway(handler=route)
        elif is_class_and_subclass(route, View):
            route = Gateway(handler=cast(View, route))
        elif isinstance(route, WebSocketHandler):
            route = WebSocketGateway(handler=route)
        if not isinstance(
            route,
            (
                Include,
                Gateway,
                WebSocketGateway,
                LilyaBasePath,
                Host,
                Router,
            ),
        ) or isinstance(cast(Any, route), (WebhookGateway, WebhookHandler)):
            # WebhookGateway is subclass of LilyaBasePath
            raise ImproperlyConfigured(
                f"The route {route} must be of type HTTPHandler, WebSocketHandler, Gateway, WebSocketGateway or Include"
            )
        new_routes.append(route)

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

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

    super().__init__(
        redirect_slashes=redirect_slashes,
        routes=new_routes,
        default=default,
        lifespan=lifespan,
        on_shutdown=on_shutdown,
        on_startup=on_startup,
        permissions=self.__lilya_permissions__,  # type: ignore
        before_request=before_request,
        after_request=after_request,
    )
    self.path = path
    self.on_startup = [] if on_startup is None else list(on_startup)
    self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
    self.parent: Optional[ParentType] = parent or app
    self.dependencies = dependencies or {}
    self.exception_handlers = exception_handlers or {}
    self.interceptors: Sequence[Interceptor] = interceptors or []

    self.permissions: Sequence[Permission] = [
        permission
        for permission in self.__base_permissions__ or []
        if is_esmerald_permission(permission)
    ]  # type: ignore

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

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

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

    self.activate()

path instance-attribute

path = path

parent instance-attribute

parent = parent or app

on_shutdown instance-attribute

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

on_startup instance-attribute

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

redirect_slashes instance-attribute

redirect_slashes = redirect_slashes

default instance-attribute

default = handle_not_found if default is None else default

routes instance-attribute

routes = new_routes

dependencies instance-attribute

dependencies = dependencies or {}

exception_handlers instance-attribute

exception_handlers = exception_handlers or {}

interceptors instance-attribute

interceptors = interceptors or []

permissions instance-attribute

permissions = [permission for permission in __base_permissions__ or [] if is_esmerald_permission(permission)]

middleware instance-attribute

middleware = middleware or []

response_class instance-attribute

response_class = response_class

response_cookies instance-attribute

response_cookies = response_cookies or []

response_headers instance-attribute

response_headers = response_headers or {}

deprecated instance-attribute

deprecated = deprecated

security instance-attribute

security = security or []

tags instance-attribute

tags = tags or []

app async

app(scope, receive, send)

Handle ASGI messages, managing different scopes and routing.

PARAMETER	DESCRIPTION
scope	TYPE: Scope
receive	TYPE: Receive
send	TYPE: Send

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

Source code in lilya/routing.py

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

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

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

    sniffer = SendReceiveSniffer(receive, send)

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

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

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

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

    await self.handle_default(scope, receive, send)

lifespan async

lifespan(scope, receive, send)

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

PARAMETER	DESCRIPTION
scope	TYPE: Scope
receive	TYPE: Receive
send	TYPE: Send

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

Source code in lilya/routing.py

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

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

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

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

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

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

            await startup_complete()
            completed_startup = True
            await receive()

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

    else:
        await shutdown_complete()

add_apiview

add_apiview(value)

Adds an APIView or related to the application routing.

Example

from esmerald import Router, APIView, Gateway, get


class View(APIView):
    path = "/"

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


gateway = Gateway(handler=View)

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

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

Source code in esmerald/routing/router.py

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

    **Example**

    ```python
    from esmerald import Router, APIView, Gateway, get


    class View(APIView):
        path = "/"

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


    gateway = Gateway(handler=View)

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

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

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

add_route

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

Adds a Route to the application routing.

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

Example

from esmerald import get


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


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

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

Source code in esmerald/routing/router.py

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

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

    **Example**

    ```python
    from esmerald import get


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


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

add_websocket_route

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

Adds a websocket Route to the application routing.

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

Example

from esmerald import websocket


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

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


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

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

Source code in esmerald/routing/router.py

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

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

    **Example**

    ```python
    from esmerald import websocket


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

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


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