import enum from typing import Self from pydantic import BaseModel, ConfigDict, Field, model_validator from dmp.adapters.google.enums import ( GoogleAdAccountStatus, GoogleAdAccountUserAccessRole, ) from dmp.core.types import PlainToken class Empty(BaseModel): model_config = ConfigDict(extra="forbid") class GoogleGrantType(enum.StrEnum): AUTHORIZATION_CODE = "authorization_code" REFRESH_TOKEN = "refresh_token" class GoogleTokenRequest(BaseModel): grant_type: GoogleGrantType code: str | None = None redirect_uri: str | None = None refresh_token: str | None = None code_verifier: str | None = None @model_validator(mode="after") def validate_grant_type(self) -> Self: if self.grant_type == GoogleGrantType.AUTHORIZATION_CODE: if not self.code: raise ValueError( f"code is required when grant_type is {self.grant_type}." ) if not self.redirect_uri: raise ValueError( f"redirect_uri is required when grant_type is {self.grant_type}." ) if not self.code_verifier: raise ValueError( f"code_verifier is required when grant_type is {self.grant_type}." ) if self.grant_type == GoogleGrantType.REFRESH_TOKEN and not self.refresh_token: raise ValueError( f"refresh_token is required when grant_type is {self.grant_type}." ) if self.code and self.refresh_token: raise ValueError( "Only one of code or refresh_token should be provided - both are given." ) return self class Response[T](BaseModel): status_code: int content: T | None | Empty = None text: str class AccessToken(BaseModel): access_token: PlainToken scope: str @property def scope_list(self) -> list[str]: return self.scope.split(" ") def scopes_sufficient(self, scopes: list[str], /) -> bool: return set(self.scope_list).issuperset(scopes) class RefreshToken(AccessToken): refresh_token: PlainToken class User(BaseModel): id: str name: str class AdAccountResource(BaseModel): name: str @property def ad_account_id(self) -> int: return int(self.name.split("/")[-1]) class AdAccountResourceResponse(BaseModel): resource_names: list[str] = Field(alias="resourceNames") def get_resources(self) -> list[AdAccountResource]: return [AdAccountResource(name=i) for i in self.resource_names] class Customer(BaseModel): id: str resource_name: str = Field(alias="resourceName") descriptive_name: str | None = Field(alias="descriptiveName", default=None) test_account: bool = Field(alias="testAccount") status: GoogleAdAccountStatus class CustomerUserAccess(BaseModel): access_role: GoogleAdAccountUserAccessRole = Field(alias="accessRole") class AdAccount(BaseModel): customer: Customer customer_user_access: CustomerUserAccess = Field(alias="customerUserAccess") @property def is_eligible(self) -> bool: return self.customer_user_access.access_role in ( GoogleAdAccountUserAccessRole.STANDARD, GoogleAdAccountUserAccessRole.ADMIN, ) class ClientAdAccount(BaseModel): id: str resource_name: str = Field(alias="resourceName") descriptive_name: str | None = Field(alias="descriptiveName", default=None) test_account: bool = Field(alias="testAccount") status: GoogleAdAccountStatus class NestedAdAccount(BaseModel): customer_client: ClientAdAccount = Field(alias="customerClient") @property def is_eligible(self) -> bool: return self.customer_client.status.is_valid class AdAccountApiResponse(BaseModel): results: list[AdAccount] class NestedAdAccountApiResponse(BaseModel): results: list[NestedAdAccount] = Field(default=[]) class Summary(BaseModel): total_count: int class CampaignsResponse(BaseModel): summary: Summary class UserAdAccount(BaseModel): id: str name: str user_role: GoogleAdAccountUserAccessRole campaigns: CampaignsResponse login_customer_id: str | None = None @property def campaigns_count(self) -> int: return self.campaigns.summary.total_count class CampaignApiResponse(BaseModel): total_results_count: int = Field(alias="totalResultsCount", default=0) class Audience(BaseModel): resource_name: str = Field(alias="resourceName") @property def id(self) -> str: return self.resource_name.split("/")[-1] class AudienceApiResponse(BaseModel): results: list[Audience]