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.tiktok.exceptions import ( TikTokAudienceNotFoundError, TikTokUserConnectionNotFoundError, ) from dmp.tiktok.repositories import TikTokAudienceRepository from dmp.tiktok.services import TikTokUserConnectionService @dataclass class GetTikTokAudienceShareRequest: share_id: str @dataclass class GetTikTokAudienceShareResponse: audience_id: str audience_external_id: str | None audience_name: str fans_count: int ad_account_external_id: str user_access_token: EncryptedToken @singleton class GetTikTokAudienceShareHandler: def __init__( self, db: DefaultDB, auth_service: AuthService, tiktok_audience_repository: TikTokAudienceRepository, user_connection_service: TikTokUserConnectionService, ) -> None: self.db = db self.auth_service = auth_service self.tiktok_audience_repository = tiktok_audience_repository self.user_connection_service = user_connection_service @transactional def handle( self, request: GetTikTokAudienceShareRequest ) -> GetTikTokAudienceShareResponse: tiktok_audience = self.tiktok_audience_repository.get_by_share_id( request.share_id ) if tiktok_audience is None: raise TikTokAudienceNotFoundError share = tiktok_audience.share user_connection = ( self.user_connection_service.get_active_audience_sharing_connection( audience_id=tiktok_audience.audience_id, ad_account_id=tiktok_audience.ad_account_id, identity_id=share.created_by, ) ) if user_connection is None: raise TikTokUserConnectionNotFoundError return GetTikTokAudienceShareResponse( audience_id=tiktok_audience.id, audience_external_id=tiktok_audience.external_id, audience_name=tiktok_audience.name, fans_count=tiktok_audience.fans_count, ad_account_external_id=tiktok_audience.ad_account.external_id, user_access_token=user_connection.token, )