Add system test and docs for Facebook Ads operators (#8503)

apache · May 3, 2020 · bc45fa6 · bc45fa6
1 parent ac59735
commit bc45fa6
Show file tree

Hide file tree

Showing 15 changed files with 194 additions and 128 deletions.
diff --git a/airflow/providers/facebook/ads/hooks/ads.py b/airflow/providers/facebook/ads/hooks/ads.py
@@ -74,27 +74,26 @@ def __init__(
     def _get_service(self) -> FacebookAdsApi:
         """ Returns Facebook Ads Client using a service account"""
         config = self.facebook_ads_config
-        missings = [_each for _each in self.client_required_fields if _each not in config]
-        if missings:
-            message = "{missings} fields are missing".format(missings=missings)
-            raise AirflowException(message)
         return FacebookAdsApi.init(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 facebook_ads_config(self) -> None:
+    def facebook_ads_config(self) -> Dict:
         """
         Gets Facebook ads connection from meta db and sets
         facebook_ads_config attribute with returned config file
         """
         self.log.info("Fetching fb connection: %s", self.facebook_conn_id)
         conn = self.get_connection(self.facebook_conn_id)
-        if "facebook_ads_client" not in conn.extra_dejson:
-            raise AirflowException("facebook_ads_client not found")
-        return conn.extra_dejson["facebook_ads_client"]
+        config = conn.extra_dejson
+        missings_keys = self.client_required_fields - config.keys()
+        if missings_keys:
+            message = "{missings_keys} fields are missing".format(missings_keys=missings_keys)
+            raise AirflowException(message)
+        return config
 
     def bulk_facebook_report(
         self,

diff --git a/...ok_ads_to_gcs/example_dags/example_ads.py → ...ample_dags/example_facebook_ads_to_gcs.py b/...ok_ads_to_gcs/example_dags/example_ads.py → ...ample_dags/example_facebook_ads_to_gcs.py
@@ -24,10 +24,12 @@
 
 from airflow import models
 from airflow.providers.google.cloud.operators.bigquery import (
-    BigQueryCreateEmptyDatasetOperator, BigQueryCreateEmptyTableOperator, BigQueryExecuteQueryOperator,
+    BigQueryCreateEmptyDatasetOperator, BigQueryCreateEmptyTableOperator, BigQueryDeleteDatasetOperator,
+    BigQueryExecuteQueryOperator,
 )
+from airflow.providers.google.cloud.operators.facebook_ads_to_gcs import FacebookAdsReportToGcsOperator
+from airflow.providers.google.cloud.operators.gcs import GCSCreateBucketOperator, GCSDeleteBucketOperator
 from airflow.providers.google.cloud.operators.gcs_to_bigquery import GCSToBigQueryOperator
-from airflow.providers.google.facebook_ads_to_gcs.operators.ads import FacebookAdsReportToGcsOperator
 from airflow.utils.dates import days_ago
 
 # [START howto_GCS_env_variables]
@@ -56,13 +58,21 @@
 default_args = {"start_date": days_ago(1)}
 
 with models.DAG(
-    "example_fb_operator",
+    "example_facebook_ads_to_gcs",
     default_args=default_args,
     schedule_interval=None,  # Override to match your needs
 ) as dag:
 
-    create_dataset = BigQueryCreateEmptyDatasetOperator(task_id="create-dataset",
-                                                        dataset_id=DATASET_NAME)
+    create_bucket = GCSCreateBucketOperator(
+        task_id="create_bucket",
+        bucket_name=GCS_BUCKET,
+        project_id=GCP_PROJECT_ID,
+    )
+
+    create_dataset = BigQueryCreateEmptyDatasetOperator(
+        task_id="create_dataset",
+        dataset_id=DATASET_NAME,
+    )
 
     create_table = BigQueryCreateEmptyTableOperator(
         task_id="create_table",
@@ -77,7 +87,7 @@
         ],
     )
 
-    # [START howto_FB_ADS_to_gcs_operator]
+    # [START howto_operator_facebook_ads_to_gcs]
     run_operator = FacebookAdsReportToGcsOperator(
         task_id='run_fetch_data',
         start_date=days_ago(2),
@@ -88,23 +98,33 @@
         gcp_conn_id=GCS_CONN_ID,
         object_name=GCS_OBJ_PATH,
     )
-    # [END howto_FB_ADS_to_gcs_operator]
+    # [END howto_operator_facebook_ads_to_gcs]
 
-    # [START howto_operator_gcs_to_bq]
     load_csv = GCSToBigQueryOperator(
         task_id='gcs_to_bq_example',
         bucket=GCS_BUCKET,
         source_objects=[GCS_OBJ_PATH],
         destination_project_dataset_table=f"{DATASET_NAME}.{TABLE_NAME}",
-        write_disposition='WRITE_TRUNCATE')
-    # [END howto_operator_gcs_to_bq]
+        write_disposition='WRITE_TRUNCATE'
+    )
 
-    # [START howto_operator_read_data_from_gcs]
     read_data_from_gcs_many_chunks = BigQueryExecuteQueryOperator(
         task_id="read_data_from_gcs_many_chunks",
         sql=f"SELECT COUNT(*) FROM `{GCP_PROJECT_ID}.{DATASET_NAME}.{TABLE_NAME}`",
         use_legacy_sql=False,
     )
-    # [END howto_operator_read_data_from_gcs]
 
-    create_dataset >> create_table >> run_operator >> load_csv >> read_data_from_gcs_many_chunks
+    delete_bucket = GCSDeleteBucketOperator(
+        task_id="delete_bucket",
+        bucket_name=GCS_BUCKET,
+    )
+
+    delete_dataset = BigQueryDeleteDatasetOperator(
+        task_id="delete_dataset",
+        project_id=GCP_PROJECT_ID,
+        dataset_id=DATASET_NAME,
+        delete_contents=True,
+    )
+
+    create_bucket >> create_dataset >> create_table >> run_operator >> load_csv
+    load_csv >> read_data_from_gcs_many_chunks >> delete_bucket >> delete_dataset
diff --git a/...ogle/facebook_ads_to_gcs/operators/ads.py → ...le/cloud/operators/facebook_ads_to_gcs.py b/...ogle/facebook_ads_to_gcs/operators/ads.py → ...le/cloud/operators/facebook_ads_to_gcs.py
@@ -42,6 +42,10 @@ class FacebookAdsReportToGcsOperator(BaseOperator):
         For more information on the Facebook Ads Python SDK, take a look at the docs:
         https://github.com/facebook/facebook-python-business-sdk
 
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:FacebookAdsReportToGcsOperator`
+
     :param bucket: The GCS bucket to upload to
     :type bucket: str
     :param obj: GCS path to save the object. Must be the full file path (ex. `path/to/file.txt`)

diff --git a/airflow/providers/google/facebook_ads_to_gcs/__init__.py b/airflow/providers/google/facebook_ads_to_gcs/__init__.py
diff --git a/airflow/providers/google/facebook_ads_to_gcs/example_dags/__init__.py b/airflow/providers/google/facebook_ads_to_gcs/example_dags/__init__.py
diff --git a/airflow/providers/google/facebook_ads_to_gcs/operators/__init__.py b/airflow/providers/google/facebook_ads_to_gcs/operators/__init__.py
diff --git a/airflow/utils/db.py b/airflow/utils/db.py
@@ -245,15 +245,12 @@ def create_default_connections(session=None):
         Connection(
             conn_id="facebook_default",
             conn_type="facebook_social",
-            schema="""
-            {
-                "facebook_ads_client": {
-                    "account_id": "act_123456789",
-                    "app_id": "1234567890",
-                    "app_secret": "1f45tghxxxx12345",
-                    "access_token": "ABcdEfghiJKlmnoxxyz"
+            extra="""
+                {   "account_id": "<AD_ACCOUNNT_ID>",
+                    "app_id": "<FACEBOOK_APP_ID>",
+                    "app_secret": "<FACEBOOK_APP_SECRET>",
+                    "access_token": "<FACEBOOK_AD_ACCESS_TOKEN>"
                 }
-            }
             """,
         ),
         session

diff --git a/docs/autoapi_templates/index.rst b/docs/autoapi_templates/index.rst
@@ -110,8 +110,6 @@ All operators are in the following packages:
 
   airflow/providers/exasol/operators/index
 
-  airflow/providers/google/facebook_ads_to_gcs/operators/index
-
   airflow/providers/ftp/sensors/index
 
   airflow/providers/google/ads/operators/index

diff --git a/docs/howto/operator/gcp/facebook_ads_to_gcs.rst b/docs/howto/operator/gcp/facebook_ads_to_gcs.rst
@@ -0,0 +1,52 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+
+Facebook Ads To GCS Operators
+==============================
+
+.. contents::
+  :depth: 1
+  :local:
+
+Prerequisite Tasks
+^^^^^^^^^^^^^^^^^^
+
+.. include:: _partials/prerequisite_tasks.rst
+
+.. _howto/operator:FacebookAdsReportToGcsOperator:
+
+FacebookAdsReportToGcsOperator
+------------------------------
+
+Use the
+:class:`~airflow.providers.google.cloud.operators.facebook_ads_to_gcs.FacebookAdsReportToGcsOperator`
+to execute a Facebook ads report fetch and load to GCS.
+
+.. exampleinclude:: ../../../../airflow/providers/google/cloud/example_dags/example_facebook_ads_to_gcs.py
+    :language: python
+    :start-after: [START howto_operator_facebook_ads_to_gcs]
+    :end-before: [END howto_operator_facebook_ads_to_gcs]
+
+Reference
+^^^^^^^^^
+
+For further information, look at:
+
+* `Client Library Documentation <https://github.com/facebook/facebook-python-business-sdk>`__
+* `Product Documentation <https://developers.facebook.com/docs/business-manager-api/>`__
diff --git a/docs/operators-and-hooks-ref.rst b/docs/operators-and-hooks-ref.rst
@@ -806,8 +806,8 @@ These integrations allow you to copy data from/to Google Cloud Platform.
 
    * - `Facebook Ads <http://business.facebook.com>`__
      - `Google Cloud Storage (GCS) <https://cloud.google.com/gcs/>`__
-     -
-     - :mod:`airflow.providers.google.facebook_ads_to_gcs.operators.ads`
+     - :doc:`How to use <howto/operator/gcp/facebook_ads_to_gcs>`
+     - :mod:`airflow.providers.google.cloud.operators.facebook_ads_to_gcs`
 
    * - `Google BigQuery <https://cloud.google.com/bigquery/>`__
      - `MySQL <https://www.mysql.com/>`__

diff --git a/tests/providers/facebook/ads/hooks/test_ads.py b/tests/providers/facebook/ads/hooks/test_ads.py
@@ -22,12 +22,10 @@
 
 API_VERSION = "api_version"
 EXTRAS = {
-    "facebook_ads_client": {
-        "account_id": "act_12345",
-        "app_id": "12345",
-        "app_secret": "1fg444",
-        "access_token": "Ab35gf7E"
-    }
+    "account_id": "act_12345",
+    "app_id": "12345",
+    "app_secret": "1fg444",
+    "access_token": "Ab35gf7E"
 }
 FIELDS = [
     "campaign_name",
@@ -55,11 +53,10 @@ class TestFacebookAdsReportingHook:
     def test_get_service(self, mock_api, mock_hook):
         mock_hook._get_service()
         api = mock_api.init
-        creds = EXTRAS["facebook_ads_client"]
-        api.assert_called_once_with(app_id=creds["app_id"],
-                                    app_secret=creds["app_secret"],
-                                    access_token=creds["access_token"],
-                                    account_id=creds["account_id"],
+        api.assert_called_once_with(app_id=EXTRAS["app_id"],
+                                    app_secret=EXTRAS["app_secret"],
+                                    access_token=EXTRAS["access_token"],
+                                    account_id=EXTRAS["account_id"],
                                     api_version=API_VERSION)
 
     @mock.patch("airflow.providers.facebook.ads.hooks.ads.AdAccount")

diff --git a/...facebook_ads_to_gcs/operators/test_ads.py → ...oud/operators/test_facebook_ads_to_gcs.py b/...facebook_ads_to_gcs/operators/test_ads.py → ...oud/operators/test_facebook_ads_to_gcs.py
@@ -16,7 +16,7 @@
 # under the License.
 from unittest import mock
 
-from airflow.providers.google.facebook_ads_to_gcs.operators.ads import FacebookAdsReportToGcsOperator
+from airflow.providers.google.cloud.operators.facebook_ads_to_gcs import FacebookAdsReportToGcsOperator
 
 GCS_BUCKET = "airflow_bucket_fb"
 GCS_OBJ_PATH = "Temp/this_is_my_report_json.json"
@@ -47,8 +47,8 @@
 
 class TestFacebookAdsReportToGcsOperator:
 
-    @mock.patch("airflow.providers.google.facebook_ads_to_gcs.operators.ads.FacebookAdsReportingHook")
-    @mock.patch("airflow.providers.google.facebook_ads_to_gcs.operators.ads.GCSHook")
+    @mock.patch("airflow.providers.google.cloud.operators.facebook_ads_to_gcs.FacebookAdsReportingHook")
+    @mock.patch("airflow.providers.google.cloud.operators.facebook_ads_to_gcs.GCSHook")
     def test_execute(self, mock_gcs_hook, mock_ads_hook):
         mock_ads_hook.return_value.bulk_facebook_report.return_value = FACEBOOK_RETURN_VALUE
         op = FacebookAdsReportToGcsOperator(facebook_conn_id=FACEBOOK_ADS_CONN_ID,