Token strategy

To make library more flexible, and independent from any specific token type(JWT, Session, Redis, etc.), it uses TokenStrategy class. TokenStrategy is a class responsible for token decoding, and encoding.

class TokenStrategy(Generic[UP, ID], ABC):
    def __init__(self, config: FastAuthConfig):
        self._config = config

    @abstractmethod
    async def read_token(self, token: str, **kwargs) -> dict[str, Any]:
        raise NotImplementedError

    @abstractmethod
    async def write_token(self, user: UP, token_type: TokenType, **kwargs) -> str:
        raise NotImplementedError

It has 2 methods: read_token and write_token, which are responsible for decoding and encoding token respectively. read_token method take token string decoding it and return token data dict. write_token method take User model and TokenType, decoding it and return token string.

FastAuth supports most popular token strategies:

• JWT
• Redis
• Database

You can create your own strategy, just implement TokenStrategy class interface.

JWTStrategy

JWTStrategy is a class responsible for JWT token decoding, and encoding. It uses pyjwt library under the hood.

JWTStrategy implementation

from typing import Any, Generic

from jwt import DecodeError

from fastauth import exceptions
from fastauth.config import FastAuthConfig
from fastauth.models import ID, UP
from fastauth.strategy.base import TokenStrategy
from fastauth.types import TokenType
from fastauth.utils.jwt_helper import JWTHelper


class JWTStrategy(Generic[UP, ID], TokenStrategy[UP, ID]):
    _config: FastAuthConfig

    def __init__(self, config: FastAuthConfig):
        super().__init__(config)
        self.encoder = JWTHelper(config.JWT_SECRET, config.JWT_ALGORITHM)

    async def read_token(self, token: str, **kwargs) -> dict[str, Any]:
        """
        Read jwt token and return the payload
        :param token: jwt token string
        :param kwargs: Extra PyJWT decoder data(audience, leeway, issuer, etc.)
        :return: Token payload dict
        :raise InvalidToken: If the token is invalid
        """
        try:
            return self.encoder.decode_token(
                token,
                audience=kwargs.pop("audience", self._config.JWT_DEFAULT_AUDIENCE),
                **kwargs,
            )

        except DecodeError as e:
            msg = f"Invalid JWTHelper token: {e}"
            raise exceptions.InvalidToken(msg) from e

    async def write_token(self, user: UP, token_type: TokenType, **kwargs) -> str:
        """
        Write jwt token for the user model
        :param user: User model
        :param token_type: Token type (access or refresh)
        :param kwargs: extra token data(audience, max_age, headers, extra_data)
        :return: Token string
        """

        payload = {
            "sub": str(user.id),
            "type": token_type,
        }
        for field in self._config.USER_FIELDS_IN_TOKEN:
            if user.__dict__.get(field, False):
                payload.update({field: str(user.__dict__[field])})

        max_age = kwargs.pop(
            "max_age",
            (
                self._config.JWT_ACCESS_TOKEN_MAX_AGE
                if token_type == "access"
                else self._config.JWT_REFRESH_TOKEN_MAX_AGE
            ),
        )
        audience = kwargs.pop("audience", self._config.JWT_DEFAULT_AUDIENCE)
        headers = kwargs.pop("headers", None)
        if extra := kwargs.get("extra_data", {}):
            payload.update(extra)

        return self.encoder.encode_token(
            payload,
            token_type,
            max_age=max_age,
            audience=audience,
            headers=headers,
            **kwargs,
        )

RedisStrategy

Not implemented yet

DatabaseStrategy

Not implemented yet

Dependency

You can create dependency callable for token strategy to use it later. First argument of callable should be FastAuthConfig instance. We use DI, because strategy can use some dependencies, like Redis connection, Database connection, etc. For JWTStrategy it looks like this:

from fastauth.strategy import JWTStrategy
from fastauth.config import FastAuthConfig

async def get_strategy(config: FastAuthConfig, **kwargs):
    return JWTStrategy(config)