from __future__ import annotations import enum from dmp.adapters.fivetran.enums import ConnectionSetupState, ConnectionSyncState from dmp.adapters.fivetran.models import ConnectionStatus class AppConnectionStatus(enum.StrEnum): CONNECTED = "CONNECTED" SYNCING = "SYNCING" WAITING_FOR_PROCESSING = "WAITING_FOR_PROCESSING" SYNCING_INITIAL = "SYNCING_INITIAL" ERROR = "ERROR" PENDING = "PENDING" NOT_CONNECTED = "NOT_CONNECTED" DELETED = "DELETED" @classmethod def from_connection_status( cls, connection_status: ConnectionStatus, ) -> AppConnectionStatus: if connection_status.setup_state == ConnectionSetupState.INCOMPLETE: return cls.PENDING elif connection_status.setup_state == ConnectionSetupState.BROKEN: return cls.ERROR elif connection_status.setup_state == ConnectionSetupState.CONNECTED: if connection_status.sync_state in ( ConnectionSyncState.SYNCING, ConnectionSyncState.RESCHEDULED, ): if connection_status.is_historical_sync: return cls.SYNCING_INITIAL return cls.SYNCING return cls.CONNECTED return cls.PENDING @classmethod def ordered(cls) -> tuple[AppConnectionStatus, ...]: return ( cls.CONNECTED, cls.SYNCING, cls.WAITING_FOR_PROCESSING, cls.SYNCING_INITIAL, cls.ERROR, cls.PENDING, cls.NOT_CONNECTED, cls.DELETED, ) @classmethod def pre_processed_statuses(cls) -> tuple[AppConnectionStatus, ...]: return (cls.SYNCING_INITIAL, cls.WAITING_FOR_PROCESSING) @classmethod def schema_update_statuses(cls) -> tuple[AppConnectionStatus, ...]: return (cls.CONNECTED, cls.SYNCING) class AdsConnectionPlatform(enum.StrEnum): META = "META" TIKTOK = "TIKTOK" GOOGLE = "GOOGLE" class AppConnectionPlatform(enum.StrEnum): META = "META" TIKTOK = "TIKTOK" SHOPIFY = "SHOPIFY" GOOGLE = "GOOGLE"