JWT Token Utilities

Utility functions and data schemas for JWT lifecycle management.

Data Schemas

class TokenPayloadBase(iat: float, exp: float, jti: str, sub: str)[source]

Bases: Struct

Represents the base set of validated JWT claims.

iat: float
exp: float
jti: str
sub: str
property is_expired: bool
class TokenPayloadAccess(iat: float, exp: float, jti: str, sub: str, email: str)[source]

Bases: TokenPayloadBase

Stores payload data specific to the access token.

Inherits all claims from TokenPayloadBase.

email: str
class TokenPayloadRefresh(iat: float, exp: float, jti: str, sub: str)[source]

Bases: TokenPayloadBase

Stores payload data specific to the refresh token.

Inherits all claims from TokenPayloadBase.

JWT Operations

create_access_token(user_id: UUID, email: str) str[source]

Create a short-lived JWT access token.

Parameters:
  • user_id (UUID) – The unique identifier of the user.

  • email (str) – The user’s email address.

Returns:

The encoded access token string.

Return type:

str

create_refresh_token(user_id: UUID) str[source]

Create a long-lived JWT refresh token.

Parameters:

user_id (UUID) – The unique identifier of the user.

Returns:

The encoded refresh token string.

Return type:

str

async get_access_token_payload(token: str) TokenPayloadAccess[source]

Decode an access token and validate its claims via cache fast-path.

Parameters:

token – The raw encoded JWT string.

Returns:

The validated and parsed access token data.

Return type:

TokenPayloadAccess

Raises:

UnauthorizedException – If the token has expired, its signature is invalid, or internal deserialization fails.

get_refresh_token_payload(token: str) TokenPayloadRefresh[source]

Decode a refresh token and validate its claims.

Parameters:

token – The raw encoded JWT string.

Returns:

The validated and parsed refresh token data.

Return type:

TokenPayloadRefresh

Raises:

UnauthorizedException – If the token has expired, its signature is invalid, or internal deserialization fails.

Token Management & Revocation

get_unverified_jti(token: str) str | None[source]

Extract the jti claim from a JWT token without validation.

async add_token_to_blacklist(refresh_token_identifier: str, ttl: int) None[source]

Add a refresh token identifier (JTI) to cache for revocation.

async is_token_in_blacklist(refresh_token_identifier: str) bool[source]

Check if a refresh token identifier (JTI) exists in the blacklist.