Interceptors¶

Interceptors are special Esmerald objects that implement the InterceptorProtocol via EsmeraldInterceptor.

What are the interceptors¶

In many occasions you will find yourself needing some sort of logic that captures the request before hitting your final API endpoint.

These can be extremely useful, for example, to capture some data before passing to the API, for logging anything particulary useful or any other useful logic.

Interceptors

Overview¶

Interceptors have a set of useful capabilities inspired by AOP (Aspect Oriented Programming) techniques. This makes it possible to:

Add extra logic before request
Throw exceptions before hitting the route handler
Extend basic the functionality
Add extra logic to it. E.g: Caching, logging...

And whatever you might see suitable.

Esmerald does not implement two way method execution, meaning, interceptors are used to capture the request but not the response.

EsmeraldInterceptor¶

This is the main object that should be used to create your own interceptors. Every class should derive from this object and implement the intercept functionality.

from esmerald import EsmeraldInterceptor

or

from esmerald.interceptors.interceptor import EsmeraldInterceptor

Example¶

Let us assume you need to create one interceptor that will log a simple message before hitting the route handler.

We will be creating:

A logging interceptor
The route handler

The logging interceptor

from loguru import logger

from esmerald import EsmeraldInterceptor
from lilya.types import Receive, Scope, Send


class LoggingInterceptor(EsmeraldInterceptor):
    async def intercept(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        # Log a message here
        logger.info("This is my interceptor being called before reaching the handler.")

The application with handlers and applying the interceptor

from esmerald import Esmerald, Gateway, JSONResponse, get

from .myapp.interceptors import LoggingInterceptor


@get("/home")
async def home() -> JSONResponse:
    return JSONResponse({"message": "Welcome home"})


app = Esmerald(routes=[Gateway(handler=home, interceptors=[LoggingInterceptor])])

Custom interceptor¶

Is this the only way of creating an interceptor? No but it is advised to subclass the EsmeraldInterceptor as shown above.

Let us see how it would look like the same app with a custom interceptor then.

The logging interceptor

from loguru import logger

from lilya.types import Receive, Scope, Send


class LoggingInterceptor:
    async def intercept(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        # Log a message here
        logger.info("This is my interceptor being called before reaching the handler.")

The application with handlers and applying the interceptor

from esmerald import Esmerald, Gateway, JSONResponse, get

from .myapp.interceptors import LoggingInterceptor


@get("/home")
async def home() -> JSONResponse:
    return JSONResponse({"message": "Welcome home"})


app = Esmerald(routes=[Gateway(handler=home, interceptors=[LoggingInterceptor])])

It is very similar correct? Yes but the main difference here happens within the EsmeraldInterceptor as this one implements the InterceptorProtocol from Esmerald and therefore makes it the right way of using it.

Interceptors and levels¶

Like everything in Esmerald works in levels, the interceptors are no exception to this but has some constraints.

The interceptors only work on Esmerald, ChildEsmerald, Router, Gateway, WebsocketGateway and Include and do not work on handlers directly.
When working with Esmerald and ChildEsmerald, the interceptors work in isolation.

Examples using levels¶

Let us assume we have two interceptors. One intercepts and changes the value of the request parameter and another tries to parse a value into an int type.

The examples below are just that, examples and you will not be doing too much with those but you can get the idea of it.

RequestParamInterceptor

from esmerald import EsmeraldInterceptor
from esmerald.requests import Request
from lilya.types import Receive, Scope, Send


class RequestParamInterceptor(EsmeraldInterceptor):
    async def intercept(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        request = Request(scope=scope, receive=receive, send=send)
        request.path_params["name"] = "intercept"

CookieInterceptor

from esmerald import EsmeraldInterceptor
from esmerald.exceptions import NotAuthorized
from esmerald.requests import Request
from lilya.types import Receive, Scope, Send


class CookieInterceptor(EsmeraldInterceptor):
    async def intercept(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
        request = Request(scope=scope, receive=receive, send=send)
        max_length = request.cookies["max_length"]

        try:
            int(max_length)
        except (TypeError, ValueError):
            raise NotAuthorized()

The application

The application calling both interceptors on different levels, the app level and the gateway level.

from esmerald import Esmerald, Gateway, JSONResponse, get

from .myapp.interceptors import CookieInterceptor, RequestParamInterceptor


@get("/home")
async def home() -> JSONResponse:
    return JSONResponse({"message": "Welcome home"})


app = Esmerald(
    routes=[Gateway(handler=home, interceptors=[CookieInterceptor])],
    interceptors=[RequestParamInterceptor],
)

The same logic can algo be applied to Include and nested Include.

All of the levels described here allow to pass interceptors.

Working in isolation¶

Every Esmerald and ChildEsmerald application is considered independent, which means, the resources can be "isolated" but Esmerald also allows the share of the resources across parent and children.

For example, using the example from before, adding RequestParamInterceptor on the top of an Esmerald app and adding the CookieInterceptor in the ChildEsmerald will work separately.

from esmerald import ChildEsmerald, Esmerald, Gateway, Include, JSONResponse, get

from .myapp.interceptors import CookieInterceptor, RequestParamInterceptor


@get("/home")
async def home() -> JSONResponse:
    return JSONResponse({"message": "Welcome home"})


@get("/")
async def home_child() -> JSONResponse:
    return JSONResponse({"message": "Welcome home, child"})


child_esmerald = ChildEsmerald(
    routes=[Gateway(handler=home_child, interceptors=[CookieInterceptor])]
)

app = Esmerald(
    routes=[Include("/child", app=child_esmerald), Gateway(handler=home)],
    interceptors=[RequestParamInterceptor],
)

The RequestParamInterceptor will work for the routes of the Esmerald and subsequent chilren, the ChildEsmerald, which means, you can also achieve the same result by doing this:

from esmerald import ChildEsmerald, Esmerald, Gateway, Include, JSONResponse, get

from .myapp.interceptors import CookieInterceptor, RequestParamInterceptor


@get("/home")
async def home() -> JSONResponse:
    return JSONResponse({"message": "Welcome home"})


@get("/")
async def home_child() -> JSONResponse:
    return JSONResponse({"message": "Welcome home, child"})


child_esmerald = ChildEsmerald(
    routes=[Gateway(handler=home_child, interceptors=[CookieInterceptor, RequestParamInterceptor])]
)

app = Esmerald(
    routes=[Include("/child", app=child_esmerald), Gateway(handler=home)],
)

Tip

Prior to version 1.0.0, sharing resources between Esmerald and ChildEsmerald was not allowed and it needed to be treated as completely isolated application. In the version 1.0.0 you can still isolate them but you can also share resources.

Interceptors and the application¶

To add interceptors to the main application as defaults, the way of doing it is by passing the parameters when creating the application.

from esmerald import Esmerald, Gateway, JSONResponse, get

from .myapp.interceptors import RequestParamInterceptor


@get("/home")
async def home() -> JSONResponse:
    return JSONResponse({"message": "Welcome home"})


app = Esmerald(
    routes=[Gateway(handler=home)],
    interceptors=[RequestParamInterceptor],
)

Interceptors and settings module¶

Like everything in Esmerald, the settings also allow to pass the interceptors instead of passing directly when creating the Esmerald intance. A cleaner way of doing it.

settings.py

from typing import TYPE_CHECKING, List

from esmerald import EsmeraldAPISettings

from .myapp.interceptors import RequestParamInterceptor

if TYPE_CHECKING:
    from esmerald.interceptors.types import Interceptor


class AppSettings(EsmeraldAPISettings):
    def interceptors(self) -> List[Interceptor]:
        """
        Loads the default interceptors from the settings.
        """
        return [RequestParamInterceptor]

app.py

from esmerald import Esmerald, Gateway, JSONResponse, get


@get("/home")
async def home() -> JSONResponse:
    return JSONResponse({"message": "Welcome home"})


app = Esmerald(routes=[Gateway(handler=home)])

With the settings and the app created you can simply start the server and pass the newly settings module.

ESMERALD_SETTINGS_MODULE=settings.AppSettings uvicorn src:app --reload

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [28720]
INFO:     Started server process [28722]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Note

You should replace the location of the settings in the example by the one you have in your project. This was used as example only.

API Reference¶

Check out the API Reference for Interceptors for more details.