from dataclasses import dataclass from anydi import singleton from fansifter_common.auth.identity import Identity from fansifter_common.auth.services import AuthService from fansifter_common.auth.types import Permission from dmp.adapters.features import ( AUDIENCE_SHOW_ADD_SECONDARY_FANS_TOGGLE, AUDIENCE_SHOW_SMS_CAMPAIGNS_PAGE, Features, ) from dmp.audiences.enums import AudienceTarget from dmp.audiences.exceptions import AudienceTargetNotSupportedError from dmp.audiences.models import Audience from dmp.audiences.services import AudienceService @dataclass class GetAudienceRequest: identity: Identity audience_id: str @singleton class GetAudienceHandler: ad_target_permission = Permission("audience", "view_ad_target") email_target_permission = Permission("audience", "view_email_target") text_target_permission = Permission("audience", "view_text_target") def __init__( self, auth_service: AuthService, audience_service: AudienceService, features: Features, ) -> None: self.auth_service = auth_service self.audience_service = audience_service self.features = features def handle(self, request: GetAudienceRequest) -> Audience: audience = self.audience_service.get_audience(request.audience_id) match audience.target: case AudienceTarget.EMAIL: self.auth_service.check_account_resource( request.identity.id, account=audience.account, permission=self.email_target_permission, resource_id=audience.id, ) case AudienceTarget.ADS: self.auth_service.check_account_resource( request.identity.id, account=audience.account, permission=self.ad_target_permission, resource_id=audience.id, ) case AudienceTarget.TEXT if self.features.is_on_for_identity( feature_flag_name=AUDIENCE_SHOW_SMS_CAMPAIGNS_PAGE, identity_id=request.identity.id, ): self.auth_service.check_account_resource( request.identity.id, account=audience.account, permission=self.text_target_permission, resource_id=audience.id, ) case _: raise AudienceTargetNotSupportedError # Update secondary fans if the feature flag is on if ( self.features.is_on_for_identity( feature_flag_name=AUDIENCE_SHOW_ADD_SECONDARY_FANS_TOGGLE, identity_id=request.identity.id, ) and audience.filters ): audience.filters.set_include_secondary_fans(target=audience.target) return audience