Skip to main content

authn

Authentication manager for Datalayer Core.

class AuthenticationManager(iam_url: str, storage: Optional[datalayer_core.authn.storage.TokenStorage] = None)​

Authentication Manager for Datalayer Core.

Provides a unified interface for all authentication methods using the strategy pattern.

Examples​

auth = AuthenticationManager("https://prod1.datalayer.run")
user, token = await auth.login(token="abc123")
user, token = await auth.login(handle="user", password="pass")
await auth.logout()

__init__(iam_url: str, storage: Optional[datalayer_core.authn.storage.TokenStorage] = None)​

Initialize AuthenticationManager.

Args: iam_url: IAM service URL (e.g., "https://prod1.datalayer.run") storage: Optional token storage backend. Defaults to auto-detected storage.

login(token: Optional[str] = None, handle: Optional[str] = None, password: Optional[str] = None, use_browser: bool = False, no_store: bool = False) -> Tuple[Dict[str, Any], str]​

Login using various authentication methods.

Automatically selects the appropriate strategy based on provided options.

Parameters​

  • token : str, optional

    Existing authentication token.

  • handle : str, optional

    User handle for credentials auth.

  • password : str, optional

    Password for credentials auth.

  • use_browser : bool, default False

    Use browser-based OAuth flow.

  • no_store : bool, default False

    Don't store the token after authentication.

Returns​

  • tuple of (dict, str)

    Tuple of (user_dict, token_string).

Raises​

  • ValueError

    If no suitable authentication strategy found

  • Exception

    If authentication fails

Examples​

Token auth:

user, token = await auth.login(token="abc123")

Credentials auth:

user, token = await auth.login(handle="user", password="pass")

Storage auth (uses stored token):

user, token = await auth.login()

logout() -> None​

Logout the current user.

Calls the logout API and clears stored tokens.

whoami() -> Optional[Dict[str, Any]]​

Get the current user profile.

Uses cached user if available, otherwise fetches from API using cached token or stored token.

Returns​

  • dict or None

    User dictionary or None if not authenticated.

validate_token(token: str) -> Dict[str, Any]​

Validate a token.

Checks if a token is valid by attempting to get the user profile.

Parameters​

  • token : str

    Token to validate.

Returns​

  • dict

    Dictionary with 'valid' key and optionally 'user' or 'error' keys.

Examples​

result = await auth.validate_token("abc123")
if result['valid']:
    print(f"Token is valid for user: {result['user']}")

get_stored_token() -> Optional[str]​

Get the stored token from storage.

Returns​

  • str or None

    Stored token or None.

store_token(token: str) -> None​

Store a token in storage.

Parameters​

  • token : str

    Token to store.

clear_stored_token() -> None​

Clear the stored token.

get_current_user() -> Optional[Dict[str, Any]]​

Get the current cached user.

Returns​

  • dict or None

    User dictionary or None.

get_current_token() -> Optional[str]​

Get the current token.

Returns​

  • str or None

    Current token or None.

is_authenticated() -> bool​

Check if user is currently authenticated.

Returns​

  • bool

    True if user is authenticated.