Core API Reference

Core components available directly from fabias.

Authentication

fabias.ServicePrincipalAuth

Bases: AuthProvider

OAuth2 client credentials authentication using service principal.

This provider authenticates using Azure AD client credentials flow, suitable for service-to-service authentication without user interaction.

Tokens are cached and automatically refreshed when expired.

Attributes:

Name	Type	Description
tenant_id	str	Azure AD tenant ID
client_id	str	Application (client) ID
client_secret	str	Client secret value

Examples:

>>> auth = ServicePrincipalAuth(
...     tenant_id="your-tenant-id",
...     client_id="your-client-id",
...     client_secret="your-secret"
... )
>>> token = auth.getToken("https://api.fabric.microsoft.com/.default")

Source code in src/fabias/_shared/auth.py

class ServicePrincipalAuth(AuthProvider):
    """
    OAuth2 client credentials authentication using service principal.

    This provider authenticates using Azure AD client credentials flow,
    suitable for service-to-service authentication without user interaction.

    Tokens are cached and automatically refreshed when expired.

    Attributes:
        tenant_id (str): Azure AD tenant ID
        client_id (str): Application (client) ID
        client_secret (str): Client secret value

    Examples:
        >>> auth = ServicePrincipalAuth(
        ...     tenant_id="your-tenant-id",
        ...     client_id="your-client-id",
        ...     client_secret="your-secret"
        ... )
        >>> token = auth.getToken("https://api.fabric.microsoft.com/.default")
    """

    def __init__(self, tenant_id: str, client_id: str, client_secret: str):
        """
        Initialize the service principal authentication provider.

        Args:
            tenant_id: Azure AD tenant ID (GUID or domain)
            client_id: Application (client) ID from Azure AD
            client_secret: Client secret value
        """
        self._tenant_id = tenant_id
        self._client_id = client_id
        self._client_secret = client_secret

        # Token cache: scope -> (token, expiry)
        self._token_cache: Dict[str, tuple] = {}

    def getToken(self, scope: str) -> str:
        """
        Get an access token for the specified scope.

        Retrieves a cached token if still valid, otherwise acquires a new
        token using the OAuth2 client credentials flow.

        Args:
            scope: OAuth2 scope (e.g., "https://api.fabric.microsoft.com/.default")

        Returns:
            str: Access token (without "Bearer " prefix)

        Raises:
            AuthenticationError: If token acquisition fails
        """
        # Check cache
        if scope in self._token_cache:
            token, expiry = self._token_cache[scope]
            # Add buffer of 60 seconds before expiry
            if expiry > datetime.now(timezone.utc) + timedelta(seconds=60):
                return token

        # Acquire new token
        token, expiry = self._acquire_token(scope)
        self._token_cache[scope] = (token, expiry)
        return token

    def _acquire_token(self, scope: str) -> Tuple[str, datetime]:
        """
        Acquire a new token from Azure AD.

        Args:
            scope: OAuth2 scope

        Returns:
            Tuple[str, datetime]: (access_token, expiry_datetime)

        Raises:
            AuthenticationError: If token acquisition fails
        """
        token_url = f"https://login.microsoftonline.com/{self._tenant_id}/oauth2/v2.0/token"

        headers = {"Content-Type": "application/x-www-form-urlencoded"}

        data = {
            "grant_type": "client_credentials",
            "scope": scope,
            "client_id": self._client_id,
            "client_secret": self._client_secret,
        }

        try:
            response = requests.post(token_url, data=data, headers=headers)

            if response.status_code != 200:
                error_data = (
                    response.json()
                    if response.headers.get("content-type", "").startswith("application/json")
                    else {}
                )
                error_msg = error_data.get(
                    "error_description", error_data.get("error", response.text)
                )
                raise AuthenticationError(f"Token acquisition failed: {error_msg}")

            token_data = response.json()
            access_token = token_data.get("access_token")
            expires_in: int = token_data.get("expires_in", 3600)

            if not access_token:
                raise AuthenticationError("No access token in response")

            expiry = datetime.now(timezone.utc) + timedelta(seconds=expires_in)

            return str(access_token), expiry

        except requests.RequestException as e:
            raise AuthenticationError("Network error during authentication", cause=e)

Functions

__init__(tenant_id, client_id, client_secret)

Initialize the service principal authentication provider.

Parameters:

Name	Type	Description	Default
tenant_id	str	Azure AD tenant ID (GUID or domain)	required
client_id	str	Application (client) ID from Azure AD	required
client_secret	str	Client secret value	required

Source code in src/fabias/_shared/auth.py

def __init__(self, tenant_id: str, client_id: str, client_secret: str):
    """
    Initialize the service principal authentication provider.

    Args:
        tenant_id: Azure AD tenant ID (GUID or domain)
        client_id: Application (client) ID from Azure AD
        client_secret: Client secret value
    """
    self._tenant_id = tenant_id
    self._client_id = client_id
    self._client_secret = client_secret

    # Token cache: scope -> (token, expiry)
    self._token_cache: Dict[str, tuple] = {}

getToken(scope)

Get an access token for the specified scope.

Retrieves a cached token if still valid, otherwise acquires a new token using the OAuth2 client credentials flow.

Parameters:

Name	Type	Description	Default
scope	str	OAuth2 scope (e.g., "https://api.fabric.microsoft.com/.default")	required

Returns:

Name	Type	Description
str	str	Access token (without "Bearer " prefix)

Raises:

Type	Description
AuthenticationError	If token acquisition fails

Source code in src/fabias/_shared/auth.py

def getToken(self, scope: str) -> str:
    """
    Get an access token for the specified scope.

    Retrieves a cached token if still valid, otherwise acquires a new
    token using the OAuth2 client credentials flow.

    Args:
        scope: OAuth2 scope (e.g., "https://api.fabric.microsoft.com/.default")

    Returns:
        str: Access token (without "Bearer " prefix)

    Raises:
        AuthenticationError: If token acquisition fails
    """
    # Check cache
    if scope in self._token_cache:
        token, expiry = self._token_cache[scope]
        # Add buffer of 60 seconds before expiry
        if expiry > datetime.now(timezone.utc) + timedelta(seconds=60):
            return token

    # Acquire new token
    token, expiry = self._acquire_token(scope)
    self._token_cache[scope] = (token, expiry)
    return token

fabias.RefreshTokenAuth

Bases: AuthProvider

User-delegated authentication using refresh tokens.

This provider supports two modes:

Self-managed mode: You provide a refresh_token, and the provider handles token rotation automatically. Use on_refresh callback to persist the new refresh token (e.g., to Key Vault).

Delegated mode: You provide a token_provider callable that handles all token acquisition externally (e.g., via an Azure Function token broker). The provider just calls your function.

Tokens are cached and automatically refreshed when expired.

Attributes:

Name	Type	Description
tenant_id	str	Azure AD tenant ID
client_id	str	Application (client) ID

Examples:

Self-managed with Key Vault persistence:

>>> from fabias.integrations import keyvault
>>> auth = RefreshTokenAuth(
...     tenant_id="your-tenant-id",
...     client_id="your-client-id",
...     refresh_token=keyvault.get("graph-refresh-token"),
...     on_refresh=lambda token: keyvault.set("graph-refresh-token", token)
... )
>>> token = auth.getToken("https://graph.microsoft.com/.default")

Delegated mode with external token broker:

>>> def get_from_broker(scope: str) -> tuple[str, datetime]:
...     response = requests.post("https://my-function/api/token", json={"scope": scope})
...     data = response.json()
...     return data["access_token"], datetime.fromisoformat(data["expires_at"])
...
>>> auth = RefreshTokenAuth(
...     tenant_id="your-tenant-id",
...     client_id="your-client-id",
...     token_provider=get_from_broker
... )

Warning

Self-managed mode is NOT safe for concurrent use across multiple jobs that might refresh the token simultaneously. For production multi-job scenarios, use delegated mode with a token broker service.

Source code in src/fabias/_shared/auth.py

class RefreshTokenAuth(AuthProvider):
    """
    User-delegated authentication using refresh tokens.

    This provider supports two modes:

    **Self-managed mode**: You provide a refresh_token, and the provider
    handles token rotation automatically. Use `on_refresh` callback to
    persist the new refresh token (e.g., to Key Vault).

    **Delegated mode**: You provide a `token_provider` callable that
    handles all token acquisition externally (e.g., via an Azure Function
    token broker). The provider just calls your function.

    Tokens are cached and automatically refreshed when expired.

    Attributes:
        tenant_id (str): Azure AD tenant ID
        client_id (str): Application (client) ID

    Examples:
        Self-managed with Key Vault persistence:

        >>> from fabias.integrations import keyvault
        >>> auth = RefreshTokenAuth(
        ...     tenant_id="your-tenant-id",
        ...     client_id="your-client-id",
        ...     refresh_token=keyvault.get("graph-refresh-token"),
        ...     on_refresh=lambda token: keyvault.set("graph-refresh-token", token)
        ... )
        >>> token = auth.getToken("https://graph.microsoft.com/.default")

        Delegated mode with external token broker:

        >>> def get_from_broker(scope: str) -> tuple[str, datetime]:
        ...     response = requests.post("https://my-function/api/token", json={"scope": scope})
        ...     data = response.json()
        ...     return data["access_token"], datetime.fromisoformat(data["expires_at"])
        ...
        >>> auth = RefreshTokenAuth(
        ...     tenant_id="your-tenant-id",
        ...     client_id="your-client-id",
        ...     token_provider=get_from_broker
        ... )

    Warning:
        Self-managed mode is NOT safe for concurrent use across multiple jobs
        that might refresh the token simultaneously. For production multi-job
        scenarios, use delegated mode with a token broker service.
    """

    def __init__(
        self,
        tenant_id: str,
        client_id: str,
        refresh_token: Optional[str] = None,
        on_refresh: Optional["Callable[[str], None]"] = None,
        token_provider: Optional["Callable[[str], Tuple[str, datetime]]"] = None,
    ):
        """
        Initialize the refresh token authentication provider.

        Args:
            tenant_id: Azure AD tenant ID (GUID or domain)
            client_id: Application (client) ID from Azure AD
            refresh_token: Initial refresh token (for self-managed mode)
            on_refresh: Callback to persist new refresh token (for self-managed mode).
                        Called with the new refresh_token string after each refresh.
            token_provider: External function that returns (access_token, expiry)
                           for a given scope (for delegated mode). If provided,
                           refresh_token and on_refresh are ignored.

        Raises:
            ValueError: If neither refresh_token nor token_provider is provided
        """
        if not refresh_token and not token_provider:
            raise ValueError(
                "Either refresh_token (self-managed) or token_provider (delegated) is required"
            )

        self._tenant_id = tenant_id
        self._client_id = client_id
        self._refresh_token = refresh_token
        self._on_refresh = on_refresh
        self._token_provider = token_provider

        # Token cache: scope -> (token, expiry)
        self._token_cache: Dict[str, tuple] = {}

    @property
    def refresh_token(self) -> Optional[str]:
        """Get the current refresh token (may have been rotated)."""
        return self._refresh_token

    def getToken(self, scope: str) -> str:
        """
        Get an access token for the specified scope.

        Retrieves a cached token if still valid, otherwise acquires a new
        token using either the token_provider or refresh token flow.

        Args:
            scope: OAuth2 scope (e.g., "https://graph.microsoft.com/.default")

        Returns:
            str: Access token (without "Bearer " prefix)

        Raises:
            AuthenticationError: If token acquisition fails
        """
        # Check cache
        if scope in self._token_cache:
            token, expiry = self._token_cache[scope]
            # Add buffer of 60 seconds before expiry
            if expiry > datetime.now(timezone.utc) + timedelta(seconds=60):
                return token

        # Acquire new token
        token, expiry = self._acquire_token(scope)
        self._token_cache[scope] = (token, expiry)
        return token

    def _acquire_token(self, scope: str) -> Tuple[str, datetime]:
        """
        Acquire a new access token.

        Uses token_provider if available, otherwise uses refresh token flow.

        Args:
            scope: OAuth2 scope

        Returns:
            Tuple[str, datetime]: (access_token, expiry_datetime)

        Raises:
            AuthenticationError: If token acquisition fails
        """
        # Delegated mode - use external provider
        if self._token_provider:
            try:
                return self._token_provider(scope)
            except Exception as e:
                raise AuthenticationError(f"Token provider failed: {str(e)}", cause=e)

        # Self-managed mode - use refresh token
        return self._do_refresh(scope)

    def _do_refresh(self, scope: str) -> Tuple[str, datetime]:
        """
        Exchange refresh token for new access token.

        Also handles token rotation - the new refresh token is passed
        to the on_refresh callback if provided.

        Args:
            scope: OAuth2 scope

        Returns:
            Tuple[str, datetime]: (access_token, expiry_datetime)

        Raises:
            AuthenticationError: If refresh fails
        """
        if not self._refresh_token:
            raise AuthenticationError("No refresh token available")

        token_url = f"https://login.microsoftonline.com/{self._tenant_id}/oauth2/v2.0/token"

        headers = {"Content-Type": "application/x-www-form-urlencoded"}

        data = {
            "grant_type": "refresh_token",
            "scope": scope,
            "client_id": self._client_id,
            "refresh_token": self._refresh_token,
        }

        try:
            response = requests.post(token_url, data=data, headers=headers)

            if response.status_code != 200:
                error_data = (
                    response.json()
                    if response.headers.get("content-type", "").startswith("application/json")
                    else {}
                )
                error = error_data.get("error", "unknown")
                error_msg = error_data.get(
                    "error_description", error_data.get("error", response.text)
                )

                # Check for invalid_grant - token may have been rotated by another process
                if error == "invalid_grant":
                    raise AuthenticationError(
                        f"Refresh token invalid or expired. {error_msg}. "
                        "The token may have been rotated by another process, "
                        "or it has expired (90 day limit)."
                    )

                raise AuthenticationError(f"Token refresh failed: {error_msg}")

            token_data = response.json()
            access_token = token_data.get("access_token")
            expires_in: int = token_data.get("expires_in", 3600)
            new_refresh_token = token_data.get("refresh_token")

            if not access_token:
                raise AuthenticationError("No access token in response")

            # Update stored refresh token (it rotates with each use)
            if new_refresh_token:
                self._refresh_token = new_refresh_token

                # Call persistence callback if provided
                if self._on_refresh:
                    try:
                        self._on_refresh(new_refresh_token)
                    except Exception as e:
                        # Log but don't fail - the token worked, persistence is secondary
                        import logging

                        logging.warning(f"Failed to persist refresh token: {e}")

            expiry = datetime.now(timezone.utc) + timedelta(seconds=expires_in)

            return str(access_token), expiry

        except requests.RequestException as e:
            raise AuthenticationError("Network error during token refresh", cause=e)

Attributes

refresh_token property

Get the current refresh token (may have been rotated).

Functions

__init__(tenant_id, client_id, refresh_token=None, on_refresh=None, token_provider=None)

Initialize the refresh token authentication provider.

Parameters:

Name	Type	Description	Default
tenant_id	str	Azure AD tenant ID (GUID or domain)	required
client_id	str	Application (client) ID from Azure AD	required
refresh_token	Optional[str]	Initial refresh token (for self-managed mode)	None
on_refresh	Optional[Callable[[str], None]]	Callback to persist new refresh token (for self-managed mode). Called with the new refresh_token string after each refresh.	None
token_provider	Optional[Callable[[str], Tuple[str, datetime]]]	External function that returns (access_token, expiry) for a given scope (for delegated mode). If provided, refresh_token and on_refresh are ignored.	None

Raises:

Type	Description
ValueError	If neither refresh_token nor token_provider is provided

Source code in src/fabias/_shared/auth.py

def __init__(
    self,
    tenant_id: str,
    client_id: str,
    refresh_token: Optional[str] = None,
    on_refresh: Optional["Callable[[str], None]"] = None,
    token_provider: Optional["Callable[[str], Tuple[str, datetime]]"] = None,
):
    """
    Initialize the refresh token authentication provider.

    Args:
        tenant_id: Azure AD tenant ID (GUID or domain)
        client_id: Application (client) ID from Azure AD
        refresh_token: Initial refresh token (for self-managed mode)
        on_refresh: Callback to persist new refresh token (for self-managed mode).
                    Called with the new refresh_token string after each refresh.
        token_provider: External function that returns (access_token, expiry)
                       for a given scope (for delegated mode). If provided,
                       refresh_token and on_refresh are ignored.

    Raises:
        ValueError: If neither refresh_token nor token_provider is provided
    """
    if not refresh_token and not token_provider:
        raise ValueError(
            "Either refresh_token (self-managed) or token_provider (delegated) is required"
        )

    self._tenant_id = tenant_id
    self._client_id = client_id
    self._refresh_token = refresh_token
    self._on_refresh = on_refresh
    self._token_provider = token_provider

    # Token cache: scope -> (token, expiry)
    self._token_cache: Dict[str, tuple] = {}

getToken(scope)

Get an access token for the specified scope.

Retrieves a cached token if still valid, otherwise acquires a new token using either the token_provider or refresh token flow.

Parameters:

Name	Type	Description	Default
scope	str	OAuth2 scope (e.g., "https://graph.microsoft.com/.default")	required

Returns:

Name	Type	Description
str	str	Access token (without "Bearer " prefix)

Raises:

Type	Description
AuthenticationError	If token acquisition fails

Source code in src/fabias/_shared/auth.py

def getToken(self, scope: str) -> str:
    """
    Get an access token for the specified scope.

    Retrieves a cached token if still valid, otherwise acquires a new
    token using either the token_provider or refresh token flow.

    Args:
        scope: OAuth2 scope (e.g., "https://graph.microsoft.com/.default")

    Returns:
        str: Access token (without "Bearer " prefix)

    Raises:
        AuthenticationError: If token acquisition fails
    """
    # Check cache
    if scope in self._token_cache:
        token, expiry = self._token_cache[scope]
        # Add buffer of 60 seconds before expiry
        if expiry > datetime.now(timezone.utc) + timedelta(seconds=60):
            return token

    # Acquire new token
    token, expiry = self._acquire_token(scope)
    self._token_cache[scope] = (token, expiry)
    return token

fabias.FabricAuth

Bases: AuthProvider

Authentication provider using Fabric's notebookutils.

This provider leverages the notebookutils.credentials module available in Microsoft Fabric notebooks to obtain tokens without requiring explicit credentials.

Only works when running inside a Fabric notebook environment.

Examples:

>>> from fabias import runtime
>>> if runtime.isFabric:
...     auth = FabricAuth()
...     token = auth.getToken("https://api.fabric.microsoft.com/.default")

Source code in src/fabias/_shared/auth.py

class FabricAuth(AuthProvider):
    """
    Authentication provider using Fabric's notebookutils.

    This provider leverages the notebookutils.credentials module available
    in Microsoft Fabric notebooks to obtain tokens without requiring
    explicit credentials.

    Only works when running inside a Fabric notebook environment.

    Examples:
        >>> from fabias import runtime
        >>> if runtime.isFabric:
        ...     auth = FabricAuth()
        ...     token = auth.getToken("https://api.fabric.microsoft.com/.default")
    """

    def __init__(self):
        """
        Initialize the Fabric authentication provider.

        Raises:
            AuthenticationError: If not running inside Fabric environment
        """
        if not runtime.isFabric:
            raise AuthenticationError(
                "FabricAuth requires Fabric notebook environment. "
                "Use ServicePrincipalAuth for standalone execution."
            )

        self._notebookutils = runtime.notebookutils

    def getToken(self, scope: str) -> str:
        """
        Get an access token using notebookutils.

        Args:
            scope: OAuth2 scope (e.g., "https://api.fabric.microsoft.com/.default")

        Returns:
            str: Access token

        Raises:
            AuthenticationError: If token acquisition fails
        """
        try:
            # notebookutils.credentials.getToken expects a resource URL
            # Convert scope to resource if needed (remove /.default suffix)
            resource = scope.replace("/.default", "")

            token = self._notebookutils.credentials.getToken(resource)

            if not token:
                raise AuthenticationError(f"No token returned for scope: {scope}")

            return cast(str, token)

        except AttributeError as e:
            raise AuthenticationError("notebookutils.credentials.getToken not available", cause=e)
        except Exception as e:
            raise AuthenticationError(f"Failed to get token from notebookutils: {str(e)}", cause=e)

Functions

__init__()

Initialize the Fabric authentication provider.

Raises:

Type	Description
AuthenticationError	If not running inside Fabric environment

Source code in src/fabias/_shared/auth.py

def __init__(self):
    """
    Initialize the Fabric authentication provider.

    Raises:
        AuthenticationError: If not running inside Fabric environment
    """
    if not runtime.isFabric:
        raise AuthenticationError(
            "FabricAuth requires Fabric notebook environment. "
            "Use ServicePrincipalAuth for standalone execution."
        )

    self._notebookutils = runtime.notebookutils

getToken(scope)

Get an access token using notebookutils.

Parameters:

Name	Type	Description	Default
scope	str	OAuth2 scope (e.g., "https://api.fabric.microsoft.com/.default")	required

Returns:

Name	Type	Description
str	str	Access token

Raises:

Type	Description
AuthenticationError	If token acquisition fails

Source code in src/fabias/_shared/auth.py

def getToken(self, scope: str) -> str:
    """
    Get an access token using notebookutils.

    Args:
        scope: OAuth2 scope (e.g., "https://api.fabric.microsoft.com/.default")

    Returns:
        str: Access token

    Raises:
        AuthenticationError: If token acquisition fails
    """
    try:
        # notebookutils.credentials.getToken expects a resource URL
        # Convert scope to resource if needed (remove /.default suffix)
        resource = scope.replace("/.default", "")

        token = self._notebookutils.credentials.getToken(resource)

        if not token:
            raise AuthenticationError(f"No token returned for scope: {scope}")

        return cast(str, token)

    except AttributeError as e:
        raise AuthenticationError("notebookutils.credentials.getToken not available", cause=e)
    except Exception as e:
        raise AuthenticationError(f"Failed to get token from notebookutils: {str(e)}", cause=e)

fabias.AutoAuth

Bases: AuthProvider

Automatic authentication provider that selects the best method.

This provider automatically chooses between FabricAuth (when inside Fabric) and ServicePrincipalAuth (when standalone) based on the detected runtime environment.

For standalone usage, credentials must be provided either directly or via environment variables/configuration.

Examples:

Inside Fabric:

>>> auth = AutoAuth()  # Automatically uses FabricAuth

Standalone with explicit credentials:

>>> auth = AutoAuth(
...     tenant_id="...",
...     client_id="...",
...     client_secret="..."
... )

Source code in src/fabias/_shared/auth.py

class AutoAuth(AuthProvider):
    """
    Automatic authentication provider that selects the best method.

    This provider automatically chooses between FabricAuth (when inside
    Fabric) and ServicePrincipalAuth (when standalone) based on the
    detected runtime environment.

    For standalone usage, credentials must be provided either directly
    or via environment variables/configuration.

    Examples:
        Inside Fabric:
        >>> auth = AutoAuth()  # Automatically uses FabricAuth

        Standalone with explicit credentials:
        >>> auth = AutoAuth(
        ...     tenant_id="...",
        ...     client_id="...",
        ...     client_secret="..."
        ... )
    """

    def __init__(
        self,
        tenant_id: Optional[str] = None,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
    ):
        """
        Initialize with optional service principal credentials.

        If running in Fabric, credentials are ignored. If running standalone,
        credentials are required.

        Args:
            tenant_id: Azure AD tenant ID (for standalone)
            client_id: Application client ID (for standalone)
            client_secret: Client secret (for standalone)
        """
        self._provider: Optional[AuthProvider] = None

        if runtime.isFabric:
            self._provider = FabricAuth()
        else:
            if not all([tenant_id, client_id, client_secret]):
                raise AuthenticationError(
                    "Service principal credentials required for standalone execution. "
                    "Provide tenant_id, client_id, and client_secret."
                )
            # Type narrowing: we've already validated these are not None
            self._provider = ServicePrincipalAuth(
                tenant_id=cast(str, tenant_id),
                client_id=cast(str, client_id),
                client_secret=cast(str, client_secret),
            )

    def getToken(self, scope: str) -> str:
        """
        Get an access token using the automatically selected provider.

        Args:
            scope: OAuth2 scope

        Returns:
            str: Access token
        """
        assert self._provider is not None, "Provider not initialized"
        return self._provider.getToken(scope)

Functions

__init__(tenant_id=None, client_id=None, client_secret=None)

Initialize with optional service principal credentials.

If running in Fabric, credentials are ignored. If running standalone, credentials are required.

Parameters:

Name	Type	Description	Default
tenant_id	Optional[str]	Azure AD tenant ID (for standalone)	None
client_id	Optional[str]	Application client ID (for standalone)	None
client_secret	Optional[str]	Client secret (for standalone)	None

Source code in src/fabias/_shared/auth.py

def __init__(
    self,
    tenant_id: Optional[str] = None,
    client_id: Optional[str] = None,
    client_secret: Optional[str] = None,
):
    """
    Initialize with optional service principal credentials.

    If running in Fabric, credentials are ignored. If running standalone,
    credentials are required.

    Args:
        tenant_id: Azure AD tenant ID (for standalone)
        client_id: Application client ID (for standalone)
        client_secret: Client secret (for standalone)
    """
    self._provider: Optional[AuthProvider] = None

    if runtime.isFabric:
        self._provider = FabricAuth()
    else:
        if not all([tenant_id, client_id, client_secret]):
            raise AuthenticationError(
                "Service principal credentials required for standalone execution. "
                "Provide tenant_id, client_id, and client_secret."
            )
        # Type narrowing: we've already validated these are not None
        self._provider = ServicePrincipalAuth(
            tenant_id=cast(str, tenant_id),
            client_id=cast(str, client_id),
            client_secret=cast(str, client_secret),
        )

getToken(scope)

Get an access token using the automatically selected provider.

Parameters:

Name	Type	Description	Default
scope	str	OAuth2 scope	required

Returns:

Name	Type	Description
str	str	Access token

Source code in src/fabias/_shared/auth.py

def getToken(self, scope: str) -> str:
    """
    Get an access token using the automatically selected provider.

    Args:
        scope: OAuth2 scope

    Returns:
        str: Access token
    """
    assert self._provider is not None, "Provider not initialized"
    return self._provider.getToken(scope)

Exceptions

fabias.FabiasError

Bases: Exception

Base exception for all fabias errors.

Provides exception chaining support similar to Java's exception chaining, allowing nested exceptions to preserve the full error context.

Attributes:

Name	Type	Description
message	str	Human-readable error description
cause	Exception	Original exception that caused this error, if any

Examples:

>>> try:
...     risky_operation()
... except SomeError as e:
...     raise FabiasError("Operation failed", cause=e)

Source code in src/fabias/_shared/exceptions.py

class FabiasError(Exception):
    """
    Base exception for all fabias errors.

    Provides exception chaining support similar to Java's exception chaining,
    allowing nested exceptions to preserve the full error context.

    Attributes:
        message (str): Human-readable error description
        cause (Exception): Original exception that caused this error, if any

    Examples:
        >>> try:
        ...     risky_operation()
        ... except SomeError as e:
        ...     raise FabiasError("Operation failed", cause=e)
    """

    def __init__(self, message: str, cause: Optional[Exception] = None):
        """
        Initialize the exception with a message and optional cause.

        Args:
            message: Human-readable error description
            cause: Original exception that caused this error
        """
        self.message = message
        self.cause = cause

        # Build full message including cause
        full_message = message
        if cause:
            full_message = f"{message} -> {type(cause).__name__}: {str(cause)}"

        super().__init__(full_message)

        # Set the __cause__ for Python's built-in exception chaining
        if cause:
            self.__cause__ = cause

Functions

__init__(message, cause=None)

Initialize the exception with a message and optional cause.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error description	required
cause	Optional[Exception]	Original exception that caused this error	None

Source code in src/fabias/_shared/exceptions.py

def __init__(self, message: str, cause: Optional[Exception] = None):
    """
    Initialize the exception with a message and optional cause.

    Args:
        message: Human-readable error description
        cause: Original exception that caused this error
    """
    self.message = message
    self.cause = cause

    # Build full message including cause
    full_message = message
    if cause:
        full_message = f"{message} -> {type(cause).__name__}: {str(cause)}"

    super().__init__(full_message)

    # Set the __cause__ for Python's built-in exception chaining
    if cause:
        self.__cause__ = cause

fabias.AuthenticationError

Bases: FabiasError

Exception raised when authentication fails.

This exception is raised when: - OAuth2 token acquisition fails - Credentials are invalid or expired - Service principal authentication fails - notebookutils credentials are unavailable

Examples:

>>> try:
...     token = auth.getToken()
... except AuthenticationError as e:
...     print(f"Auth failed: {e.message}")

Source code in src/fabias/_shared/exceptions.py

class AuthenticationError(FabiasError):
    """
    Exception raised when authentication fails.

    This exception is raised when:
    - OAuth2 token acquisition fails
    - Credentials are invalid or expired
    - Service principal authentication fails
    - notebookutils credentials are unavailable

    Examples:
        >>> try:
        ...     token = auth.getToken()
        ... except AuthenticationError as e:
        ...     print(f"Auth failed: {e.message}")
    """

    pass

fabias.ApiError

Bases: FabiasError

Exception raised when an API request fails.

This exception is raised when: - HTTP request returns an error status code (4xx, 5xx) - Response cannot be parsed - Request times out - Network errors occur

Attributes:

Name	Type	Description
status_code	int	HTTP status code, if available
response_body	Any	Response body content, if available

Examples:

>>> try:
...     response = client.get("/workspaces")
... except ApiError as e:
...     print(f"API error {e.status_code}: {e.message}")

Source code in src/fabias/_shared/exceptions.py

class ApiError(FabiasError):
    """
    Exception raised when an API request fails.

    This exception is raised when:
    - HTTP request returns an error status code (4xx, 5xx)
    - Response cannot be parsed
    - Request times out
    - Network errors occur

    Attributes:
        status_code (int): HTTP status code, if available
        response_body (Any): Response body content, if available

    Examples:
        >>> try:
        ...     response = client.get("/workspaces")
        ... except ApiError as e:
        ...     print(f"API error {e.status_code}: {e.message}")
    """

    def __init__(
        self,
        message: str,
        cause: Optional[Exception] = None,
        status_code: Optional[int] = None,
        response_body: Any = None,
    ):
        """
        Initialize the API error with details.

        Args:
            message: Human-readable error description
            cause: Original exception that caused this error
            status_code: HTTP status code from the response
            response_body: Response body content for debugging
        """
        super().__init__(message, cause)
        self.status_code = status_code
        self.response_body = response_body

Functions

__init__(message, cause=None, status_code=None, response_body=None)

Initialize the API error with details.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error description	required
cause	Optional[Exception]	Original exception that caused this error	None
status_code	Optional[int]	HTTP status code from the response	None
response_body	Any	Response body content for debugging	None

Source code in src/fabias/_shared/exceptions.py

def __init__(
    self,
    message: str,
    cause: Optional[Exception] = None,
    status_code: Optional[int] = None,
    response_body: Any = None,
):
    """
    Initialize the API error with details.

    Args:
        message: Human-readable error description
        cause: Original exception that caused this error
        status_code: HTTP status code from the response
        response_body: Response body content for debugging
    """
    super().__init__(message, cause)
    self.status_code = status_code
    self.response_body = response_body

fabias.NotFoundError

Bases: ApiError

Exception raised when a requested resource is not found.

This is a specific case of ApiError for 404 responses. It provides clearer semantics when handling "not found" scenarios separately from other API errors.

Attributes:

Name	Type	Description
resource_type	str	Type of resource that wasn't found
resource_id	str	Identifier of the missing resource

Examples:

>>> try:
...     workspace = client.workspace("nonexistent")
... except NotFoundError as e:
...     print(f"{e.resource_type} '{e.resource_id}' not found")

Source code in src/fabias/_shared/exceptions.py

class NotFoundError(ApiError):
    """
    Exception raised when a requested resource is not found.

    This is a specific case of ApiError for 404 responses. It provides
    clearer semantics when handling "not found" scenarios separately
    from other API errors.

    Attributes:
        resource_type (str): Type of resource that wasn't found
        resource_id (str): Identifier of the missing resource

    Examples:
        >>> try:
        ...     workspace = client.workspace("nonexistent")
        ... except NotFoundError as e:
        ...     print(f"{e.resource_type} '{e.resource_id}' not found")
    """

    def __init__(
        self,
        message: str,
        resource_type: Optional[str] = None,
        resource_id: Optional[str] = None,
        cause: Optional[Exception] = None,
    ):
        """
        Initialize the not found error with resource details.

        Args:
            message: Human-readable error description
            resource_type: Type of resource (e.g., "Workspace", "Pipeline")
            resource_id: Identifier that was searched for
            cause: Original exception that caused this error
        """
        super().__init__(message, cause=cause, status_code=404)
        self.resource_type = resource_type
        self.resource_id = resource_id

Functions

__init__(message, resource_type=None, resource_id=None, cause=None)

Initialize the not found error with resource details.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error description	required
resource_type	Optional[str]	Type of resource (e.g., "Workspace", "Pipeline")	None
resource_id	Optional[str]	Identifier that was searched for	None
cause	Optional[Exception]	Original exception that caused this error	None

Source code in src/fabias/_shared/exceptions.py

def __init__(
    self,
    message: str,
    resource_type: Optional[str] = None,
    resource_id: Optional[str] = None,
    cause: Optional[Exception] = None,
):
    """
    Initialize the not found error with resource details.

    Args:
        message: Human-readable error description
        resource_type: Type of resource (e.g., "Workspace", "Pipeline")
        resource_id: Identifier that was searched for
        cause: Original exception that caused this error
    """
    super().__init__(message, cause=cause, status_code=404)
    self.resource_type = resource_type
    self.resource_id = resource_id

Runtime

fabias.runtime = Runtime() module-attribute