Skip to content

domain.enums.connection_status

src.domain.enums.connection_status

Provider connection lifecycle states.

Defines the status state machine for provider connections.

State Machine

PENDING → ACTIVE ↔ EXPIRED/REVOKED → DISCONNECTED

  • PENDING: Auth initiated, awaiting completion
  • ACTIVE: Connected with valid credentials
  • EXPIRED: Credentials expired, needs re-auth
  • REVOKED: Access revoked by user or provider
  • FAILED: Authentication failed
  • DISCONNECTED: User explicitly disconnected (terminal)
Reference
  • docs/architecture/provider-domain-model.md
Usage

from src.domain.enums import ConnectionStatus

if connection.status == ConnectionStatus.ACTIVE: # Connection is usable

Classes

ConnectionStatus

Bases: str, Enum

Provider connection lifecycle states.

Defines valid states for a provider connection and the transitions between them. Used to track connection health and determine if credentials need refresh or re-authentication.

String Enum

Inherits from str for easy serialization and database storage. Values are lowercase for consistency.

State Transitions

PENDING → ACTIVE: Successful authentication PENDING → FAILED: Authentication failed or timed out ACTIVE → EXPIRED: Credentials past expiration ACTIVE → REVOKED: User or provider revoked access EXPIRED → ACTIVE: Successful re-authentication REVOKED → ACTIVE: Successful re-authentication Any → DISCONNECTED: User explicitly disconnects (terminal)

Source code in src/domain/enums/connection_status.py
class ConnectionStatus(str, Enum):
    """Provider connection lifecycle states.

    Defines valid states for a provider connection and the transitions
    between them. Used to track connection health and determine if
    credentials need refresh or re-authentication.

    String Enum:
        Inherits from str for easy serialization and database storage.
        Values are lowercase for consistency.

    State Transitions:
        PENDING → ACTIVE: Successful authentication
        PENDING → FAILED: Authentication failed or timed out
        ACTIVE → EXPIRED: Credentials past expiration
        ACTIVE → REVOKED: User or provider revoked access
        EXPIRED → ACTIVE: Successful re-authentication
        REVOKED → ACTIVE: Successful re-authentication
        Any → DISCONNECTED: User explicitly disconnects (terminal)
    """

    PENDING = "pending"
    """Authentication initiated, awaiting completion.

    Initial state when user starts provider connection flow.
    Transitions to ACTIVE on success, FAILED on error.
    """

    ACTIVE = "active"
    """Connected with valid credentials.

    Connection is healthy and can be used for data sync.
    Credentials are valid and not expired.
    """

    EXPIRED = "expired"
    """Credentials expired, needs re-authentication.

    Connection was previously active but credentials have expired.
    User must re-authenticate to restore connection.
    """

    REVOKED = "revoked"
    """Access revoked by user or provider.

    Provider or user explicitly revoked access.
    Differs from EXPIRED in that credentials are invalid
    regardless of expiration time.
    """

    FAILED = "failed"
    """Authentication failed.

    Initial authentication attempt failed.
    User may retry the connection flow.
    """

    DISCONNECTED = "disconnected"
    """User explicitly disconnected.

    Terminal state - user has removed this connection.
    Credentials are cleared and connection is inactive.
    """

    @classmethod
    def values(cls) -> list[str]:
        """Get all status values as strings.

        Returns:
            list[str]: List of status values.
        """
        return [status.value for status in cls]

    @classmethod
    def is_valid(cls, value: str) -> bool:
        """Check if a string is a valid status.

        Args:
            value: String to check.

        Returns:
            bool: True if value is a valid status.
        """
        return value in cls.values()

    @classmethod
    def active_states(cls) -> list["ConnectionStatus"]:
        """Get states where connection is usable.

        Returns:
            list[ConnectionStatus]: States where sync is possible.
        """
        return [cls.ACTIVE]

    @classmethod
    def needs_reauth_states(cls) -> list["ConnectionStatus"]:
        """Get states requiring re-authentication.

        Returns:
            list[ConnectionStatus]: States needing user action.
        """
        return [cls.EXPIRED, cls.REVOKED, cls.FAILED]

    @classmethod
    def terminal_states(cls) -> list["ConnectionStatus"]:
        """Get terminal states (no recovery).

        Returns:
            list[ConnectionStatus]: Terminal states.
        """
        return [cls.DISCONNECTED]
Attributes
PENDING class-attribute instance-attribute
PENDING = 'pending'

Authentication initiated, awaiting completion.

Initial state when user starts provider connection flow. Transitions to ACTIVE on success, FAILED on error.

ACTIVE class-attribute instance-attribute
ACTIVE = 'active'

Connected with valid credentials.

Connection is healthy and can be used for data sync. Credentials are valid and not expired.

EXPIRED class-attribute instance-attribute
EXPIRED = 'expired'

Credentials expired, needs re-authentication.

Connection was previously active but credentials have expired. User must re-authenticate to restore connection.

REVOKED class-attribute instance-attribute
REVOKED = 'revoked'

Access revoked by user or provider.

Provider or user explicitly revoked access. Differs from EXPIRED in that credentials are invalid regardless of expiration time.

FAILED class-attribute instance-attribute
FAILED = 'failed'

Authentication failed.

Initial authentication attempt failed. User may retry the connection flow.

DISCONNECTED class-attribute instance-attribute
DISCONNECTED = 'disconnected'

User explicitly disconnected.

Terminal state - user has removed this connection. Credentials are cleared and connection is inactive.

Functions
values classmethod
values() -> list[str]

Get all status values as strings.

Returns:

Type Description
list[str]

list[str]: List of status values.

Source code in src/domain/enums/connection_status.py
@classmethod
def values(cls) -> list[str]:
    """Get all status values as strings.

    Returns:
        list[str]: List of status values.
    """
    return [status.value for status in cls]
is_valid classmethod
is_valid(value: str) -> bool

Check if a string is a valid status.

Parameters:

Name Type Description Default
value str

String to check.

required

Returns:

Name Type Description
bool bool

True if value is a valid status.

Source code in src/domain/enums/connection_status.py
@classmethod
def is_valid(cls, value: str) -> bool:
    """Check if a string is a valid status.

    Args:
        value: String to check.

    Returns:
        bool: True if value is a valid status.
    """
    return value in cls.values()
active_states classmethod
active_states() -> list[ConnectionStatus]

Get states where connection is usable.

Returns:

Type Description
list[ConnectionStatus]

list[ConnectionStatus]: States where sync is possible.

Source code in src/domain/enums/connection_status.py
@classmethod
def active_states(cls) -> list["ConnectionStatus"]:
    """Get states where connection is usable.

    Returns:
        list[ConnectionStatus]: States where sync is possible.
    """
    return [cls.ACTIVE]
needs_reauth_states classmethod
needs_reauth_states() -> list[ConnectionStatus]

Get states requiring re-authentication.

Returns:

Type Description
list[ConnectionStatus]

list[ConnectionStatus]: States needing user action.

Source code in src/domain/enums/connection_status.py
@classmethod
def needs_reauth_states(cls) -> list["ConnectionStatus"]:
    """Get states requiring re-authentication.

    Returns:
        list[ConnectionStatus]: States needing user action.
    """
    return [cls.EXPIRED, cls.REVOKED, cls.FAILED]
terminal_states classmethod
terminal_states() -> list[ConnectionStatus]

Get terminal states (no recovery).

Returns:

Type Description
list[ConnectionStatus]

list[ConnectionStatus]: Terminal states.

Source code in src/domain/enums/connection_status.py
@classmethod
def terminal_states(cls) -> list["ConnectionStatus"]:
    """Get terminal states (no recovery).

    Returns:
        list[ConnectionStatus]: Terminal states.
    """
    return [cls.DISCONNECTED]