Skip to content

Authx#

authx.main.AuthX #

AuthX(config=AuthXConfig(), model=None)

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: None

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: None

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: None

Source code in authx/main.py
def __init__(
    self, config: AuthXConfig = AuthXConfig(), model: Optional[T] = None
) -> 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].
    """
    self.model: Union[T, Dict[str, Any]] = model if model is not None else {}
    super().__init__(model=model)
    super(_CallbackHandler, self).__init__()
    self._config = config

MSG_TokenError class-attribute instance-attribute #

MSG_TokenError = 'Token Error'

MSG_MissingTokenError class-attribute instance-attribute #

MSG_MissingTokenError = 'Missing JWT in request'

MSG_MissingCSRFTokenError class-attribute instance-attribute #

MSG_MissingCSRFTokenError = 'Missing CSRF double submit token in request'

MSG_TokenTypeError class-attribute instance-attribute #

MSG_TokenTypeError = 'Bad token type'

MSG_RevokedTokenError class-attribute instance-attribute #

MSG_RevokedTokenError = 'Invalid token'

MSG_TokenRequiredError class-attribute instance-attribute #

MSG_TokenRequiredError = 'Token required'

MSG_FreshTokenRequiredError class-attribute instance-attribute #

MSG_FreshTokenRequiredError = 'Fresh token required'

MSG_AccessTokenRequiredError class-attribute instance-attribute #

MSG_AccessTokenRequiredError = 'Access token required'

MSG_RefreshTokenRequiredError class-attribute instance-attribute #

MSG_RefreshTokenRequiredError = 'Refresh token required'

MSG_CSRFError class-attribute instance-attribute #

MSG_CSRFError = 'CSRF double submit does not match'

MSG_JWTDecodeError class-attribute instance-attribute #

MSG_JWTDecodeError = '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

model instance-attribute #

model = model if model is not None else {}

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_TokenError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.MissingCSRFTokenError,
        status_code=401,
        message=self.MSG_MissingCSRFTokenError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.TokenTypeError,
        status_code=401,
        message=self.MSG_TokenTypeError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.RevokedTokenError,
        status_code=401,
        message=self.MSG_RevokedTokenError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.TokenRequiredError,
        status_code=401,
        message=self.MSG_TokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.FreshTokenRequiredError,
        status_code=401,
        message=self.MSG_FreshTokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.AccessTokenRequiredError,
        status_code=401,
        message=self.MSG_AccessTokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.RefreshTokenRequiredError,
        status_code=401,
        message=self.MSG_RefreshTokenRequiredError,
    )
    self._set_app_exception_handler(
        app,
        exception=exceptions.CSRFError,
        status_code=401,
        message=self.MSG_CSRFError,
    )

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: Optional[str]

**kwargs

TYPE: ParamSpecKwargs DEFAULT: {}

Source code in authx/_internal/_callback.py
def is_token_in_blocklist(
    self, token: Optional[str], **kwargs: ParamSpecKwargs
) -> bool:
    """Check if token is in blocklist"""
    if self._check_token_callback_is_set(ignore_errors=True):
        callback: Optional[TokenCallback] = self.callback_is_token_in_blocklist
        if callback is not None and token is not None:
            return callback(token, **kwargs)  # type: ignore
    return False

load_config #

load_config(config)

Loads a AuthXConfig as the new configuration b Args: config (AuthXConfig): Configuration to load

PARAMETER DESCRIPTION
config

TYPE: AuthXConfig

Source code in authx/main.py
def load_config(self, config: AuthXConfig) -> None:
    """Loads a AuthXConfig as the new configuration
    b
            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,
        audience=self.config.JWT_DECODE_AUDIENCE,
        issuer=self.config.JWT_DECODE_ISSUER,
    )

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

TYPE: Any DEFAULT: ()

**kwargs

TYPE: Any 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: Any,
    **kwargs: Any,
) -> 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

TYPE: Any DEFAULT: ()

**kwargs

TYPE: Any 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: Any,
    **kwargs: Any,
) -> 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 parameter. 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 parameter. 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 parameter. 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 parameter. 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 additional 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 beforehand

RETURNS DESCRIPTION
AuthXDeps

The contextful AuthX object

TYPE: AuthXDependency[Any]

Source code in authx/main.py
def get_dependency(
    self, request: Request, response: Response
) -> AuthXDependency[Any]:
    """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 additional 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 beforehand

    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], Awaitable[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], Awaitable[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) -> Any:
        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
Callable[[Request], Awaitable[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
) -> Callable[[Request], Awaitable[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) -> Optional[RequestToken]:
        # Specify locations as None since it's not provided in this context
        # Convert `optional` to a literal True or False based on its value
        optional_literal: Literal[True, False] = optional
        return await self._get_token_from_request(
            request,
            None,  # Explicitly passing None for locations
            refresh=(type == "refresh"),
            optional=optional_literal,
        )

    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: Callable[[Request], Coroutine[Any, Any, Response]]

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: Callable[[Request], Coroutine[Any, Any, Response]],
) -> 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)

    if self.config.has_location(
        "cookies"
    ) and self._implicit_refresh_enabled_for_request(request):
        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 (
                datetime.timedelta(payload.time_until_expiry)  # type: ignore
                < 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