From 7fa4dcd1315349d96828b1fbc33f129994775186 Mon Sep 17 00:00:00 2001 From: Ali Hesari Date: Sun, 20 Sep 2026 03:37:36 +0200 Subject: [PATCH] feat(ads): add catalogs, reach and frequency, the ad archive and account settings Also adds goals() so a caller asks the connection what it can run rather than assuming. --- README.md | 2 +- src/fopost/models.py | 212 ++++++++++ src/fopost/resources/ads.py | 770 +++++++++++++++++++++++++++++++++++- 3 files changed, 982 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 828b424..339b5fa 100644 --- a/README.md +++ b/README.md @@ -207,7 +207,7 @@ except FopostError as err: | `media` | `presign`, `complete`, `upload_direct` | | `ai` | `credits`, `generate_caption`, `rewrite`, `repurpose_url` | | `inbox` | `list`, `threads`, `conversations`, `unread_count`, `accounts`, `platforms`, `mark_thread_read`, `refresh`, `update`, `edit_comment`, `reply`, `hide`, `unhide`, `delete`, `like`, `unlike`, `pin`, `unpin`, `react`, `start_conversation`, `set_typing`, `list_approvals`, `approve_reply`, `reject_reply` | -| `ads` | `list`, `external`, `boostable`, `connections`, `sources`, `authorize_meta`, `delete_connection`, `boost`, `create`, `refresh`, `set_status`, `delete`, `audiences`, `create_audience`, `search_targeting`, `lead_forms`, `create_lead_form`, `leads`, `account_tree`, `create_campaign`, `get_campaign`, `update_campaign`, `delete_campaign`, `duplicate_campaign`, `create_ad_set`, `get_ad_set`, `update_ad_set`, `delete_ad_set`, `duplicate_ad_set`, `create_network_ad`, `get_network_ad`, `update_network_ad`, `delete_network_ad`, `duplicate_network_ad`, `bulk_set_status`, `creatives`, `create_creative`, `get_creative`, `delete_creative`, `get_audience`, `update_audience`, `delete_audience`, `add_audience_users`, `estimate_reach`, `insights`, `ad_insights`, `get_lead_form`, `archive_lead_form`, `leads_feed`, `lead_pages`, `subscribe_lead_page`, `unsubscribe_lead_page` | +| `ads` | `list`, `external`, `boostable`, `connections`, `sources`, `authorize_meta`, `delete_connection`, `boost`, `create`, `refresh`, `set_status`, `delete`, `audiences`, `create_audience`, `search_targeting`, `lead_forms`, `create_lead_form`, `leads`, `account_tree`, `create_campaign`, `get_campaign`, `update_campaign`, `delete_campaign`, `duplicate_campaign`, `create_ad_set`, `get_ad_set`, `update_ad_set`, `delete_ad_set`, `duplicate_ad_set`, `create_network_ad`, `get_network_ad`, `update_network_ad`, `delete_network_ad`, `duplicate_network_ad`, `bulk_set_status`, `creatives`, `create_creative`, `get_creative`, `delete_creative`, `get_audience`, `update_audience`, `delete_audience`, `add_audience_users`, `estimate_reach`, `insights`, `ad_insights`, `get_lead_form`, `archive_lead_form`, `leads_feed`, `lead_pages`, `subscribe_lead_page`, `unsubscribe_lead_page`, `goals`, `catalogs`, `create_catalog`, `get_catalog`, `update_catalog`, `delete_catalog`, `catalog_products`, `write_catalog_products`, `product_feeds`, `create_product_feed`, `delete_product_feed`, `feed_uploads`, `start_feed_upload`, `product_sets`, `create_product_set`, `update_product_set`, `delete_product_set`, `reach_frequency`, `create_reach_frequency`, `get_reach_frequency`, `reserve_reach_frequency`, `cancel_reach_frequency`, `library`, `partnership_creators`, `request_partnership`, `revoke_partnership`, `account_activity`, `labels`, `create_label`, `update_label`, `delete_label`, `apply_label`, `studies`, `create_study`, `get_study`, `delete_study`, `ios_campaign_limits`, `high_demand_periods`, `create_high_demand_period`, `delete_high_demand_period`, `value_rule_sets`, `create_value_rule_set`, `delete_value_rule_set` | | `validate` | `post`, `length`, `media` | For an endpoint the SDK does not wrap yet, `client.request` sends an diff --git a/src/fopost/models.py b/src/fopost/models.py index 7a0b4ec..21f1f93 100644 --- a/src/fopost/models.py +++ b/src/fopost/models.py @@ -93,6 +93,26 @@ "LeadPageSubscription", "LeadsFeedPage", "NetworkAd", + "AdActivity", + "AdActivityResult", + "AdLabel", + "AdLibraryEntry", + "AdLibraryPage", + "AdStudy", + "CatalogBatchResult", + "CatalogProduct", + "CatalogProductsPage", + "HighDemandPeriod", + "IosCampaignLimits", + "PartnershipCreator", + "ProductCatalog", + "ProductCatalogsResult", + "ProductFeed", + "ProductFeedUpload", + "ProductSet", + "ReachFrequencyPrediction", + "ReachFrequencyResult", + "ValueRuleSet", "ReachEstimate", ] @@ -959,6 +979,198 @@ class LeadPageSubscription(FopostModel): backfilled: int = 0 +# ─── Product catalogs ────────────────────────────────────────────── + + +class ProductCatalog(FopostModel): + """A product catalog on the connection's business portfolio, read live.""" + + id: str + name: str + vertical: str | None = None + product_count: int | None = None + + +class ProductCatalogsResult(FopostModel): + catalogs: list[ProductCatalog] = [] + workspace_id: str | None = None + + +class CatalogProduct(FopostModel): + id: str + #: Your own key for the product. + retailer_id: str + name: str + description: str | None = None + availability: str | None = None + condition: str | None = None + #: Minor units of ``currency``. + price_minor: int | None = None + currency: str | None = None + image_url: str | None = None + url: str | None = None + + +class CatalogProductsPage(FopostModel): + products: list[CatalogProduct] = [] + next_cursor: str | None = None + + +class CatalogBatchResult(FopostModel): + handles: list[str] = [] + #: Products sent in this batch. + accepted: int = 0 + + +class ProductFeed(FopostModel): + id: str + name: str + #: Set when the network fetches the file on a schedule. + url: str | None = None + schedule: str | None = None + created_at: str | None = None + + +class ProductFeedUpload(FopostModel): + id: str + started_at: str | None = None + ended_at: str | None = None + status: str | None = None + error_count: int | None = None + warning_count: int | None = None + + +class ProductSet(FopostModel): + """The slice of a catalog one catalog ad runs from.""" + + id: str + name: str + product_count: int | None = None + #: The network's own product-set filter. + filter: dict[str, Any] | None = None + + +# ─── Reach and frequency ─────────────────────────────────────────── + + +class ReachFrequencyPrediction(FopostModel): + id: str + name: str | None = None + status: str | None = None + reach: int | None = None + impressions: int | None = None + frequency_cap: int | None = None + #: Account currency, minor units. + budget_minor: int | None = None + start_at: str | None = None + end_at: str | None = None + #: True once the prediction holds inventory. + reserved: bool = False + + +class ReachFrequencyResult(FopostModel): + predictions: list[ReachFrequencyPrediction] = [] + workspace_id: str | None = None + + +# ─── Ad Library ──────────────────────────────────────────────────── + + +class AdLibraryEntry(FopostModel): + """One public archive entry. Read live on every search and stored nowhere.""" + + id: str + page_id: str | None = None + page_name: str | None = None + bodies: list[str] = [] + titles: list[str] = [] + link_urls: list[str] = [] + snapshot_url: str | None = None + publisher_platforms: list[str] = [] + started_at: str | None = None + ended_at: str | None = None + #: Only on the archive's disclosure entries. + currency: str | None = None + spend_lower: int | None = None + spend_upper: int | None = None + impressions_lower: int | None = None + impressions_upper: int | None = None + + +class AdLibraryPage(FopostModel): + entries: list[AdLibraryEntry] = [] + next_cursor: str | None = None + + +# ─── Partnership ads ─────────────────────────────────────────────── + + +class PartnershipCreator(FopostModel): + """A creator who allowlisted this advertiser for partnership ads.""" + + id: str + username: str | None = None + name: str | None = None + status: str | None = None + permissions: list[str] = [] + + +# ─── Ad account settings ─────────────────────────────────────────── + + +class AdActivity(FopostModel): + id: str + event_type: str | None = None + actor_name: str | None = None + object_name: str | None = None + object_type: str | None = None + extra_data: str | None = None + created_at: str | None = None + + +class AdActivityResult(FopostModel): + activity: list[AdActivity] = [] + workspace_id: str | None = None + + +class AdLabel(FopostModel): + id: str + name: str + created_at: str | None = None + + +class AdStudy(FopostModel): + id: str + name: str + description: str | None = None + type: str | None = None + status: str | None = None + start_at: str | None = None + end_at: str | None = None + + +class IosCampaignLimits(FopostModel): + #: How many iOS 14 campaigns the account may run at once. + limit: int | None = None + used: int | None = None + app_id: str | None = None + + +class HighDemandPeriod(FopostModel): + id: str + start_at: str | None = None + end_at: str | None = None + budget_value: float | None = None + budget_value_type: str | None = None + + +class ValueRuleSet(FopostModel): + id: str + name: str + status: str | None = None + rules: list[dict[str, Any]] = [] + + class ContentSignal(FopostModel): level: Literal["info", "warn"] | str code: str diff --git a/src/fopost/resources/ads.py b/src/fopost/resources/ads.py index a9fd434..fb5beeb 100644 --- a/src/fopost/resources/ads.py +++ b/src/fopost/resources/ads.py @@ -1,4 +1,4 @@ -"""``client.ads`` — Meta ads, audiences and lead forms. +"""``client.ads`` — Meta ads, catalogs, audiences, the ad archive and lead forms. Every method needs the ``ads`` scope; ``boost``, ``create``, ``set_status``, ``delete``, ``bulk_set_status`` and every create, update, delete or duplicate on @@ -15,17 +15,25 @@ from ..models import ( Ad, AdAccountTree, + AdActivityResult, AdCampaign, AdConnection, AdCreative, AdInsightsReport, + AdLabel, + AdLibraryPage, AdSet, AdSource, + AdStudy, Audience, AudiencesResult, BoostablePost, BulkAdStatusResult, + CatalogBatchResult, + CatalogProductsPage, ExternalAd, + HighDemandPeriod, + IosCampaignLimits, LeadFormDetail, LeadFormSource, LeadPage, @@ -33,8 +41,17 @@ LeadsFeedPage, LeadsPage, NetworkAd, + PartnershipCreator, + ProductCatalog, + ProductCatalogsResult, + ProductFeed, + ProductFeedUpload, + ProductSet, ReachEstimate, + ReachFrequencyPrediction, + ReachFrequencyResult, TargetingOption, + ValueRuleSet, ) from ._base import UNSET, Resource, drop_unset, parse_list @@ -815,6 +832,757 @@ def unsubscribe_lead_page(self, page_id: str, *, workspace_id: str, connection_i params={"workspace_id": workspace_id, "connection_id": connection_id}, ) + # ─── Goals ───────────────────────────────────────────────────── + + def goals(self, *, connection_id: str, workspace_id: str | None = None) -> builtins.list[str]: + """The goals this connection's network can run right now. + + Ask rather than assume: a goal the deployment is not set up for is + absent here and is refused if you send it anyway. + """ + result = unwrap( + self._http.get( + "/ads/goals", + {"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ) + return [str(goal) for goal in result] if isinstance(result, list) else [] + + # ─── Product catalogs ────────────────────────────────────────── + + def catalogs( + self, *, connection_id: str, workspace_id: str | None = None + ) -> ProductCatalogsResult: + """Catalogs the connection's business portfolios reach. Read live, never stored.""" + return ProductCatalogsResult.model_validate( + unwrap( + self._http.get( + "/ads/catalogs", + {"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ) + ) + + def create_catalog( + self, + *, + workspace_id: str, + connection_id: str, + name: str, + vertical: str | None = None, + ) -> ProductCatalog: + """Created on the connection's business portfolio. Also needs ``publish``.""" + body: dict[str, Any] = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "name": name, + } + if vertical is not None: + body["vertical"] = vertical + return ProductCatalog.model_validate(unwrap(self._http.post("/ads/catalogs", body))) + + def get_catalog( + self, catalog_id: str, *, connection_id: str, workspace_id: str | None = None + ) -> ProductCatalog: + return ProductCatalog.model_validate( + unwrap( + self._http.get( + f"/ads/catalogs/{catalog_id}", + {"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ) + ) + + def update_catalog( + self, catalog_id: str, *, workspace_id: str, connection_id: str, name: str + ) -> ProductCatalog: + """Also needs ``publish``.""" + body = {"workspaceId": workspace_id, "connectionId": connection_id, "name": name} + return ProductCatalog.model_validate( + unwrap( + self._http.request( + "PATCH", + f"/ads/catalogs/{catalog_id}", + json=body, + params={"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ) + ) + + def delete_catalog(self, catalog_id: str, *, workspace_id: str, connection_id: str) -> None: + """Deletes every product, feed and set in it. Also needs ``publish``.""" + self._http.request( + "DELETE", + f"/ads/catalogs/{catalog_id}", + params={"workspace_id": workspace_id, "connection_id": connection_id}, + ) + + def catalog_products( + self, + catalog_id: str, + *, + connection_id: str, + workspace_id: str | None = None, + after: str | None = None, + ) -> CatalogProductsPage: + """One page of products; pass ``next_cursor`` back as ``after``.""" + return CatalogProductsPage.model_validate( + unwrap( + self._http.get( + f"/ads/catalogs/{catalog_id}/products", + { + "workspace_id": workspace_id, + "connection_id": connection_id, + "after": after, + }, + ) + ) + ) + + def write_catalog_products( + self, + catalog_id: str, + *, + workspace_id: str, + connection_id: str, + products: Sequence[Mapping[str, Any]], + ) -> CatalogBatchResult: + """Up to 500 upserts and deletes in one batch, keyed by ``retailerId``. + + Also needs ``publish``. + """ + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "products": [dict(p) for p in products], + } + return CatalogBatchResult.model_validate( + unwrap(self._http.post(f"/ads/catalogs/{catalog_id}/products", body)) + ) + + def product_feeds( + self, catalog_id: str, *, connection_id: str, workspace_id: str | None = None + ) -> builtins.list[ProductFeed]: + return parse_list( + ProductFeed, + unwrap( + self._http.get( + f"/ads/catalogs/{catalog_id}/feeds", + {"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ), + ) + + def create_product_feed( + self, + catalog_id: str, + *, + workspace_id: str, + connection_id: str, + name: str, + url: str | None = None, + schedule: str | None = None, + ) -> ProductFeed: + """``schedule`` is ``HOURLY``, ``DAILY`` or ``WEEKLY`` and needs ``url``. + + Also needs ``publish``. + """ + body: dict[str, Any] = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "name": name, + } + optional = {"url": url, "schedule": schedule} + body.update({k: v for k, v in optional.items() if v is not None}) + return ProductFeed.model_validate( + unwrap(self._http.post(f"/ads/catalogs/{catalog_id}/feeds", body)) + ) + + def delete_product_feed( + self, catalog_id: str, feed_id: str, *, workspace_id: str, connection_id: str + ) -> None: + """Also needs ``publish``.""" + self._http.request( + "DELETE", + f"/ads/catalogs/{catalog_id}/feeds/{feed_id}", + params={"workspace_id": workspace_id, "connection_id": connection_id}, + ) + + def feed_uploads( + self, + catalog_id: str, + feed_id: str, + *, + connection_id: str, + workspace_id: str | None = None, + ) -> builtins.list[ProductFeedUpload]: + """Each run the network made of the feed.""" + return parse_list( + ProductFeedUpload, + unwrap( + self._http.get( + f"/ads/catalogs/{catalog_id}/feeds/{feed_id}/uploads", + {"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ), + ) + + def start_feed_upload( + self, + catalog_id: str, + feed_id: str, + *, + workspace_id: str, + connection_id: str, + url: str | None = None, + ) -> str: + """Fetches the feed now; the id of the run. Also needs ``publish``.""" + body: dict[str, Any] = {"workspaceId": workspace_id, "connectionId": connection_id} + if url is not None: + body["url"] = url + result = unwrap( + self._http.post(f"/ads/catalogs/{catalog_id}/feeds/{feed_id}/uploads", body) + ) + upload_id = result.get("id") if isinstance(result, dict) else None + return str(upload_id) if upload_id else "" + + def product_sets( + self, catalog_id: str, *, connection_id: str, workspace_id: str | None = None + ) -> builtins.list[ProductSet]: + """A catalog ad runs from a product set, not the whole catalog.""" + return parse_list( + ProductSet, + unwrap( + self._http.get( + f"/ads/catalogs/{catalog_id}/product-sets", + {"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ), + ) + + def create_product_set( + self, + catalog_id: str, + *, + workspace_id: str, + connection_id: str, + name: str, + filter: Mapping[str, Any] | None = None, + ) -> ProductSet: + """Without a ``filter`` the set is the whole catalog. Also needs ``publish``.""" + body: dict[str, Any] = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "name": name, + } + if filter is not None: + body["filter"] = dict(filter) + return ProductSet.model_validate( + unwrap(self._http.post(f"/ads/catalogs/{catalog_id}/product-sets", body)) + ) + + def update_product_set( + self, + catalog_id: str, + set_id: str, + *, + workspace_id: str, + connection_id: str, + name: str, + filter: Mapping[str, Any] | None = None, + ) -> ProductSet: + """Also needs ``publish``.""" + body: dict[str, Any] = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "name": name, + } + if filter is not None: + body["filter"] = dict(filter) + return ProductSet.model_validate( + unwrap( + self._http.request( + "PATCH", + f"/ads/catalogs/{catalog_id}/product-sets/{set_id}", + json=body, + params={"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ) + ) + + def delete_product_set( + self, catalog_id: str, set_id: str, *, workspace_id: str, connection_id: str + ) -> None: + """Also needs ``publish``.""" + self._http.request( + "DELETE", + f"/ads/catalogs/{catalog_id}/product-sets/{set_id}", + params={"workspace_id": workspace_id, "connection_id": connection_id}, + ) + + # ─── Reach and frequency ─────────────────────────────────────── + + def reach_frequency( + self, *, connection_id: str, ad_account_id: str, workspace_id: str | None = None + ) -> ReachFrequencyResult: + return ReachFrequencyResult.model_validate( + unwrap( + self._http.get( + "/ads/reach-frequency", + { + "workspace_id": workspace_id, + "connection_id": connection_id, + "ad_account_id": ad_account_id, + }, + ) + ) + ) + + def create_reach_frequency( + self, + *, + workspace_id: str, + connection_id: str, + ad_account_id: str, + name: str, + targeting: Mapping[str, Any], + placements: Sequence[str], + budget_minor: int, + start_at: str, + end_at: str, + frequency_cap: int | None = None, + ) -> ReachFrequencyPrediction: + """Prices a flight. Nothing is bought until you reserve it.""" + body: dict[str, Any] = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + "name": name, + "targeting": dict(targeting), + "placements": list(placements), + "budgetMinor": budget_minor, + "startAt": start_at, + "endAt": end_at, + } + if frequency_cap is not None: + body["frequencyCap"] = frequency_cap + return ReachFrequencyPrediction.model_validate( + unwrap(self._http.post("/ads/reach-frequency", body)) + ) + + def get_reach_frequency( + self, + prediction_id: str, + *, + connection_id: str, + ad_account_id: str, + workspace_id: str | None = None, + ) -> ReachFrequencyPrediction: + return ReachFrequencyPrediction.model_validate( + unwrap( + self._http.get( + f"/ads/reach-frequency/{prediction_id}", + { + "workspace_id": workspace_id, + "connection_id": connection_id, + "ad_account_id": ad_account_id, + }, + ) + ) + ) + + def reserve_reach_frequency( + self, prediction_id: str, *, workspace_id: str, connection_id: str, ad_account_id: str + ) -> ReachFrequencyPrediction: + """Holds the inventory the prediction priced. Also needs ``publish``.""" + return self._reach_frequency_action( + prediction_id, "reserve", workspace_id, connection_id, ad_account_id + ) + + def cancel_reach_frequency( + self, prediction_id: str, *, workspace_id: str, connection_id: str, ad_account_id: str + ) -> ReachFrequencyPrediction: + """Also needs ``publish``.""" + return self._reach_frequency_action( + prediction_id, "cancel", workspace_id, connection_id, ad_account_id + ) + + # ─── Ad Library ──────────────────────────────────────────────── + + def library( + self, + *, + connection_id: str, + countries: Sequence[str], + workspace_id: str | None = None, + q: str | None = None, + page_ids: Sequence[str] | None = None, + active_status: str | None = None, + limit: int | None = None, + after: str | None = None, + ) -> AdLibraryPage: + """The public ad archive: ads anyone is running, by keyword or by Page. + + Read live on every call and stored nowhere, so an ad that stops running + is simply absent from the next search. Search by ``q`` or ``page_ids``. + """ + params: dict[str, Any] = { + "workspace_id": workspace_id, + "connection_id": connection_id, + "countries": ",".join(countries), + "q": q, + "page_ids": ",".join(page_ids) if page_ids else None, + "active_status": active_status, + "limit": limit, + "after": after, + } + return AdLibraryPage.model_validate(unwrap(self._http.get("/ads/library", params))) + + # ─── Partnership ads ─────────────────────────────────────────── + + def partnership_creators( + self, *, connection_id: str, page_id: str, workspace_id: str | None = None + ) -> builtins.list[PartnershipCreator]: + """Creators who allowlisted this Page to run partnership ads on their posts.""" + return parse_list( + PartnershipCreator, + unwrap( + self._http.get( + "/ads/partnership/creators", + { + "workspace_id": workspace_id, + "connection_id": connection_id, + "page_id": page_id, + }, + ) + ), + ) + + def request_partnership( + self, *, workspace_id: str, connection_id: str, page_id: str, creator_id: str + ) -> builtins.list[PartnershipCreator]: + """Asks a creator for permission; the list as it now stands.""" + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "pageId": page_id, + "creatorId": creator_id, + } + return parse_list( + PartnershipCreator, unwrap(self._http.post("/ads/partnership/creators", body)) + ) + + def revoke_partnership( + self, creator_id: str, *, workspace_id: str, connection_id: str, page_id: str + ) -> None: + self._http.request( + "DELETE", + f"/ads/partnership/creators/{creator_id}", + params={ + "workspace_id": workspace_id, + "connection_id": connection_id, + "page_id": page_id, + }, + ) + + # ─── Ad account settings ─────────────────────────────────────── + + def account_activity( + self, + *, + connection_id: str, + ad_account_id: str, + workspace_id: str | None = None, + since: str | None = None, + until: str | None = None, + ) -> AdActivityResult: + """Who changed what on the ad account, and when. Dates are ``YYYY-MM-DD``.""" + params = self._account_params(workspace_id, connection_id, ad_account_id) + params.update({"since": since, "until": until}) + return AdActivityResult.model_validate( + unwrap(self._http.get("/ads/account/activity", params)) + ) + + def labels( + self, *, connection_id: str, ad_account_id: str, workspace_id: str | None = None + ) -> builtins.list[AdLabel]: + return parse_list( + AdLabel, + unwrap( + self._http.get( + "/ads/account/labels", + self._account_params(workspace_id, connection_id, ad_account_id), + ) + ), + ) + + def create_label( + self, *, workspace_id: str, connection_id: str, ad_account_id: str, name: str + ) -> AdLabel: + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + "name": name, + } + return AdLabel.model_validate(unwrap(self._http.post("/ads/account/labels", body))) + + def update_label( + self, + label_id: str, + *, + workspace_id: str, + connection_id: str, + ad_account_id: str, + name: str, + ) -> AdLabel: + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + "name": name, + } + return AdLabel.model_validate( + unwrap( + self._http.request( + "PATCH", + f"/ads/account/labels/{label_id}", + json=body, + params={"workspace_id": workspace_id, "connection_id": connection_id}, + ) + ) + ) + + def delete_label( + self, label_id: str, *, workspace_id: str, connection_id: str, ad_account_id: str + ) -> None: + self._http.request( + "DELETE", + f"/ads/account/labels/{label_id}", + params=self._account_params(workspace_id, connection_id, ad_account_id), + ) + + def apply_label( + self, + label_id: str, + *, + workspace_id: str, + connection_id: str, + ad_account_id: str, + object_id: str, + level: str, + ) -> None: + """Keeps whatever labels the object already carries. ``level`` is + ``campaign``, ``ad_set`` or ``ad``. + """ + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + "objectId": object_id, + "level": level, + } + self._http.post(f"/ads/account/labels/{label_id}/apply", body) + + def studies( + self, *, connection_id: str, ad_account_id: str, workspace_id: str | None = None + ) -> builtins.list[AdStudy]: + return parse_list( + AdStudy, + unwrap( + self._http.get( + "/ads/account/studies", + self._account_params(workspace_id, connection_id, ad_account_id), + ) + ), + ) + + def create_study( + self, + *, + workspace_id: str, + connection_id: str, + ad_account_id: str, + name: str, + start_at: str, + end_at: str, + cells: Sequence[Mapping[str, Any]], + description: str | None = None, + ) -> AdStudy: + """Splits traffic evenly across two to five ``cells`` of ``name`` and ``objectIds``.""" + body: dict[str, Any] = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + "name": name, + "startAt": start_at, + "endAt": end_at, + "cells": [dict(c) for c in cells], + } + if description is not None: + body["description"] = description + return AdStudy.model_validate(unwrap(self._http.post("/ads/account/studies", body))) + + def get_study( + self, + study_id: str, + *, + connection_id: str, + ad_account_id: str, + workspace_id: str | None = None, + ) -> AdStudy: + return AdStudy.model_validate( + unwrap( + self._http.get( + f"/ads/account/studies/{study_id}", + self._account_params(workspace_id, connection_id, ad_account_id), + ) + ) + ) + + def delete_study( + self, study_id: str, *, workspace_id: str, connection_id: str, ad_account_id: str + ) -> None: + self._http.request( + "DELETE", + f"/ads/account/studies/{study_id}", + params=self._account_params(workspace_id, connection_id, ad_account_id), + ) + + def ios_campaign_limits( + self, *, connection_id: str, ad_account_id: str, workspace_id: str | None = None + ) -> builtins.list[IosCampaignLimits]: + """How many iOS 14 campaigns the account may run at once, per app.""" + return parse_list( + IosCampaignLimits, + unwrap( + self._http.get( + "/ads/account/ios-limits", + self._account_params(workspace_id, connection_id, ad_account_id), + ) + ), + ) + + def high_demand_periods( + self, *, connection_id: str, ad_account_id: str, workspace_id: str | None = None + ) -> builtins.list[HighDemandPeriod]: + return parse_list( + HighDemandPeriod, + unwrap( + self._http.get( + "/ads/account/high-demand-periods", + self._account_params(workspace_id, connection_id, ad_account_id), + ) + ), + ) + + def create_high_demand_period( + self, + *, + workspace_id: str, + connection_id: str, + ad_account_id: str, + start_at: str, + end_at: str, + budget_value: float, + budget_value_type: str, + ) -> HighDemandPeriod: + """Tells the network to expect heavier spend over a window, so pacing allows for it. + + ``budget_value_type`` is ``ABSOLUTE`` or ``MULTIPLIER``. + """ + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + "startAt": start_at, + "endAt": end_at, + "budgetValue": budget_value, + "budgetValueType": budget_value_type, + } + return HighDemandPeriod.model_validate( + unwrap(self._http.post("/ads/account/high-demand-periods", body)) + ) + + def delete_high_demand_period( + self, period_id: str, *, workspace_id: str, connection_id: str, ad_account_id: str + ) -> None: + self._http.request( + "DELETE", + f"/ads/account/high-demand-periods/{period_id}", + params=self._account_params(workspace_id, connection_id, ad_account_id), + ) + + def value_rule_sets( + self, *, connection_id: str, ad_account_id: str, workspace_id: str | None = None + ) -> builtins.list[ValueRuleSet]: + return parse_list( + ValueRuleSet, + unwrap( + self._http.get( + "/ads/account/value-rule-sets", + self._account_params(workspace_id, connection_id, ad_account_id), + ) + ), + ) + + def create_value_rule_set( + self, + *, + workspace_id: str, + connection_id: str, + ad_account_id: str, + name: str, + rules: Sequence[Mapping[str, Any]], + ) -> ValueRuleSet: + """Weights conversions so some audiences count for more than others.""" + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + "name": name, + "rules": [dict(r) for r in rules], + } + return ValueRuleSet.model_validate( + unwrap(self._http.post("/ads/account/value-rule-sets", body)) + ) + + def delete_value_rule_set( + self, rule_set_id: str, *, workspace_id: str, connection_id: str, ad_account_id: str + ) -> None: + self._http.request( + "DELETE", + f"/ads/account/value-rule-sets/{rule_set_id}", + params=self._account_params(workspace_id, connection_id, ad_account_id), + ) + + @staticmethod + def _account_params( + workspace_id: str | None, connection_id: str, ad_account_id: str + ) -> dict[str, Any]: + return { + "workspace_id": workspace_id, + "connection_id": connection_id, + "ad_account_id": ad_account_id, + } + + def _reach_frequency_action( + self, + prediction_id: str, + action: str, + workspace_id: str, + connection_id: str, + ad_account_id: str, + ) -> ReachFrequencyPrediction: + body = { + "workspaceId": workspace_id, + "connectionId": connection_id, + "adAccountId": ad_account_id, + } + return ReachFrequencyPrediction.model_validate( + unwrap(self._http.post(f"/ads/reach-frequency/{prediction_id}/{action}", body)) + ) + def _object( self, method: str,