TokenStrategy

fastauth.strategy.base.TokenStrategy

TokenStrategy(config)

Bases: Generic[UP, ID], ABC

Source code in fastauth/strategy/base.py

def __init__(self, config: FastAuthConfig):
    self._config = config

read_token abstractmethod async

read_token(token, **kwargs)

Decode token and try fetch User model

PARAMETER	DESCRIPTION
token	Token string TYPE: str
kwargs	Extra data DEFAULT: {}

RETURNS	DESCRIPTION
dict[str, Any]	Token payload dict

Source code in fastauth/strategy/base.py

@abstractmethod
async def read_token(self, token: str, **kwargs) -> dict[str, Any]:
    """
    Decode token and try fetch User model
    :param token: Token string
    :param kwargs: Extra data
    :return: Token payload dict
    """
    raise NotImplementedError

write_token abstractmethod async

write_token(user, token_type, **kwargs)

Create token from User model

PARAMETER	DESCRIPTION
user	User model TYPE: UP
token_type	Token type TYPE: TokenType
kwargs	Extra user data DEFAULT: {}

RETURNS	DESCRIPTION
str	Token string

Source code in fastauth/strategy/base.py

@abstractmethod
async def write_token(self, user: UP, token_type: TokenType, **kwargs) -> str:
    """
    Create token from User model
    :param user: User model
    :param token_type: Token type
    :param kwargs: Extra user data
    :return: Token string
    """
    raise NotImplementedError

JWT Token Strategy

fastauth.strategy.jwt.JWTStrategy

JWTStrategy(config)

Bases: Generic[UP, ID], TokenStrategy[UP, ID]

Source code in fastauth/strategy/jwt.py

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

encoder instance-attribute

encoder = JWTHelper(JWT_SECRET, JWT_ALGORITHM)

read_token async

read_token(token, **kwargs)

Read jwt token and return the payload

PARAMETER	DESCRIPTION
token	jwt token string TYPE: str
kwargs	Extra PyJWT decoder data(audience, leeway, issuer, etc.) DEFAULT: {}

RETURNS	DESCRIPTION
dict[str, Any]	Token payload dict

RAISES	DESCRIPTION
InvalidToken	If the token is invalid

Source code in fastauth/strategy/jwt.py

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

write_token async

write_token(user, token_type, **kwargs)

Write jwt token for the user model

PARAMETER	DESCRIPTION
user	User model TYPE: UP
token_type	Token type (access or refresh) TYPE: TokenType
kwargs	extra token data(audience, max_age, headers, extra_data) DEFAULT: {}

RETURNS	DESCRIPTION
str	Token string

Source code in fastauth/strategy/jwt.py

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,
    )