Skip to content

Example

Since Mongoz if from the same author of Esmerald, it gives some extra motivation for its use and therefore an example in how to use the JWTAuthMiddleware, even if in a very simplistic way, within your Esmerald application.

Let us build a simple integration and application where we will be creating:

We will be using a local MongoDB database for this example but feel free to integrate with your database.

Tip

For local development you can always use docker as it provides a clean way of using MongoDB without going throught the hassle of installing everything locally.

We will also be assuming the following:

  • Documents are inside an accounts/documents.py
  • Controllers/APIs are inside an accounts/controllers.py
  • The main application is inside an app.py
  • The jwt_config is inside your global settings.

Lets go!

Create user document

First, we need to create a document that will be storing the users in the system. We will be defaulting to the one document provided by Esmerald out-of-the-box.

accounts/documents.py
from datetime import datetime
from enum import Enum

import mongoz
from edgy import Registry

from esmerald.contrib.auth.edgy.base_user import User as BaseUser

database = "mongodb://localhost:27017"
registry = Registry(database)


class UserType(str, Enum):
    ADMIN = "admin"
    USER = "user"
    OTHER = "other"


class Role(mongoz.EmbeddedDocument):
    name: str = mongoz.String(max_length=255, default=UserType.USER)


class User(BaseUser):
    """
    Inherits from the BaseUser all the fields and adds extra unique ones.
    """

    date_of_birth: datetime = mongoz.Date()
    is_verified: bool = mongoz.Boolean(default=False)
    role: Role = mongoz.Embed(Role)

    class Meta:
        registry = registry
        database = "my_db"

    def __str__(self):
        return f"{self.email} - {self.role.name}"

Create user API

Now that the user document is defined and created, it is time to create an api that allows the creation of users in the system.

This example won't cover corner cases like integrity in case of duplicates and so on as this is something that you can easily manage.

accounts/controllers.py
from accounts.models import User
from pydantic import BaseModel

from esmerald import post


class UserIn(BaseModel):
    first_name: str
    last_name: str
    email: str
    password: str
    username: str


@post(tags=["user"])
async def create_user(data: UserIn) -> None:
    """
    Creates a user in the system and returns the default 201
    status code.
    """
    await User.create_user(
        first_name=data.first_name,
        last_name=data.last_name,
        email=data.email,
        password=data.password,
        username=data.username,
    )

Login API

Now the create user is available to us to be used later on, we need a view that also allow us to login and return the JWT access token.

For this API to work, we need to guarantee the data being sent is valid, authenticate and then return the JWT token.

accounts/controllers.py
from datetime import datetime, timedelta

from accounts.models import User
from mongoz.exceptions import DocumentNotFound
from pydantic import BaseModel

from esmerald import JSONResponse, post, status
from esmerald.conf import settings
from esmerald.security.jwt.token import Token


class LoginIn(BaseModel):
    email: str
    password: str


class BackendAuthentication(BaseModel):
    """
    Utility class that helps with the authentication process.
    """

    email: str
    password: str

    async def authenticate(self) -> str:
        """Authenticates a user and returns a JWT string"""
        try:
            user: User = await User.objects.get(email=self.email)
        except DocumentNotFound:
            # Run the default password hasher once to reduce the timing
            # difference between an existing and a nonexistent user.
            await User().set_password(self.password)
        else:
            is_password_valid = await user.check_password(self.password)
            if is_password_valid and self.is_user_able_to_authenticate(user):
                # The lifetime of a token should be short, let us make 5 minutes.
                # You can use also the access_token_lifetime from the JWT config directly
                time = datetime.now() + settings.jwt_config.access_token_lifetime
                return self.generate_user_token(user, time=time)

    def is_user_able_to_authenticate(self, user):
        """
        Reject users with is_active=False. Custom user models that don't have
        that attribute are allowed.
        """
        return getattr(user, "is_active", True)

    def generate_user_token(self, user: User, time=None):
        """
        Generates the JWT token for the authenticated user.
        """
        if not time:
            later = datetime.now() + timedelta(minutes=20)
        else:
            later = time

        token = Token(sub=user.id, exp=later)
        return token.encode(
            key=settings.jwt_config.signing_key, algorithm=settings.jwt_config.algorithm
        )


@post(status_code=status.HTTP_200_OK, tags=["auth"])
async def login(data: LoginIn) -> JSONResponse:
    """
    Login a user and returns a JWT token, else raises ValueError
    """
    auth = BackendAuthentication(email=data.email, password=data.password)
    token = await auth.authenticate()
    return JSONResponse({settings.jwt_config.access_token_name: token})

Ooof! There is a lot going on here right? Well, yes but this is also intentional. The login is actually very simple, it just receives a payload and throws that payload into validation inside the BackendAuthentication.

For those familiar with similar objects, like Django backends, this BackendAuthentication does roughly the same thing and it is quite robust since it is using pydantic when creating the instance which takes advantage of the validations automatically for you.

The BackendAuthentication once created inside the login and validated with the given fields, simply proceeds with the authenticate method where it will return the JWT for the user.

Warning

As mentioned before in the assumptions on the top of the document, it was assumed you put your jwt_config inside your global settings.

Home API

Now it is time to create the api that will be returning the email of the logged in user when hit. The API is pretty much simple and clean.

accounts/controllers.py
from esmerald import JSONResponse, Request, get


@get(tags=["home"])
async def home(request: Request) -> JSONResponse:
    """
    Esmerald request has a `user` property that also
    comes from its origins (Lilya).

    When building an authentication middleware, it
    is recommended to inherit from the `BaseAuthMiddleware`.

    See more info here: https://esmerald.dymmond.com/middleware/middleware/?h=baseauthmiddleware#baseauthmiddleware
    """
    return JSONResponse({"message": f"hello {request.user.email}"})

Assemble the APIs

Now it the time where we assemble everything in one place and create our Esmerald application.

app.py
#!/usr/bin/env python
"""
Generated by 'esmerald createproject'
"""
import os
import sys
from pathlib import Path

from esmerald import Esmerald, Gateway, Include
from esmerald.conf import settings
from esmerald.contrib.auth.mongoz.middleware import JWTAuthMiddleware
from lilya.middleware import DefineMiddleware as LilyaMiddleware


def build_path():
    """
    Builds the path of the project and project root.
    """
    Path(__file__).resolve().parent.parent
    SITE_ROOT = os.path.dirname(os.path.realpath(__file__))

    if SITE_ROOT not in sys.path:
        sys.path.append(SITE_ROOT)
        sys.path.append(os.path.join(SITE_ROOT, "apps"))


def get_application():
    """
    This is optional. The function is only used for organisation purposes.
    """
    build_path()
    from accounts.models import User
    from accounts.views import create_user, home, login

    app = Esmerald(
        routes=[
            Gateway("/login", handler=login),
            Gateway("/create", handler=create_user),
            Include(
                routes=[Gateway(handler=home)],
                middleware=[
                    LilyaMiddleware(JWTAuthMiddleware, config=settings.jwt_config, user_model=User)
                ],
            ),
        ],
    )
    return app


app = get_application()

Did you notice the import of the JWTAuthMiddleware is inside the Include and not in the main Esmerald instance?

It is intentional! Each include handles its own middlewares and to create a user and login you don't want to be logged-in and for that reason, the JWTAuthMiddleware is only for those endpoints that require authentication.

Now this assembling is actually very clean, right? Yes and the reason for that is because Esmerald itself promotes clean design.

We have imported all the APIs directly in the app.py but this is not mandatory. You can take advantage of the Include and clean your application even more.

Refreshing the token

All of these APIs are great to start with but an application using JWT usually needs something that allows to refresh the existing token. That process can be done in many different ways.

Esmerald provides an example how to refresh the token with details that can serve and help you with your process.

The example contains ways of taking advantage of the existing tools provided by Esmerald as well as assumptions how to structure it.

Check out how to implement a refresh token.

Extra

Come on, give it a try, create your own version and then try to access the home.

Let us see how we could access / using the current setup.

For this will be using httpx but you are free to use whatever client you prefer.

Steps

  1. Create a user.
  2. Login and get the jwt token.
  3. Access the home /.
import httpx

# The password is automatically encrypted when using the
# User model provided by Esmerald
user_data = {
    "first_name": "John",
    "last_name": "Doe",
    "email": "john@doe.com",
    "username": "john.doe",
    "password": "johnspassword1234@!",
}

# Create a user
# This returns a 201
async with httpx.AsyncClient() as client:
    client.post("/create", json=user_data)

# Login the user
# Returns the response with the JWT token
user_login = {"email": user_data["email"], "password": user_data["password"]}

async with httpx.AsyncClient() as client:
    response = client.post("/login", json=user_login)

# Access the home '/' endpoint
# The default header for the JWTConfig used is `Authorization``
# The default auth_header_types of the JWTConfig is ["Bearer"]
access_token = response.json()["access_token"]

async with httpx.AsyncClient() as client:
    response = client.get("/", headers={"Authorization": f"Bearer {access_token}"})

print(response.json()["message"])
# hello john@doe.com

Did you notice the Authorization in the headers? Well that is because the default api_key_header from the JWTConfig is called Authorization and the contrib middleware from Esmerald to provide integration with Mongoz uses it to validate if is passed in the header or not.

Like everything in Esmerald, that is also configurable. If you change the header to something else in that config, it will automatically reflect across the contib middlewares.

Conclusions

This is just a simple example how you could use Mongoz with the provided JWTAuthMiddleware from Esmerald and build a quick, yet robust, login system and access protected APIs.