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.uuid import uuid_string from ows_text_campaigns.adapters.db.decorators import transactional from ows_text_campaigns.campaigns.enums import ShorteningMethod from ows_text_campaigns.campaigns.models import ShortenedUrl from ows_text_campaigns.campaigns.services import CampaignService from ows_text_campaigns.campaigns.types import UrlPathLike from ows_text_campaigns.campaigns.validators import ShortenedUrlValidator @dataclass(kw_only=True) class ValidateCampaignShortenedUrlRequest(AuthRequest): campaign_id: str shortened_url_id: str | None url: str method: ShorteningMethod domain: str path: UrlPathLike @singleton class ValidateCampaignShortenedUrlHandler: permission = Permission("text_campaign", "edit") def __init__( self, auth_service: AuthService, campaign_service: CampaignService, shortened_url_validator: ShortenedUrlValidator, ) -> None: self.auth_service = auth_service self.campaign_service = campaign_service self.shortened_url_validator = shortened_url_validator @transactional def handle(self, request: ValidateCampaignShortenedUrlRequest) -> ShortenedUrl: campaign = self.campaign_service.get_campaign(request.campaign_id) self.auth_service.check_account_resource( request.identity_id, account=campaign.account, permission=self.permission, resource_id=campaign.id, ) # Validate allowed url domain url_domain = self.shortened_url_validator.validate_allowed_domain( identity_id=request.identity_id, account=campaign.account, domain=request.domain, ) shortened_urls_by_id: dict[str, ShortenedUrl] = { shortened_url.id: shortened_url for shortened_url in campaign.shortened_urls } # Validate path is available for shortening if request.shortened_url_id: existing_shortened_url = shortened_urls_by_id.get(request.shortened_url_id) else: existing_shortened_url = None if ( existing_shortened_url is None or existing_shortened_url.path != request.path ) and request.method == ShorteningMethod.STANDARD: self.shortened_url_validator.validate_availability( method=request.method, path=request.path, ) return ShortenedUrl( id=request.shortened_url_id or uuid_string(), url=request.url, method=request.method, domain=url_domain.domain, path=request.path, )