Spaces:
Sleeping
Sleeping
File size: 1,796 Bytes
a1bab2d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """Safe application boundary for provider connected accounts."""
from __future__ import annotations
from dataclasses import asdict, dataclass
from enum import StrEnum
from typing import Any, Protocol
class ConnectedAccountLoginMode(StrEnum):
"""Supported interactive authorization flows."""
BROWSER = "browser"
DEVICE = "device"
class ConnectedAccountState(StrEnum):
"""Customer-visible connected-account lifecycle states."""
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
ERROR = "error"
@dataclass(frozen=True, slots=True)
class ConnectedAccountStatus:
"""Credential-free state safe to return from the local Admin API."""
provider_id: str
state: ConnectedAccountState
connected: bool
revision: int
attempt_id: str | None = None
email: str | None = None
mode: ConnectedAccountLoginMode | None = None
authorization_url: str | None = None
verification_url: str | None = None
user_code: str | None = None
expires_at: int | None = None
model_count: int | None = None
message: str | None = None
def as_dict(self) -> dict[str, Any]:
"""Serialize only the explicitly safe status contract."""
return {key: value for key, value in asdict(self).items() if value is not None}
class ConnectedAccountPort(Protocol):
"""One process-lifetime provider authentication owner."""
def is_connected(self) -> bool: ...
def status(self) -> ConnectedAccountStatus: ...
async def start_login(
self, mode: ConnectedAccountLoginMode
) -> ConnectedAccountStatus: ...
async def cancel_login(self) -> ConnectedAccountStatus: ...
async def disconnect(self) -> ConnectedAccountStatus: ...
async def close(self) -> None: ...
|