source-facebook-marketing: attempting to allow for multiple acc…

estuary-cdk/common.Dockerfile

@@ -1,5 +1,5 @@
 # syntax=docker/dockerfile:1
-FROM python:3.12-slim as base
+FROM python:3.11 as base
 FROM base as builder
 
 ARG CONNECTOR_NAME
@@ -18,7 +18,7 @@
 RUN poetry install
 
 
 FROM base as runner
 
 ARG CONNECTOR_NAME
 ARG CONNECTOR_TYPE
@@ -44,4 +44,4 @@
 ENV DOCS_URL=${DOCS_URL}
 ENV CONNECTOR_NAME=${CONNECTOR_NAME}
 
 CMD /opt/venv/bin/python -m $(echo "$CONNECTOR_NAME" | tr '-' '_')

source-facebook-marketing/acmeCo/flow.yaml

@@ -3,11 +3,7 @@ collections:
   acmeCo/activities:
     schema: activities.schema.yaml
     key:
-      - /object_id
       - /actor_id
-      - /application_id
-      - /event_time
-      - /event_type
   acmeCo/ad_account:
     schema: ad_account.schema.yaml
     key:

source-facebook-marketing/connector_config.yaml

@@ -1,4 +1,5 @@
-account_id_sops: ENC[AES256_GCM,data:DuMXhQYiyxxthsOHllmw,iv:wNw1RFVBe/AWvY6uycliIY3nPpzBiz+P4UBDr6f4nLI=,tag:7s2DQ+ApVrcfjMBNqOK+/w==,type:int]
+account_ids:
+    - 196765372540622
 credentials:
     access_token_sops: ENC[AES256_GCM,data:Re73N0PHhhJs7Umo5ONjzSXdb4xKe+XtOBI3r1padMyQo5Bki3bW7l+1+/v4ifQuq4RypBwxQOAXdJfarwqLMW6vKCpxzq+kcbEHkvFd+NTJIWduYl8lC5bbFHT6uTrBNMxaTv/N7LNTwohBss+iKc9kf+Dp7dwpUqikOohnvOZ9y6C4+p2lvanOUpVfw7mGL52uW8VALPiLWh1hgbn6ZXlk7MOAjZAboIzf2bCzkzZQIiTz/4NYVLcmbFbPx8DRjrUd6XU1yeLsfXbT3CNRbAJyNWwDJN9vwKErGQIKcfsXsP5hVb0rpFuSW+WBvpJ0avCR6N8vvM8I8mc5W47EMuMTAxI=,iv:hWmqMuuIUBm8Vm30DVzv5jkesBAIuxZnCzn2h/ysUZ8=,tag:uKzEZpI3M4IWiXGXIqncZQ==,type:str]
     auth_type: OAuth Credentials
@@ -28,13 +29,13 @@ custom_insights:
       level: ad
       name: ads_insights_publisher_platform
       start_date: "2021-11-01T00:00:00Z"
-      time_increment: 1
+      time_increment: 360
 fetch_thumbnail_images: false
 include_deleted: true
 insights_lookback_window: 28
 max_batch_size: 50
 page_size: 100
-start_date: "2021-11-01T00:00:00Z"
+start_date: "2024-11-01T00:00:00Z"
 sops:
     kms: []
     gcp_kms:
@@ -44,8 +45,8 @@ sops:
     azure_kv: []
     hc_vault: []
     age: []
-    lastmodified: "2024-02-05T23:41:33Z"
-    mac: ENC[AES256_GCM,data:ByAdg0VEy2xMpZXxAwa5oKB7CfG+eSA5ej/TJ2g9ZsIR+XhwJklI1XBW62VvahoGeC2AP9wYC/Mk7bP03914JZCz1v2gsbn6nL4WVuooIt0XMFgaquwWjoVrDa7Hy3iTHvr2Uwf4YcBWI/M3NwKpD/3f9TpNQRstrJq8pw6AC0k=,iv:Spd541rsKs5Si4AAdtxxB/muD9XiokkyEGBC8/wqAsg=,tag:eXTEjYrz/Vq4526D3E6whw==,type:str]
+    lastmodified: "2024-11-11T19:07:30Z"
+    mac: ENC[AES256_GCM,data:DnjkguTUM5RBYREQlUp9UJTbdQ1N8DNfTLWeF+ZYSPRzMGrfwYsAhrE/08GYy3/4mAJSOTvMVRnqc0FArZ5MxIl5ttgyEZpRWMA/Z2x7BQAlByvBO2FZcsv+PdmSbtThYO/9jLeAeCCyl6LUvguiUkHxCB4HMUdHTXbLCnQYe1w=,iv:nabT9Nmhz/hTHzM+8NLTFDNIQmciFAtpMxf9/P49b4A=,tag:qeFSf7/FvFI6md+fb7TVRg==,type:str]
     pgp: []
     encrypted_suffix: _sops
     version: 3.8.1

source-facebook-marketing/pyproject.toml

@@ -12,7 +12,7 @@ types-requests = "^2.31"
 facebook-business = "19.0.0"
 pydantic = "==1.10.14"
 cached-property = "^1.5.2"
-pendulum = "^3"
+pendulum = "==2.1.2"
 
 [tool.poetry.group.dev.dependencies]
 debugpy = "^1.8.0"

source-facebook-marketing/source_facebook_marketing/api.py

@@ -6,6 +6,7 @@
 import logging
 from dataclasses import dataclass
 from time import sleep
+from typing import List
 
 import backoff
 import pendulum
@@ -158,16 +159,18 @@ def call(
 class API:
     """Simple wrapper around Facebook API"""
 
-    def __init__(self, account_id: str, access_token: str):
-        self._account_id = account_id
+    def __init__(self, access_token: str):
+        self._account_ids = {}
         # design flaw in MyFacebookAdsApi requires such strange set of new default api instance
         self.api = MyFacebookAdsApi.init(access_token=access_token, crash_log=False, api_version="v19.0")
         FacebookAdsApi.set_default_api(self.api)
 
-    @cached_property
-    def account(self) -> AdAccount:
+    def get_account(self, account_id: str) -> AdAccount:
         """Find current account"""
-        return self._find_account(self._account_id)
+        if account_id in self._account_ids:
+            return self._account_ids[account_id]
+        self._account_ids[account_id] = self._find_account(account_id)
+        return self._account_ids[account_id]
 
     @staticmethod
     def _find_account(account_id: str) -> AdAccount:

source-facebook-marketing/source_facebook_marketing/migrations.py

@@ -0,0 +1,57 @@
+import logging
+from typing import Any, List, Mapping
+
+from airbyte_cdk.config_observation import create_connector_config_control_message
+from airbyte_cdk.entrypoint import AirbyteEntrypoint
+from airbyte_cdk.sources import Source
+from airbyte_cdk.sources.message import InMemoryMessageRepository, MessageRepository
+
+logger = logging.getLogger("airbyte_logger")
+
+
+class MigrateAccountId:
+
+    message_repository: MessageRepository = InMemoryMessageRepository()
+    migrate_from_key: str = "account_id"
+    migrate_to_key: str = "account_ids"
+
+    @classmethod
+    def should_migrate(cls, config: Mapping[str, Any]) -> bool:
+        return False if config.get(cls.migrate_to_key) else True
+
+    @classmethod
+    def transform(cls, config: Mapping[str, Any]) -> Mapping[str, Any]:
+        config[cls.migrate_to_key] = [config[cls.migrate_from_key]]
+        # return transformed config
+        return config
+
+    @classmethod
+    def modify_and_save(cls, config_path: str, source: Source, config: Mapping[str, Any]) -> Mapping[str, Any]:
+        # modify the config
+        migrated_config = cls.transform(config)
+        # save the config
+        source.write_config(migrated_config, config_path)
+        # return modified config
+        return migrated_config
+
+    @classmethod
+    def emit_control_message(cls, migrated_config: Mapping[str, Any]) -> None:
+        # add the Airbyte Control Message to message repo
+        cls.message_repository.emit_message(create_connector_config_control_message(migrated_config))
+        # emit the Airbyte Control Message from message queue to stdout
+        for message in cls.message_repository._message_queue:
+            print(message.json(exclude_unset=True))
+
+    @classmethod
+    def migrate(cls, args: List[str], source: Source) -> None:
+        # get config path
+        config_path = AirbyteEntrypoint(source).extract_config(args)
+        # proceed only if `--config` arg is provided
+        if config_path:
+            # read the existing config
+            config = source.read_config(config_path)
+            # migration check
+            if cls.should_migrate(config):
+                cls.emit_control_message(
+                    cls.modify_and_save(config_path, source, config),
+                )

source-facebook-marketing/source_facebook_marketing/source.py

@@ -66,17 +66,10 @@ def check_connection(self, logger: logging.Logger, config: Mapping[str, Any]) ->
             if config.end_date < config.start_date:
                 return False, "end_date must be equal or after start_date."
 
-            api = API(account_id=config.account_id, access_token=config.credentials.access_token)
-            logger.info(f"Select account {api.account}")
+            api = API(access_token=config.credentials.access_token)
         except (requests.exceptions.RequestException, ValidationError) as e:
             return False, e
 
-        # make sure that we have valid combination of "action_breakdowns" and "breakdowns" parameters
-        for stream in self.get_custom_insights_streams(api, config):
-            try:
-                stream.check_breakdowns()
-            except facebook_business.exceptions.FacebookRequestError as e:
-                return False, e._api_error_message
         return True, None
 
     def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
@@ -89,19 +82,21 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
         config.start_date = validate_start_date(config.start_date)
         config.end_date = validate_end_date(config.start_date, config.end_date)
 
-        api = API(account_id=config.account_id, access_token=config.credentials.access_token)
+        api = API(access_token=config.credentials.access_token)
 
         insights_args = dict(
-            api=api, start_date=config.start_date, end_date=config.end_date, insights_lookback_window=config.insights_lookback_window
+            api=api, start_date=config.start_date, end_date=config.end_date, insights_lookback_window=config.insights_lookback_window, account_ids=config.account_ids,
         )
         streams = [
             AdAccount(
                 api=api,
+                account_ids=config.account_ids,
                 source_defined_primary_key=["account_id"],
             ),
             AdSets(
                 api=api,
                 start_date=config.start_date,
+                account_ids=config.account_ids,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
@@ -111,6 +106,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             Ads(
                 api=api,
                 start_date=config.start_date,
+                account_ids=config.account_ids,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
@@ -120,19 +116,21 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             AdCreatives(
                 api=api,
                 fetch_thumbnail_images=config.fetch_thumbnail_images,
+                account_ids=config.account_ids,
                 page_size=config.page_size,
                 max_batch_size=config.max_batch_size,
                 source_defined_primary_key=["id"],
             ),
             AdsInsights(page_size=config.page_size, max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
             AdsInsightsAgeAndGender(page_size=config.page_size, max_batch_size=config.max_batch_size,source_defined_primary_key=["id"], **insights_args),
             AdsInsightsCountry(page_size=config.page_size, max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
-            AdsInsightsRegion(page_size=config.page_size, max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
-            AdsInsightsDma(page_size=config.page_size, max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
-            AdsInsightsPlatformAndDevice(page_size=config.page_size, max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
-            AdsInsightsActionType(page_size=config.page_size, max_batch_size=config.max_batch_size, source_defined_primary_key=["id"], **insights_args),
+            AdsInsightsRegion(page_size=config.page_size,  max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
+            AdsInsightsDma(page_size=config.page_size,  max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
+            AdsInsightsPlatformAndDevice(page_size=config.page_size,  max_batch_size=config.max_batch_size, source_defined_primary_key=["date_start", "account_id", "ad_id"], **insights_args),
+            AdsInsightsActionType(page_size=config.page_size,  max_batch_size=config.max_batch_size, source_defined_primary_key=["id"], **insights_args),
             Campaigns(
                 api=api,
+                account_ids=config.account_ids,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
@@ -142,13 +140,15 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             ),
             CustomConversions(
                 api=api,
+                account_ids=config.account_ids,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
                 max_batch_size=config.max_batch_size,
                 source_defined_primary_key=["id"],
             ),
             Images(
                 api=api,
+                account_ids=config.account_ids,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
@@ -158,6 +158,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             ),
             Videos(
                 api=api,
+                account_ids=config.account_ids,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
@@ -167,6 +168,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             ),
             Activities(
                 api=api,
+                account_ids=config.account_ids,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
@@ -206,6 +208,7 @@ def get_custom_insights_streams(self, api: API, config: ConnectorConfig) -> List
                 raise ValueError(mes)
             stream = AdsInsights(
                 api=api,
+                account_ids=config.account_ids,
                 name=f"Custom{insight.name}",
                 fields=list(insight_fields),
                 breakdowns=list(set(insight.breakdowns)),

source-facebook-marketing/source_facebook_marketing/spec.py

@@ -5,11 +5,11 @@
 import logging
 from datetime import datetime, timezone
 from enum import Enum
-from typing import Any, Dict, List, Optional, Type, Annotated, Union
+from typing import Any, Dict, List, Optional, Type, Annotated, Union, Set
 
 from airbyte_cdk.sources.config import BaseConfig
 from facebook_business.adobjects.adsinsights import AdsInsights
-from pydantic import BaseModel, Field, PositiveInt, ConfigDict
+from pydantic import BaseModel, Field, PositiveInt, ConfigDict, constr
 
 logger = logging.getLogger("airbyte")
 
@@ -131,8 +131,8 @@ class Config:
         def schema_extra(schema: Dict[str, Any], model: Type["ConnectorConfig"]) -> None:
             schema["properties"]["end_date"].pop("format")
 
-    account_id: str = Field(
-        title="Account ID",
+    account_ids: Set[constr(regex="^[0-9]+$")] = Field(
+        title="Account IDs",
         order=0,
         description=(
             "The Facebook Ad account ID to use when pulling data from the Facebook Marketing API."

source-facebook-marketing/source_facebook_marketing/streams/async_job.py

@@ -68,7 +68,7 @@ class Status(str, Enum):
 class AsyncJob(ABC):
     """Abstract AsyncJob base class"""
 
-    def __init__(self, api: FacebookAdsApi, interval: pendulum.Interval):
+    def __init__(self, api: FacebookAdsApi, interval):
         """Init generic async job
 
         :param api: FB API instance (to create batch, etc)
@@ -79,7 +79,7 @@ def __init__(self, api: FacebookAdsApi, interval: pendulum.Interval):
         self._attempt_number = 0
 
     @property
-    def interval(self) -> pendulum.Interval:
+    def interval(self):
         """Job identifier, in most cases start of the interval"""
         return self._interval
 

source-facebook-marketing/source_facebook_marketing/streams/async_job_manager.py

@@ -32,13 +32,14 @@ class InsightAsyncJobManager:
     # limit is not reliable indicator of async workload capability we still have to use this parameter.
     MAX_JOBS_IN_QUEUE = 100
 
-    def __init__(self, api: "API", jobs: Iterator[AsyncJob]):
+    def __init__(self, api: "API", jobs: Iterator[AsyncJob], account_id: str):
         """Init
 
         :param api:
         :param jobs:
         """
         self._api = api
+        self._account_id = account_id
         self._jobs = iter(jobs)
         self._running_jobs = []
 
@@ -147,4 +148,4 @@ def _update_api_throttle_limit(self):
         respond with empty list of data so api use "x-fb-ads-insights-throttle"
         header to update current insights throttle limit.
         """
-        self._api.account.get_insights()
+        self._api.get_account(account_id=self._account_id).get_insights()