Перейти к содержанию

Router class

Warning

The current page still doesn't have a translation for this language.

But you can help translating it: Contributing.

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, tags=None, deprecated=None, security=None)

Bases: 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

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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
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,
    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 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."

    super().__init__(
        redirect_slashes=redirect_slashes,
        routes=new_routes,
        default=default,
        lifespan=lifespan,
        on_shutdown=on_shutdown,
        on_startup=on_startup,
    )
    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] = permissions or []  # 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 = permissions or []

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
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
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

    partial = None

    for route in self.routes:
        match, child_scope = route.search(scope)
        if match == Match.FULL:
            path_handler = PathHandler(
                child_scope=child_scope, scope=scope, receive=receive, send=send
            )
            await self.handle_route(route, path_handler=path_handler)
            return
        elif match == Match.PARTIAL and partial is None:
            partial = route
            partial_scope = child_scope

    if partial is not None:
        await self.handle_partial(partial, partial_scope, scope, receive, send)
        return

    route_path = get_route_path(scope)
    if scope["type"] == ScopeType.HTTP and self.redirect_slashes 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

    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
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
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
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
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,
    )
    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)

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

Source code in esmerald/routing/router.py
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
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,
) -> 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,
    )
    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)

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

Source code in esmerald/routing/router.py
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
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,
) -> 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,
    )
    self.validate_root_route_parent(websocket_gateway)
    self.create_signature_models(websocket_gateway)
    self.routes.append(websocket_gateway)