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 fansifter_common.utils import timezone from ows_text_campaigns.adapters.db.decorators import transactional from ows_text_campaigns.adapters.ows_url_shortener import OwsUrlShortenerClient from ows_text_campaigns.campaigns.exceptions import CampaignMustBeDraftError from ows_text_campaigns.campaigns.services import CampaignService @dataclass(kw_only=True) class DeleteCampaignRequest(AuthRequest): campaign_id: str @singleton class DeleteCampaignHandler: permission = Permission("text_campaign", "delete") def __init__( self, auth_service: AuthService, campaign_service: CampaignService, ows_url_shortener_client: OwsUrlShortenerClient, ) -> None: self.auth_service = auth_service self.campaign_service = campaign_service self.ows_url_shortener_client = ows_url_shortener_client @transactional def handle(self, request: DeleteCampaignRequest) -> None: campaign = self.campaign_service.get_campaign(request.campaign_id) if not campaign.is_draft: raise CampaignMustBeDraftError self.auth_service.authorize_account( request.identity_id, account=campaign.account, permission=self.permission, ) standard_paths = [ shortened_url.path for shortened_url in campaign.shortened_urls if shortened_url.is_standard ] if standard_paths: self.ows_url_shortener_client.delete_paths(standard_paths) campaign.deleted_at = timezone.now() campaign.deleted_by = request.identity_id self.campaign_service.save_campaign(campaign) return None