from fastauth.transport import TokenTransport
Token transport
The token transport is class which is responsible for creating login and logout responses, and token schema validation.
It takes FastAuthConfig.TOKEN_LOCATIONS
argument and try to get token from provided locations.
class TokenTransport(ABC):
def __init__(self, config: FastAuthConfig):
self._config = config
@abstractmethod
def schema(self, request: Request, refresh: bool = False) -> type[SecurityBase]:
"""
Return the security schema for the token transport
:param request: FastAPI request
:param refresh: Flag if the token is a refresh type
:return: SecurityBase instance
"""
raise NotImplementedError
@abstractmethod
async def login_response(
self,
security: "FastAuth",
content: TokenResponse,
response: Response | None = None,
) -> Response:
"""
Generate the login response
:param security: FastAuth instance
:param content: TokenResponse instance
:param response: FastAPI Response instance
:return: FastAPI Response instance
"""
raise NotImplementedError
@abstractmethod
async def logout_response(
self, security: "FastAuth", response: Response | None = None
) -> Response:
"""
Generate the logout response
:param security: FastAuth instance
:param response: FastAPI Response instance
:return: FastAPI Response instance
"""
raise NotImplementedError
FastAuth provide 4 token locations:
- headers - BearerTransport
- cookies - CookieTransport
- query - QueryTransport
- body - BodyTransport
This class stored in TRANSPORT_GETTER dict in fastauth.transport
module.
To get token from response it user fastauth.transport.get_token_from_request
function, which iterate through provided SecurityBase
classes.
If at least one token found it return token string, else it raises MissingToken
exception.
TRANSPORT_GETTER = {
"headers": BearerTransport,
"cookies": CookieTransport,
}
def get_token_from_request(
config: FastAuthConfig,
request: Request | None = None,
refresh: bool = False,
locations: list[str] | None = None,
) -> Callable[..., Coroutine[Any, Any, str]]:
"""
Get token from request using the token transport locations specified in the FastAuthConfig.TOKEN_LOCATIONS.
If the token is not found in any of the locations, raise a MissingToken exception.
Because used FastAPI SecuredBase, in Transport.scheme() method, we need to return callable to resolve the dependency later.
:param config: FastAuthConfig
:param request: FastAPI Request
:param refresh: flag to set refresh token type
:param locations: pass locations to get token from or default will be used
:return: Callable with coroutine to pass to FastAPI Depends
"""
if locations is None:
locations = config.TOKEN_LOCATIONS
parameters: list[Parameter] = []
for location in locations:
transport = TRANSPORT_GETTER[location]
parameters.append(
Parameter(
name=location,
kind=Parameter.POSITIONAL_OR_KEYWORD,
default=Depends(transport(config).schema(request, refresh)),
)
)
@with_signature(Signature(parameters))
async def _token_locations(**kwargs) -> str:
errors: list[exceptions.MissingToken] = []
for location_name, token in kwargs.items():
if token is not None:
return token
errors.append(
exceptions.MissingToken(
msg=f"Missing token in {location_name}: Not authenticated"
)
)
if errors:
raise exceptions.MissingToken(msg=[err.detail for err in errors])
msg = f"No token found in request from {locations}"
raise exceptions.MissingToken(msg)
return _token_locations
To get login or logout response user get_login_response
and get_logout_response
from fastauth.transport
module respectively.
async def get_login_response(
security: "FastAuth", tokens: TokenResponse, response: Response | None = None
):
"""
Get login response from the token locations specified in the FastAuthConfig.TOKEN_LOCATIONS.
:param security: FastAuth instance
:param tokens: TokenResponse instance with access and/or refresh tokens
:param response: Optional FastAPI Response to modify to
:return: FastAPI Response with multiple token locations(eg. headers, cookies) set
"""
for location in security.config.TOKEN_LOCATIONS:
transport_callable = TRANSPORT_GETTER[location]
transport: TokenTransport = transport_callable(security.config)
response = await transport.login_response(
security,
tokens,
response,
)
return response
async def get_logout_response(security: "FastAuth", response: Response | None = None):
"""
Get logout response from the token locations specified in the FastAuthConfig.TOKEN_LOCATIONS.
:param security: FastAuth instance
:param response: Optional FastAPI Response to modify to
:return: FastAPI Response with unset action for multiple token locations
"""
for location in security.config.TOKEN_LOCATIONS:
transport_callable = TRANSPORT_GETTER[location]
transport: TokenTransport = transport_callable(security.config)
response = await transport.logout_response(
security,
response,
)
return response
If you want to customize or create own Transport, just implement fastauth.transport.base.TokenTransport
class, and overrides TRANSPORT_GETTER
dict
from fastauth.transport.base import TokenTransport
from fastauth import transport
class MyTransport(TokenTransport):
...
transport.TRANSPORT_GETTER["headers"] = MyTransport
If you use custom Transport name, don`t forget to set it in FastAuthConfig.TOKEN_LOCATIONS list.