Providers facebook hook multiple account

airflow/providers/facebook/ads/hooks/ads.py

@@ -18,12 +18,13 @@
 """This module contains Facebook Ads Reporting hooks"""
 import time
 from enum import Enum
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 try:
     from functools import cached_property
 except ImportError:
     from cached_property import cached_property
+
 from facebook_business.adobjects.adaccount import AdAccount
 from facebook_business.adobjects.adreportrun import AdReportRun
 from facebook_business.adobjects.adsinsights import AdsInsights
@@ -81,10 +82,14 @@ def _get_service(self) -> FacebookAdsApi:
             app_id=config["app_id"],
             app_secret=config["app_secret"],
             access_token=config["access_token"],
-            account_id=config["account_id"],
             api_version=self.api_version,
         )
 
+    @cached_property
+    def multiple_accounts(self) -> bool:
+        """Checks whether provided account_id in the Facebook Ads Connection is provided as a list"""
+        return isinstance(self.facebook_ads_config["account_id"], list)
+
     @cached_property
     def facebook_ads_config(self) -> Dict:
         """
@@ -105,24 +110,74 @@ def bulk_facebook_report(
         params: Dict[str, Any],
         fields: List[str],
         sleep_time: int = 5,
-    ) -> List[AdsInsights]:
-        """
-        Pulls data from the Facebook Ads API
+    ) -> Union[List[AdsInsights], Dict[str, List[AdsInsights]]]:
+        """Pulls data from the Facebook Ads API regarding Account ID with matching return type.
+
+        The return type and value depends on the ``account_id`` configuration. If the
+        configuration is a str representing a single Account ID, the return value is the
+        list of reports for that ID. If the configuration is a list of str representing
+        multiple Account IDs, the return value is a dict of Account IDs and their
+        respective list of reports.
 
         :param fields: List of fields that is obtained from Facebook. Found in AdsInsights.Field class.
             https://developers.facebook.com/docs/marketing-api/insights/parameters/v6.0
         :type fields: List[str]
         :param params: Parameters that determine the query for Facebook
             https://developers.facebook.com/docs/marketing-api/insights/parameters/v6.0
-        :type fields: Dict[str, Any]
+        :type params: Dict[str, Any]
         :param sleep_time: Time to sleep when async call is happening
         :type sleep_time: int
 
-        :return: Facebook Ads API response, converted to Facebook Ads Row objects
-        :rtype: List[AdsInsights]
+        :return: Facebook Ads API response,
+            converted to Facebook Ads Row objects regarding given Account ID type
+        :rtype: List[AdsInsights] or Dict[str, List[AdsInsights]]
         """
         api = self._get_service()
-        ad_account = AdAccount(api.get_default_account_id(), api=api)
+        if self.multiple_accounts:
+            all_insights = {}
+            for account_id in self.facebook_ads_config["account_id"]:
+                all_insights[account_id] = self._facebook_report(
+                    account_id=account_id, api=api, params=params, fields=fields, sleep_time=sleep_time
+                )
+                self.log.info(
+                    "%s Account Id used to extract data from Facebook Ads Iterators successfully", account_id
+                )
+            return all_insights
+        else:
+            return self._facebook_report(
+                account_id=self.facebook_ads_config["account_id"],
+                api=api,
+                params=params,
+                fields=fields,
+                sleep_time=sleep_time,
+            )
+
+    def _facebook_report(
+        self,
+        account_id: str,
+        api: FacebookAdsApi,
+        params: Dict[str, Any],
+        fields: List[str],
+        sleep_time: int = 5,
+    ) -> List[AdsInsights]:
+        """
+        Pulls data from the Facebook Ads API with given account_id
+
+        :param account_id: Facebook Account ID that holds ads information
+                https://developers.facebook.com/docs/marketing-api/reference/ads-insights/
+        :type account_id: str
+        :param api: FacebookAdsApi created in the hook
+        :type api: FacebookAdsApi
+        :param fields: List of fields that is obtained from Facebook. Found in AdsInsights.Field class.
+            https://developers.facebook.com/docs/marketing-api/insights/parameters/v6.0
+        :type fields: List[str]
+        :param params: Parameters that determine the query for Facebook
+            https://developers.facebook.com/docs/marketing-api/insights/parameters/v6.0
+        :type params: Dict[str, Any]
+        :param sleep_time: Time to sleep when async call is happening
+        :type sleep_time: int
+        """
+        ad_account = AdAccount(account_id, api=api)
         _async = ad_account.get_insights(params=params, fields=fields, is_async=True)
         while True:
             request = _async.api_get()
@@ -138,6 +193,6 @@ def bulk_facebook_report(
             time.sleep(sleep_time)
         report_run_id = _async.api_get()["report_run_id"]
         report_object = AdReportRun(report_run_id, api=api)
-        insights = report_object.get_insights()
         self.log.info("Extracting data from returned Facebook Ads Iterators")
+        insights = report_object.get_insights()
         return list(insights)

airflow/providers/google/cloud/transfers/facebook_ads_to_gcs.py

@@ -19,14 +19,24 @@
 import csv
 import tempfile
 import warnings
+from enum import Enum
 from typing import Any, Dict, List, Optional, Sequence, Union
 
+from facebook_business.adobjects.adsinsights import AdsInsights
+
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator
 from airflow.providers.facebook.ads.hooks.ads import FacebookAdsReportingHook
 from airflow.providers.google.cloud.hooks.gcs import GCSHook
 
 
+class FlushAction(Enum):
+    """Facebook Ads Export Options"""
+
+    EXPORT_ONCE = "ExportAtOnce"
+    EXPORT_EVERY_ACCOUNT = "ExportEveryAccount"
+
+
 class FacebookAdsReportToGcsOperator(BaseOperator):
     """
     Fetches the results from the Facebook Ads API as desired in the params
@@ -68,6 +78,11 @@ class FacebookAdsReportToGcsOperator(BaseOperator):
     :type parameters: Dict[str, Any]
     :param gzip: Option to compress local file or file data for upload
     :type gzip: bool
+    :param upload_as_account: Option to export file with account_id
+        This parameter only works if Account Id sets as array in Facebook Connection
+        If set as True, each file will be exported in a separate file that has a prefix of account_id
+        If set as False, a single file will be exported for all account_id
+    :type upload_as_account: bool
     :param impersonation_chain: Optional service account to impersonate using short-term
         credentials, or chained list of accounts required to get the access_token
         of the last account in the list, which will be impersonated in the request.
@@ -96,6 +111,7 @@ def __init__(
         params: Dict[str, Any] = None,
         parameters: Dict[str, Any] = None,
         gzip: bool = False,
+        upload_as_account: bool = False,
         api_version: Optional[str] = None,
         gcp_conn_id: str = "google_cloud_default",
         facebook_conn_id: str = "facebook_default",
@@ -111,6 +127,7 @@ def __init__(
         self.fields = fields
         self.parameters = parameters
         self.gzip = gzip
+        self.upload_as_account = upload_as_account
         self.impersonation_chain = impersonation_chain
 
         if params is None and parameters is None:
@@ -128,11 +145,85 @@ def execute(self, context: dict):
         service = FacebookAdsReportingHook(
             facebook_conn_id=self.facebook_conn_id, api_version=self.api_version
         )
-        rows = service.bulk_facebook_report(params=self.parameters, fields=self.fields)
+        bulk_report = service.bulk_facebook_report(params=self.parameters, fields=self.fields)
 
+        if isinstance(bulk_report, list):
+            converted_rows_with_action = self._generate_rows_with_action(False)
+            converted_rows_with_action = self._prepare_rows_for_upload(
+                rows=bulk_report, converted_rows_with_action=converted_rows_with_action, account_id=None
+            )
+        elif isinstance(bulk_report, dict):
+            converted_rows_with_action = self._generate_rows_with_action(True)
+            for account_id in bulk_report.keys():
+                rows = bulk_report.get(account_id, [])
+                if rows:
+                    converted_rows_with_action = self._prepare_rows_for_upload(
+                        rows=rows,
+                        converted_rows_with_action=converted_rows_with_action,
+                        account_id=account_id,
+                    )
+                else:
+                    self.log.warning("account_id: %s returned empty report", str(account_id))
+        else:
+            message = (
+                "Facebook Ads Hook returned different type than expected. Expected return types should be "
+                "List or Dict. Actual return type of the Hook: " + str(type(bulk_report))
+            )
+            raise AirflowException(message)
+        total_row_count = self._decide_and_flush(converted_rows_with_action=converted_rows_with_action)
+        self.log.info("Facebook Returned %s data points in total: ", total_row_count)
+
+    def _generate_rows_with_action(self, type_check: bool):
+        if type_check and self.upload_as_account:
+            return {FlushAction.EXPORT_EVERY_ACCOUNT: []}
+        else:
+            return {FlushAction.EXPORT_ONCE: []}
+
+    def _prepare_rows_for_upload(
+        self,
+        rows: List[AdsInsights],
+        converted_rows_with_action: Dict[FlushAction, list],
+        account_id: Optional[str],
+    ):
         converted_rows = [dict(row) for row in rows]
-        self.log.info("Facebook Returned %s data points", len(converted_rows))
+        if account_id is not None and self.upload_as_account:
+            converted_rows_with_action[FlushAction.EXPORT_EVERY_ACCOUNT].append(
+                {"account_id": account_id, "converted_rows": converted_rows}
+            )
+            self.log.info(
+                "Facebook Returned %s data points for account_id: %s", len(converted_rows), account_id
+            )
+        else:
+            converted_rows_with_action[FlushAction.EXPORT_ONCE].extend(converted_rows)
+            self.log.info("Facebook Returned %s data points ", len(converted_rows))
+        return converted_rows_with_action
+
+    def _decide_and_flush(self, converted_rows_with_action: Dict[FlushAction, list]):
+        total_data_count = 0
+        if FlushAction.EXPORT_ONCE in converted_rows_with_action:
+            self._flush_rows(
+                converted_rows=converted_rows_with_action.get(FlushAction.EXPORT_ONCE),
+                object_name=self.object_name,
+            )
+            total_data_count += len(converted_rows_with_action.get(FlushAction.EXPORT_ONCE))
+        elif FlushAction.EXPORT_EVERY_ACCOUNT in converted_rows_with_action:
+            for converted_rows in converted_rows_with_action.get(FlushAction.EXPORT_EVERY_ACCOUNT):
+                self._flush_rows(
+                    converted_rows=converted_rows.get("converted_rows"),
+                    object_name=self._transform_object_name_with_account_id(
+                        account_id=converted_rows.get("account_id")
+                    ),
+                )
+                total_data_count += len(converted_rows.get("converted_rows"))
+        else:
+            message = (
+                "FlushAction not found in the data. Please check the FlushAction in the operator. Converted "
+                "Rows with Action: " + str(converted_rows_with_action)
+            )
+            raise AirflowException(message)
+        return total_data_count
 
+    def _flush_rows(self, converted_rows: list, object_name: str):
         if converted_rows:
             headers = converted_rows[0].keys()
             with tempfile.NamedTemporaryFile("w", suffix=".csv") as csvfile:
@@ -146,8 +237,15 @@ def execute(self, context: dict):
                 )
                 hook.upload(
                     bucket_name=self.bucket_name,
-                    object_name=self.object_name,
+                    object_name=object_name,
                     filename=csvfile.name,
                     gzip=self.gzip,
                 )
                 self.log.info("%s uploaded to GCS", csvfile.name)
+
+    def _transform_object_name_with_account_id(self, account_id: str):
+        directory_parts = self.object_name.split("/")
+        directory_parts[len(directory_parts) - 1] = (
+            account_id + "_" + directory_parts[len(directory_parts) - 1]
+        )
+        return "/".join(directory_parts)

tests/providers/facebook/ads/hooks/test_ads.py

@@ -23,6 +23,14 @@
 
 API_VERSION = "api_version"
 EXTRAS = {"account_id": "act_12345", "app_id": "12345", "app_secret": "1fg444", "access_token": "Ab35gf7E"}
+EXTRAS_MULTIPLE = {
+    "account_id": ["act_12345", "act_12346"],
+    "app_id": "12345",
+    "app_secret": "1fg444",
+    "access_token": "Ab35gf7E",
+}
+ACCOUNT_ID_1 = "act_12345"
+ACCOUNT_ID_2 = "act_12346"
 FIELDS = [
     "campaign_name",
     "campaign_id",
@@ -41,6 +49,14 @@ def mock_hook():
         yield hook
 
 
+@pytest.fixture()
+def mock_hook_multiple():
+    with mock.patch("airflow.hooks.base.BaseHook.get_connection") as conn:
+        hook = FacebookAdsReportingHook(api_version=API_VERSION)
+        conn.return_value.extra_dejson = EXTRAS_MULTIPLE
+        yield hook
+
+
 class TestFacebookAdsReportingHook:
     @mock.patch("airflow.providers.facebook.ads.hooks.ads.FacebookAdsApi")
     def test_get_service(self, mock_api, mock_hook):
@@ -50,7 +66,6 @@ def test_get_service(self, mock_api, mock_hook):
             app_id=EXTRAS["app_id"],
             app_secret=EXTRAS["app_secret"],
             access_token=EXTRAS["access_token"],
-            account_id=EXTRAS["account_id"],
             api_version=API_VERSION,
         )
 
@@ -65,6 +80,26 @@ def test_bulk_facebook_report(self, mock_client, mock_ad_account, mock_hook):
             "async_percent_completion": 100,
         }
         mock_hook.bulk_facebook_report(params=PARAMS, fields=FIELDS)
-        mock_ad_account.assert_has_calls([mock.call(mock_client.get_default_account_id(), api=mock_client)])
+        mock_ad_account.assert_has_calls([mock.call(ACCOUNT_ID_1, api=mock_client)])
         ad_account.assert_called_once_with(params=PARAMS, fields=FIELDS, is_async=True)
         ad_account.return_value.api_get.assert_has_calls([mock.call(), mock.call()])
+
+    @mock.patch("airflow.providers.facebook.ads.hooks.ads.AdAccount")
+    @mock.patch("airflow.providers.facebook.ads.hooks.ads.FacebookAdsApi")
+    def test_bulk_facebook_report_multiple_account_id(self, mock_client, mock_ad_account, mock_hook_multiple):
+        mock_client = mock_client.init()
+        ad_account = mock_ad_account().get_insights
+        ad_account.return_value.api_get.return_value = {
+            "async_status": "Job Completed",
+            "report_run_id": "12345",
+            "async_percent_completion": 100,
+        }
+        mock_hook_multiple.bulk_facebook_report(params=PARAMS, fields=FIELDS)
+        mock_ad_account.assert_has_calls(
+            [mock.call(ACCOUNT_ID_1, api=mock_client)], [mock.call(ACCOUNT_ID_2, api=mock_client)]
+        )
+        ad_account.assert_has_calls(
+            [mock.call(params=PARAMS, fields=FIELDS, is_async=True)],
+            [mock.call(params=PARAMS, fields=FIELDS, is_async=True)],
+        )
+        ad_account.return_value.api_get.assert_has_calls([mock.call(), mock.call()])