from dataclasses import dataclass from anydi import singleton from fansifter_common.auth.services import AuthService from dmp.adapters.db import DefaultDB, transactional from dmp.core.types import EncryptedToken from dmp.meta.exceptions import ( MetaAudienceNotFoundError, MetaUserConnectionNotFoundError, ) from dmp.meta.repositories import MetaAudienceRepository from dmp.meta.services import MetaUserConnectionService @dataclass class GetMetaAudienceShareRequest: share_id: str @dataclass class GetMetaAudienceShareResponse: audience_id: str audience_external_id: str audience_name: str fans_count: int ad_account_external_id: str user_access_token: EncryptedToken @singleton class GetMetaAudienceShareHandler: def __init__( self, db: DefaultDB, auth_service: AuthService, meta_audience_repository: MetaAudienceRepository, user_connection_service: MetaUserConnectionService, ) -> None: self.db = db self.auth_service = auth_service self.meta_audience_repository = meta_audience_repository self.user_connection_service = user_connection_service @transactional def handle( self, request: GetMetaAudienceShareRequest ) -> GetMetaAudienceShareResponse: meta_audience = self.meta_audience_repository.get_by_share_id(request.share_id) if meta_audience is None: raise MetaAudienceNotFoundError share = meta_audience.share user_connection = ( self.user_connection_service.get_active_audience_sharing_connection( audience_id=meta_audience.audience_id, ad_account_id=meta_audience.ad_account_id, identity_id=share.created_by, ) ) if user_connection is None: raise MetaUserConnectionNotFoundError return GetMetaAudienceShareResponse( audience_id=meta_audience.id, audience_external_id=meta_audience.external_id, audience_name=meta_audience.name, fans_count=meta_audience.fans_count, ad_account_external_id=meta_audience.ad_account.external_id, user_access_token=user_connection.token, )