from dataclasses import dataclass from anydi import singleton from fansifter_common.auth.requests import AuthRequest from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from dmp.adapters.db import DefaultDB, transactional from dmp.audiences.enums import AudienceExportStatus from dmp.meta.exceptions import MetaAudienceNotFoundError from dmp.meta.repositories import MetaAudienceRepository @dataclass class GetMetaAudienceShareStatusRequest(AuthRequest): meta_audience_id: str @dataclass class GetMetaAudienceShareStatusResponse: audience_id: str audience_name: str ad_account_name: str status: AudienceExportStatus fans_count: int processed_fans: int progress: float seconds_remaining: int | None url: str @singleton class GetMetaAudienceShareStatusHandler: permission = Permission("ad_account", "view") def __init__( self, db: DefaultDB, auth_service: AuthService, meta_audience_repository: MetaAudienceRepository, ) -> None: self.db = db self.auth_service = auth_service self.meta_audience_repository = meta_audience_repository @transactional def handle( self, request: GetMetaAudienceShareStatusRequest ) -> GetMetaAudienceShareStatusResponse: meta_audience = self.meta_audience_repository.get(request.meta_audience_id) if meta_audience is None: raise MetaAudienceNotFoundError self.auth_service.authorize_account( request.identity_id, permission=self.permission, account=meta_audience.share.audience.account, ) return GetMetaAudienceShareStatusResponse( audience_id=meta_audience.id, audience_name=meta_audience.name, ad_account_name=meta_audience.ad_account.name or "", status=meta_audience.share.status, fans_count=meta_audience.fans_count, processed_fans=meta_audience.share.processed_fans, progress=meta_audience.share.progress, seconds_remaining=meta_audience.share.seconds_remaining, url=meta_audience.url, )