Authx

authx.main.AuthX

AuthX(config=AuthXConfig(), model=Dict[str, Any])

Bases: _CallbackHandler[T], _ErrorHandler

The base class for AuthX

AuthX enables JWT management within a FastAPI application. Its main purpose is to provide a reusable & simple syntax to protect API with JSON Web Token authentication.

PARAMETER	DESCRIPTION
config	Configuration instance to use. Defaults to AuthXConfig(). TYPE: AuthXConfig DEFAULT: AuthXConfig()
model	Model type hint. Defaults to Dict[str, Any]. TYPE: Optional[T] DEFAULT: Dict[str, Any]

Note

AuthX is a Generic python object. Its TypeVar is not mandatory but helps type hinting furing development

AuthX base object

PARAMETER	DESCRIPTION
config	TYPE: AuthXConfig DEFAULT: AuthXConfig()
model	TYPE: Optional[T] DEFAULT: Dict[str, Any]

PARAMETER	DESCRIPTION
config	Configuration instance to use. Defaults to AuthXConfig(). TYPE: AuthXConfig DEFAULT: AuthXConfig()
model	Model type hint. Defaults to Dict[str, Any]. TYPE: Optional[T] DEFAULT: Dict[str, Any]

Source code in authx/main.py

def __init__(
    self, config: AuthXConfig = AuthXConfig(), model: Optional[T] = Dict[str, Any]
) -> None:
    """AuthX base object

    Args:
        config (AuthXConfig, optional): Configuration instance to use. Defaults to AuthXConfig().
        model (Optional[T], optional): Model type hint. Defaults to Dict[str, Any].
    """
    super().__init__(model=model)
    super(_CallbackHandler, self).__init__()
    self._config = config

MSG_DEFAULT class-attribute instance-attribute

MSG_DEFAULT = 'AuthX Error'

MSG_TOKEN_ERROR class-attribute instance-attribute

MSG_TOKEN_ERROR = 'Token Error'

MSG_MISSING_TOKEN_ERROR class-attribute instance-attribute

MSG_MISSING_TOKEN_ERROR = 'Missing JWT in request'

MSG_MISSING_CSRF_ERROR class-attribute instance-attribute

MSG_MISSING_CSRF_ERROR = 'Missing CSRF double submit token in request'

MSG_TOKEN_TYPE_ERROR class-attribute instance-attribute

MSG_TOKEN_TYPE_ERROR = 'Bad token type'

MSG_REVOKED_TOKEN_ERROR class-attribute instance-attribute

MSG_REVOKED_TOKEN_ERROR = 'Invalid token'

MSG_TOKEN_REQUIRED_ERROR class-attribute instance-attribute

MSG_TOKEN_REQUIRED_ERROR = 'Token required'

MSG_FRESH_TOKEN_REQUIRED_ERROR class-attribute instance-attribute

MSG_FRESH_TOKEN_REQUIRED_ERROR = 'Fresh token required'

MSG_ACCESS_TOKEN_REQUIRED_ERROR class-attribute instance-attribute

MSG_ACCESS_TOKEN_REQUIRED_ERROR = 'Access token required'

MSG_REFRESH_TOKEN_REQUIRED_ERROR class-attribute instance-attribute

MSG_REFRESH_TOKEN_REQUIRED_ERROR = 'Refresh token required'

MSG_CSRF_ERROR class-attribute instance-attribute

MSG_CSRF_ERROR = 'CSRF double submit does not match'

MSG_DECODE_JWT_ERROR class-attribute instance-attribute

MSG_DECODE_JWT_ERROR = 'Invalid Token'

callback_get_model_instance instance-attribute

callback_get_model_instance = None

callback_is_token_in_blocklist instance-attribute

callback_is_token_in_blocklist = None

is_model_callback_set property

is_model_callback_set

Check if callback is set for model instance

is_token_callback_set property

is_token_callback_set

Check if callback is set for token

config property

config

AuthX Configuration getter

RETURNS	DESCRIPTION
AuthXConfig	Configuration BaseSettings TYPE: AuthXConfig

DEPENDENCY property

DEPENDENCY

FastAPI Dependency to return an AuthX sub-object within the route context

BUNDLE property

BUNDLE

FastAPI Dependency to return a AuthX sub-object within the route context

FRESH_REQUIRED property

FRESH_REQUIRED

FastAPI Dependency to enforce valid token availability in request

ACCESS_REQUIRED property

ACCESS_REQUIRED

FastAPI Dependency to enforce presence of an access token in request

REFRESH_REQUIRED property

REFRESH_REQUIRED

FastAPI Dependency to enforce presence of a refresh token in request

ACCESS_TOKEN property

ACCESS_TOKEN

FastAPI Dependency to retrieve access token from request

REFRESH_TOKEN property

REFRESH_TOKEN

FastAPI Dependency to retrieve refresh token from request

CURRENT_SUBJECT property

CURRENT_SUBJECT

FastAPI Dependency to retrieve the current subject from request

fresh_token_required property

fresh_token_required

FastAPI Dependency to enforce presence of a fresh access token in request

access_token_required property

access_token_required

FastAPI Dependency to enforce presence of an access token in request

refresh_token_required property

refresh_token_required

FastAPI Dependency to enforce presence of a refresh token in request

handle_errors

handle_errors(app)

Add the FastAPI.exception_handlers relative to AuthX exceptions

PARAMETER	DESCRIPTION
app	TYPE: FastAPI

PARAMETER	DESCRIPTION
app	the FastAPI application to handle errors for TYPE: FastAPI

Source code in authx/_internal/_error.py

def handle_errors(self, app: FastAPI) -> None:
    """Add the `FastAPI.exception_handlers` relative to AuthX exceptions

    Args:
        app (FastAPI): the FastAPI application to handle errors for
    """
    self._set_app_exception_handler(
        app, exception=exceptions.JWTDecodeError, status_code=422, message=None
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.MissingTokenError,
        status_code=401,
        message=self.MSG_MISSING_TOKEN_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.MissingCSRFTokenError,
        status_code=401,
        message=self.MSG_MISSING_CSRF_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.TokenTypeError,
        status_code=401,
        message=self.MSG_TOKEN_TYPE_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.RevokedTokenError,
        status_code=401,
        message=self.MSG_REVOKED_TOKEN_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.TokenRequiredError,
        status_code=401,
        message=self.MSG_TOKEN_REQUIRED_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.FreshTokenRequiredError,
        status_code=401,
        message=self.MSG_FRESH_TOKEN_REQUIRED_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.AccessTokenRequiredError,
        status_code=401,
        message=self.MSG_ACCESS_TOKEN_REQUIRED_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.RefreshTokenRequiredError,
        status_code=401,
        message=self.MSG_REFRESH_TOKEN_REQUIRED_ERROR,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.CSRFError,
        status_code=401,
        message=self.MSG_CSRF_ERROR,
    )

set_callback_get_model_instance

set_callback_get_model_instance(callback)

Set callback for model instance

PARAMETER	DESCRIPTION
callback	TYPE: ModelCallback[T]

Source code in authx/_internal/_callback.py

def set_callback_get_model_instance(self, callback: ModelCallback[T]) -> None:
    """Set callback for model instance"""
    self.callback_get_model_instance = callback

set_callback_token_blocklist

set_callback_token_blocklist(callback)

Set callback for token

PARAMETER	DESCRIPTION
callback	TYPE: TokenCallback

Source code in authx/_internal/_callback.py

def set_callback_token_blocklist(self, callback: TokenCallback) -> None:
    """Set callback for token"""
    self.callback_is_token_in_blocklist = callback

set_subject_getter

set_subject_getter(callback)

Set the callback to run for subject retrieval and serialization

PARAMETER	DESCRIPTION
callback	TYPE: ModelCallback[T]

Source code in authx/_internal/_callback.py

def set_subject_getter(self, callback: ModelCallback[T]) -> None:
    """Set the callback to run for subject retrieval and serialization"""
    self.set_callback_get_model_instance(callback)

set_token_blocklist

set_token_blocklist(callback)

Set the callback to run for validation of revoked tokens

PARAMETER	DESCRIPTION
callback	TYPE: TokenCallback

Source code in authx/_internal/_callback.py

def set_token_blocklist(self, callback: TokenCallback) -> None:
    """Set the callback to run for validation of revoked tokens"""
    self.set_callback_token_blocklist(callback)

is_token_in_blocklist

is_token_in_blocklist(token, **kwargs)

Check if token is in blocklist

PARAMETER	DESCRIPTION
token	TYPE: str
**kwargs	DEFAULT: {}

Source code in authx/_internal/_callback.py

def is_token_in_blocklist(self, token: str, **kwargs) -> bool:
    """Check if token is in blocklist"""
    if self._check_token_callback_is_set(ignore_errors=True):
        callback: TokenCallback = self.callback_is_token_in_blocklist
        return callback(token, **kwargs)
    return False

load_config

load_config(config)

Loads a AuthXConfig as the new configuration

PARAMETER	DESCRIPTION
config	TYPE: AuthXConfig

PARAMETER	DESCRIPTION
config	Configuration to load TYPE: AuthXConfig

Source code in authx/main.py

def load_config(self, config: AuthXConfig) -> None:
    """Loads a AuthXConfig as the new configuration

    Args:
        config (AuthXConfig): Configuration to load
    """
    self._config = config

get_access_token_from_request async

get_access_token_from_request(request, locations=None)

Dependency to retrieve access token from request

PARAMETER	DESCRIPTION
request	TYPE: Request
locations	TYPE: Optional[TokenLocations] DEFAULT: None

PARAMETER	DESCRIPTION
request	Request to retrieve access token from TYPE: Request
locations	Locations to retrieve token from. Defaults to None. TYPE: Optional[TokenLocations] DEFAULT: None

RAISES	DESCRIPTION
MissingTokenError	When no access token is available in request

RETURNS	DESCRIPTION
RequestToken	Request Token instance for access token type TYPE: RequestToken

Source code in authx/main.py

async def get_access_token_from_request(
    self, request: Request, locations: Optional[TokenLocations] = None
) -> RequestToken:
    """Dependency to retrieve access token from request

    Args:
        request (Request): Request to retrieve access token from
        locations (Optional[TokenLocations], optional): Locations to retrieve token from. Defaults to None.

    Raises:
        MissingTokenError: When no `access` token is available in request

    Returns:
        RequestToken: Request Token instance for `access` token type
    """
    return await self._get_token_from_request(
        request, optional=False, locations=locations
    )

get_refresh_token_from_request async

get_refresh_token_from_request(request, locations=None)

Dependency to retrieve refresh token from request

PARAMETER	DESCRIPTION
request	TYPE: Request
locations	TYPE: Optional[TokenLocations] DEFAULT: None

PARAMETER	DESCRIPTION
request	Request to retrieve refresh token from TYPE: Request
locations	Locations to retrieve token from. Defaults to None. TYPE: Optional[TokenLocations] DEFAULT: None

RAISES	DESCRIPTION
MissingTokenError	When no refresh token is available in request

RETURNS	DESCRIPTION
RequestToken	Request Token instance for refresh token type TYPE: RequestToken

Source code in authx/main.py

async def get_refresh_token_from_request(
    self, request: Request, locations: Optional[TokenLocations] = None
) -> RequestToken:
    """Dependency to retrieve refresh token from request

    Args:
        request (Request): Request to retrieve refresh token from
        locations (Optional[TokenLocations], optional): Locations to retrieve token from. Defaults to None.

    Raises:
        MissingTokenError: When no `refresh` token is available in request

    Returns:
        RequestToken: Request Token instance for `refresh` token type
    """
    return await self._get_token_from_request(
        request, refresh=True, optional=False, locations=locations
    )

verify_token

verify_token(token, verify_type=True, verify_fresh=False, verify_csrf=True)

Verify a request token

PARAMETER	DESCRIPTION
token	TYPE: RequestToken
verify_type	TYPE: bool DEFAULT: True
verify_fresh	TYPE: bool DEFAULT: False
verify_csrf	TYPE: bool DEFAULT: True

PARAMETER	DESCRIPTION
token	RequestToken instance TYPE: RequestToken
verify_type	Apply token type verification. Defaults to True. TYPE: bool DEFAULT: True
verify_fresh	Apply token freshness verification. Defaults to False. TYPE: bool DEFAULT: False
verify_csrf	Apply token CSRF verification. Defaults to True. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
TokenPayload	description TYPE: TokenPayload

Source code in authx/main.py

def verify_token(
    self,
    token: RequestToken,
    verify_type: bool = True,
    verify_fresh: bool = False,
    verify_csrf: bool = True,
) -> TokenPayload:
    """Verify a request token

    Args:
        token (RequestToken): RequestToken instance
        verify_type (bool, optional): Apply token type verification. Defaults to True.
        verify_fresh (bool, optional): Apply token freshness verification. Defaults to False.
        verify_csrf (bool, optional): Apply token CSRF verification. Defaults to True.

    Returns:
        TokenPayload: _description_
    """
    return token.verify(
        key=self.config.public_key,
        algorithms=[self.config.JWT_ALGORITHM],
        verify_fresh=verify_fresh,
        verify_type=verify_type,
        verify_csrf=verify_csrf,
    )

create_access_token

create_access_token(uid, fresh=False, headers=None, expiry=None, data=None, audience=None, *args, **kwargs)

Generate an Access Token

PARAMETER	DESCRIPTION
uid	TYPE: str
fresh	TYPE: bool DEFAULT: False
headers	TYPE: Optional[Dict[str, Any]] DEFAULT: None
expiry	TYPE: Optional[DateTimeExpression] DEFAULT: None
data	TYPE: Optional[Dict[str, Any]] DEFAULT: None
audience	TYPE: Optional[StringOrSequence] DEFAULT: None
*args	DEFAULT: ()
**kwargs	DEFAULT: {}

PARAMETER	DESCRIPTION
uid	Unique identifier to generate token for TYPE: str
fresh	Generate fresh token. Defaults to False. TYPE: bool DEFAULT: False
headers	TODO. Defaults to None. TYPE: Optional[Dict[str, Any]] DEFAULT: None
expiry	Use a user defined expiry claim. Defaults to None. TYPE: Optional[DateTimeExpression] DEFAULT: None
data	Additional data to store in token. Defaults to None. TYPE: Optional[Dict[str, Any]] DEFAULT: None
audience	Audience claim. Defaults to None. TYPE: Optional[StringOrSequence] DEFAULT: None

RETURNS	DESCRIPTION
str	Access Token TYPE: str

Source code in authx/main.py

def create_access_token(
    self,
    uid: str,
    fresh: bool = False,
    headers: Optional[Dict[str, Any]] = None,
    expiry: Optional[DateTimeExpression] = None,
    data: Optional[Dict[str, Any]] = None,
    audience: Optional[StringOrSequence] = None,
    *args,
    **kwargs,
) -> str:
    """Generate an Access Token

    Args:
        uid (str): Unique identifier to generate token for
        fresh (bool, optional): Generate fresh token. Defaults to False.
        headers (Optional[Dict[str, Any]], optional): TODO. Defaults to None.
        expiry (Optional[DateTimeExpression], optional): Use a user defined expiry claim. Defaults to None.
        data (Optional[Dict[str, Any]], optional): Additional data to store in token. Defaults to None.
        audience (Optional[StringOrSequence], optional): Audience claim. Defaults to None.

    Returns:
        str: Access Token
    """
    return self._create_token(
        uid=uid,
        type="access",
        fresh=fresh,
        headers=headers,
        expiry=expiry,
        data=data,
        audience=audience,
    )

create_refresh_token

create_refresh_token(uid, headers=None, expiry=None, data=None, audience=None, *args, **kwargs)

Generate a Refresh Token

PARAMETER	DESCRIPTION
uid	TYPE: str
headers	TYPE: Optional[Dict[str, Any]] DEFAULT: None
expiry	TYPE: Optional[DateTimeExpression] DEFAULT: None
data	TYPE: Optional[Dict[str, Any]] DEFAULT: None
audience	TYPE: Optional[StringOrSequence] DEFAULT: None
*args	DEFAULT: ()
**kwargs	DEFAULT: {}

PARAMETER	DESCRIPTION
uid	Unique identifier to generate token for TYPE: str
headers	TODO. Defaults to None. TYPE: Optional[Dict[str, Any]] DEFAULT: None
expiry	Use a user defined expiry claim. Defaults to None. TYPE: Optional[DateTimeExpression] DEFAULT: None
data	Additional data to store in token. Defaults to None. TYPE: Optional[Dict[str, Any]] DEFAULT: None
audience	Audience claim. Defaults to None. TYPE: Optional[StringOrSequence] DEFAULT: None

RETURNS	DESCRIPTION
str	Refresh Token TYPE: str

Source code in authx/main.py

def create_refresh_token(
    self,
    uid: str,
    headers: Optional[Dict[str, Any]] = None,
    expiry: Optional[DateTimeExpression] = None,
    data: Optional[Dict[str, Any]] = None,
    audience: Optional[StringOrSequence] = None,
    *args,
    **kwargs,
) -> str:
    """Generate a Refresh Token

    Args:
        uid (str): Unique identifier to generate token for
        headers (Optional[Dict[str, Any]], optional): TODO. Defaults to None.
        expiry (Optional[DateTimeExpression], optional): Use a user defined expiry claim. Defaults to None.
        data (Optional[Dict[str, Any]], optional): Additional data to store in token. Defaults to None.
        audience (Optional[StringOrSequence], optional): Audience claim. Defaults to None.

    Returns:
        str: Refresh Token
    """
    return self._create_token(
        uid=uid,
        type="refresh",
        headers=headers,
        expiry=expiry,
        data=data,
        audience=audience,
    )

set_access_cookies

set_access_cookies(token, response, max_age=None)

Add 'Set-Cookie' for access token in response header

PARAMETER	DESCRIPTION
token	TYPE: str
response	TYPE: Response
max_age	TYPE: Optional[int] DEFAULT: None

PARAMETER	DESCRIPTION
token	Access token TYPE: str
response	response to set cookie on TYPE: Response
max_age	Max Age cookie paramater. Defaults to None. TYPE: Optional[int] DEFAULT: None

Source code in authx/main.py

def set_access_cookies(
    self,
    token: str,
    response: Response,
    max_age: Optional[int] = None,
) -> None:
    """Add 'Set-Cookie' for access token in response header

    Args:
        token (str): Access token
        response (Response): response to set cookie on
        max_age (Optional[int], optional): Max Age cookie paramater. Defaults to None.
    """
    self._set_cookies(
        token=token, type="access", response=response, max_age=max_age
    )

set_refresh_cookies

set_refresh_cookies(token, response, max_age=None)

Add 'Set-Cookie' for refresh token in response header

PARAMETER	DESCRIPTION
token	TYPE: str
response	TYPE: Response
max_age	TYPE: Optional[int] DEFAULT: None

PARAMETER	DESCRIPTION
token	Refresh token TYPE: str
response	response to set cookie on TYPE: Response
max_age	Max Age cookie paramater. Defaults to None. TYPE: Optional[int] DEFAULT: None

Source code in authx/main.py

def set_refresh_cookies(
    self,
    token: str,
    response: Response,
    max_age: Optional[int] = None,
) -> None:
    """Add 'Set-Cookie' for refresh token in response header

    Args:
        token (str): Refresh token
        response (Response): response to set cookie on
        max_age (Optional[int], optional): Max Age cookie paramater. Defaults to None.
    """
    self._set_cookies(
        token=token, type="refresh", response=response, max_age=max_age
    )

unset_access_cookies

unset_access_cookies(response)

Remove 'Set-Cookie' for access token in response header

PARAMETER	DESCRIPTION
response	TYPE: Response

PARAMETER	DESCRIPTION
response	response to remove cooke from TYPE: Response

Source code in authx/main.py

def unset_access_cookies(
    self,
    response: Response,
) -> None:
    """Remove 'Set-Cookie' for access token in response header

    Args:
        response (Response): response to remove cooke from
    """
    self._unset_cookies("access", response=response)

unset_refresh_cookies

unset_refresh_cookies(response)

Remove 'Set-Cookie' for refresh token in response header

PARAMETER	DESCRIPTION
response	TYPE: Response

PARAMETER	DESCRIPTION
response	response to remove cooke from TYPE: Response

Source code in authx/main.py

def unset_refresh_cookies(
    self,
    response: Response,
) -> None:
    """Remove 'Set-Cookie' for refresh token in response header

    Args:
        response (Response): response to remove cooke from
    """
    self._unset_cookies("refresh", response=response)

unset_cookies

unset_cookies(response)

Remove 'Set-Cookie' for tokens from response headers

PARAMETER	DESCRIPTION
response	TYPE: Response

PARAMETER	DESCRIPTION
response	response to remove token cookies from TYPE: Response

Source code in authx/main.py

def unset_cookies(self, response: Response) -> None:
    """Remove 'Set-Cookie' for tokens from response headers

    Args:
        response (Response): response to remove token cookies from
    """
    self.unset_access_cookies(response)
    self.unset_refresh_cookies(response)

get_dependency

get_dependency(request, response)

FastAPI Dependency to return a AuthX sub-object within the route context

PARAMETER	DESCRIPTION
request	TYPE: Request
response	TYPE: Response

PARAMETER	DESCRIPTION
request	Request context managed by FastAPI TYPE: Request
response	Response context managed by FastAPI TYPE: Response

Note

The AuthXDeps is a utility class, to enable quick token operations within the route logic. It provides methods to avoid addtional code in your route that would be outside of the route logic

Such methods includes setting and unsetting cookies without the need to generate a response object beforhand

RETURNS	DESCRIPTION
AuthXDeps	The contextful AuthX object TYPE: AuthXDependency

Source code in authx/main.py

def get_dependency(self, request: Request, response: Response) -> AuthXDependency:
    """FastAPI Dependency to return a AuthX sub-object within the route context

    Args:
        request (Request): Request context managed by FastAPI
        response (Response): Response context managed by FastAPI

    Note:
        The AuthXDeps is a utility class, to enable quick token operations
        within the route logic. It provides methods to avoid addtional code
        in your route that would be outside of the route logic

        Such methods includes setting and unsetting cookies without the need
        to generate a response object beforhand

    Returns:
        AuthXDeps: The contextful AuthX object
    """
    return AuthXDependency(self, request=request, response=response)

token_required

token_required(type='access', verify_type=True, verify_fresh=False, verify_csrf=None, locations=None)

Dependency to enforce valid token availability in request

PARAMETER	DESCRIPTION
type	TYPE: str DEFAULT: 'access'
verify_type	TYPE: bool DEFAULT: True
verify_fresh	TYPE: bool DEFAULT: False
verify_csrf	TYPE: Optional[bool] DEFAULT: None
locations	TYPE: Optional[TokenLocations] DEFAULT: None

PARAMETER	DESCRIPTION
type	Require a given token type. Defaults to "access". TYPE: str DEFAULT: 'access'
verify_type	Apply type verification. Defaults to True. TYPE: bool DEFAULT: True
verify_fresh	Require token freshness. Defaults to False. TYPE: bool DEFAULT: False
verify_csrf	Enable CSRF verification. Defaults to None. TYPE: Optional[bool] DEFAULT: None
locations	Locations to retrieve token from. Defaults to None. TYPE: Optional[TokenLocations] DEFAULT: None

RETURNS	DESCRIPTION
Callable[[Request], TokenPayload]	Callable[[Request], TokenPayload]: Dependency for Valid token Payload retrieval

Source code in authx/main.py

def token_required(
    self,
    type: str = "access",
    verify_type: bool = True,
    verify_fresh: bool = False,
    verify_csrf: Optional[bool] = None,
    locations: Optional[TokenLocations] = None,
) -> Callable[[Request], TokenPayload]:
    """Dependency to enforce valid token availability in request

    Args:
        type (str, optional): Require a given token type. Defaults to "access".
        verify_type (bool, optional): Apply type verification. Defaults to True.
        verify_fresh (bool, optional): Require token freshness. Defaults to False.
        verify_csrf (Optional[bool], optional): Enable CSRF verification. Defaults to None.
        locations (Optional[TokenLocations], optional): Locations to retrieve token from. Defaults to None.

    Returns:
        Callable[[Request], TokenPayload]: Dependency for Valid token Payload retrieval
    """

    async def _auth_required(request: Request):
        return await self._auth_required(
            request=request,
            type=type,
            verify_csrf=verify_csrf,
            verify_type=verify_type,
            verify_fresh=verify_fresh,
            locations=locations,
        )

    return _auth_required

get_current_subject async

get_current_subject(request)

PARAMETER	DESCRIPTION
request	TYPE: Request

Source code in authx/main.py

async def get_current_subject(self, request: Request) -> Optional[T]:
    token: TokenPayload = await self._auth_required(request=request)
    uid = token.sub
    return self._get_current_subject(uid=uid)

get_token_from_request

get_token_from_request(type='access', optional=True)

Return token from response if available

PARAMETER	DESCRIPTION
type	TYPE: TokenType DEFAULT: 'access'
optional	TYPE: bool DEFAULT: True

PARAMETER	DESCRIPTION
type	The type of token to retrieve from request. Defaults to "access". TYPE: TokenType DEFAULT: 'access'
optional	Whether or not to enforce token presence in request. Defaults to True. TYPE: bool DEFAULT: True

Note

When optional=True, the return value might be None if no token is available in request

When optional=False, raises a MissingTokenError

RETURNS	DESCRIPTION
Optional[RequestToken]	Optional[RequestToken]: The RequestToken if available

Source code in authx/main.py

def get_token_from_request(
    self, type: TokenType = "access", optional: bool = True
) -> Optional[RequestToken]:
    """Return token from response if available

    Args:
        type (TokenType, optional): The type of token to retrieve from request.
            Defaults to "access".
        optional (bool, optional): Whether or not to enforce token presence in request.
            Defaults to True.

    Note:
        When `optional=True`, the return value might be `None`
        if no token is available in request

        When `optional=False`, raises a MissingTokenError

    Returns:
        Optional[RequestToken]: The RequestToken if available
    """

    async def _token_getter(request: Request):
        return await self._get_token_from_request(
            request, optional=optional, refresh=(type == "refresh")
        )

    return _token_getter

implicit_refresh_middleware async

implicit_refresh_middleware(request, call_next)

FastAPI Middleware to enable token refresh for an APIRouter

PARAMETER	DESCRIPTION
request	TYPE: Request
call_next	TYPE: Coroutine

PARAMETER	DESCRIPTION
request	Incoming request TYPE: Request
call_next	Endpoint logic to be called TYPE: Coroutine

Note

This middleware is only based on access tokens. Using implicit refresh mechanism makes use of refresh tokens unnecessary.

Note

The refreshed access token will not be considered as fresh

Note

The implicit refresh mechanism is only enabled for authorization through cookies.

RETURNS	DESCRIPTION
Response	Response with update access token cookie if relevant TYPE: Response

Source code in authx/main.py

async def implicit_refresh_middleware(
    self, request: Request, call_next: Coroutine
) -> Response:
    """FastAPI Middleware to enable token refresh for an APIRouter

    Args:
        request (Request): Incoming request
        call_next (Coroutine): Endpoint logic to be called

    Note:
        This middleware is only based on `access` tokens.
        Using implicit refresh mechanism makes use of `refresh`
        tokens unnecessary.

    Note:
        The refreshed `access` token will not be considered as
        `fresh`

    Note:
        The implicit refresh mechanism is only enabled
        for authorization through cookies.

    Returns:
        Response: Response with update access token cookie if relevant
    """
    response = await call_next(request)

    request_condition = self.config.has_location(
        "cookies"
    ) and self._implicit_refresh_enabled_for_request(request)

    if request_condition:
        with contextlib.suppress(AuthXException):
            # Refresh mechanism
            token = await self._get_token_from_request(
                request=request,
                locations=["cookies"],
                refresh=False,
                optional=False,
            )
            payload = self.verify_token(token, verify_fresh=False)
            if (
                payload.time_until_expiry
                < self.config.JWT_IMPLICIT_REFRESH_DELTATIME
            ):
                new_token = self.create_access_token(
                    uid=payload.sub, fresh=False, data=payload.extra_dict
                )
                self.set_access_cookies(new_token, response=response)
    return response