118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
"""Channel connection setup models."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
CONNECTION_STATUSES = {"draft", "pairing", "connected", "running", "degraded", "error", "revoked"}
|
|
|
|
|
|
def iso_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ChannelConnection:
|
|
connection_id: str
|
|
owner_user_id: str | None
|
|
channel_id: str
|
|
kind: str
|
|
mode: str
|
|
display_name: str
|
|
account_id: str
|
|
status: str
|
|
auth_type: str
|
|
credentials_ref: str | None = None
|
|
connector_ref: str | None = None
|
|
pairing_session_id: str | None = None
|
|
runtime_config: dict[str, Any] = field(default_factory=dict)
|
|
capabilities: list[str] = field(default_factory=list)
|
|
created_at: str = field(default_factory=iso_now)
|
|
updated_at: str = field(default_factory=iso_now)
|
|
last_seen_at: str | None = None
|
|
last_error: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ChannelConnection":
|
|
return cls(
|
|
connection_id=str(data.get("connection_id") or ""),
|
|
owner_user_id=_optional_string(data.get("owner_user_id")),
|
|
channel_id=str(data.get("channel_id") or ""),
|
|
kind=str(data.get("kind") or ""),
|
|
mode=str(data.get("mode") or ""),
|
|
display_name=str(data.get("display_name") or ""),
|
|
account_id=str(data.get("account_id") or ""),
|
|
status=str(data.get("status") or "draft"),
|
|
auth_type=str(data.get("auth_type") or ""),
|
|
credentials_ref=_optional_string(data.get("credentials_ref")),
|
|
connector_ref=_optional_string(data.get("connector_ref")),
|
|
pairing_session_id=_optional_string(data.get("pairing_session_id")),
|
|
runtime_config=dict(data.get("runtime_config") or {}),
|
|
capabilities=[str(item) for item in data.get("capabilities") or []],
|
|
created_at=str(data.get("created_at") or iso_now()),
|
|
updated_at=str(data.get("updated_at") or iso_now()),
|
|
last_seen_at=_optional_string(data.get("last_seen_at")),
|
|
last_error=_optional_string(data.get("last_error")),
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class PairingSession:
|
|
pairing_session_id: str
|
|
kind: str
|
|
scope: str
|
|
token: str
|
|
status: str
|
|
expires_at_ms: int
|
|
created_at_ms: int
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "PairingSession":
|
|
return cls(
|
|
pairing_session_id=str(data.get("pairing_session_id") or ""),
|
|
kind=str(data.get("kind") or ""),
|
|
scope=str(data.get("scope") or ""),
|
|
token=str(data.get("token") or ""),
|
|
status=str(data.get("status") or "pending"),
|
|
expires_at_ms=int(data.get("expires_at_ms") or 0),
|
|
created_at_ms=int(data.get("created_at_ms") or 0),
|
|
)
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ChannelRuntimeSpec:
|
|
channel_id: str
|
|
kind: str
|
|
mode: str
|
|
account_id: str
|
|
display_name: str
|
|
config: dict[str, Any] = field(default_factory=dict)
|
|
secrets_ref: str | None = None
|
|
external_endpoint: str | None = None
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ValidationResult:
|
|
ok: bool
|
|
status: str
|
|
account_id: str | None = None
|
|
display_name: str | None = None
|
|
error: str | None = None
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
def _optional_string(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|