Skip to main content

authentication_flow

Authentication flow and session management.

Classes​

AuthEnvironmentError​

class AuthEnvironmentError(*args, **kwargs):

Exception related to the authorization and authentication environment.

Ancestors​

BitfountSession​

class BitfountSession(authentication_handler: AuthenticationHandler | None = None):

Manages session-based interactions with Bitfount.

Extends requests.Session, appending an access token to the authorization of any requests made if an access token is present

When the token expires it will request a new token prior to sending the web request.

Ancestors​

Variables​

  • am_request_headers : dict[typing.Any, typing.Any] - Returns metadata for authenticating with message service.
  • authenticated : bool - Returns true if we have an unexpired access token or API Keys.
  • hub_request_headers : dict[typing.Any, typing.Any] - Returns metadata for authenticating with message service.
  • message_service_metadata : list[tuple[str, str]] - Returns metadata for authenticating with message service.

  • message_service_metadata_if_fresh : list[tuple[str, str]] | None - Message service metadata, if it can be had without blocking.

    message_service_metadata takes the reauthentication lock and may authenticate inline, either of which can block for minutes. Callers on an event loop use this first and fall back to resolving the blocking property off the loop, so that the common case - a token the refresh thread has already kept current - costs nothing.

    Returns: The current metadata, or None if the token needs refreshing or a refresh is already in progress.

  • username : str - Returns the username of the authenticated user.

Methods​

authenticate​

def authenticate(self) ‑> None:

Authenticates user to allow protected requests.

Prompts the user to login/authenticate and stores the tokens to use them in future requests.

Raises

  • AssertionError: If user storage path corresponds to a different username from the BitfountSession.
  • ConnectionError: If a token cannot be retrieved.

request​

def request(    self,    method: str | bytes,    url: str,    params: MutableMapping[str, Any] | None = None,    data: Any = None,    headers: MutableMapping[str, str] | None = None,    **kwargs: Any,) ‑> Response:

Performs an HTTP request.

Overrides requests.session.request, appending our access token to the request headers or API keys if present.