You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ms...@apache.org on 2021/09/11 04:08:18 UTC
[airflow] branch main updated: Use parameters instead of params
(#18143)
This is an automated email from the ASF dual-hosted git repository.
msumit pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 9140ad8 Use parameters instead of params (#18143)
9140ad8 is described below
commit 9140ad8d8f6dadd56bc592f5cdbf5585c2a8ce89
Author: Sumit Maheshwari <ms...@users.noreply.github.com>
AuthorDate: Sat Sep 11 09:37:46 2021 +0530
Use parameters instead of params (#18143)
---
.../example_dags/example_facebook_ads_to_gcs.py | 4 +--
.../google/cloud/transfers/facebook_ads_to_gcs.py | 27 +++++++++++++++++---
.../example_dags/example_display_video.py | 4 +--
.../marketing_platform/operators/display_video.py | 29 +++++++++++++++++-----
.../cloud/transfers/test_facebook_ads_to_gcs.py | 8 +++---
.../operators/test_display_video.py | 6 ++---
6 files changed, 58 insertions(+), 20 deletions(-)
diff --git a/airflow/providers/google/cloud/example_dags/example_facebook_ads_to_gcs.py b/airflow/providers/google/cloud/example_dags/example_facebook_ads_to_gcs.py
index 0ffe21c..9b6ac50 100644
--- a/airflow/providers/google/cloud/example_dags/example_facebook_ads_to_gcs.py
+++ b/airflow/providers/google/cloud/example_dags/example_facebook_ads_to_gcs.py
@@ -51,7 +51,7 @@ FIELDS = [
AdsInsights.Field.clicks,
AdsInsights.Field.impressions,
]
-PARAMS = {'level': 'ad', 'date_preset': 'yesterday'}
+PARAMETERS = {'level': 'ad', 'date_preset': 'yesterday'}
# [END howto_FB_ADS_variables]
with models.DAG(
@@ -90,7 +90,7 @@ with models.DAG(
start_date=days_ago(2),
owner='airflow',
bucket_name=GCS_BUCKET,
- params=PARAMS,
+ parameters=PARAMETERS,
fields=FIELDS,
gcp_conn_id=GCS_CONN_ID,
object_name=GCS_OBJ_PATH,
diff --git a/airflow/providers/google/cloud/transfers/facebook_ads_to_gcs.py b/airflow/providers/google/cloud/transfers/facebook_ads_to_gcs.py
index ed13fd6..7abee35 100644
--- a/airflow/providers/google/cloud/transfers/facebook_ads_to_gcs.py
+++ b/airflow/providers/google/cloud/transfers/facebook_ads_to_gcs.py
@@ -18,8 +18,10 @@
"""This module contains Facebook Ad Reporting to GCS operators."""
import csv
import tempfile
+import warnings
from typing import Any, Dict, List, Optional, Sequence, Union
+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
@@ -56,9 +58,13 @@ class FacebookAdsReportToGcsOperator(BaseOperator):
: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
+ :param params: Parameters that determine the query for Facebook. This keyword is deprecated,
+ please use `parameters` keyword to pass the parameters.
https://developers.facebook.com/docs/marketing-api/insights/parameters/v6.0
:type params: Dict[str, Any]
+ :param parameters: Parameters that determine the query for Facebook
+ https://developers.facebook.com/docs/marketing-api/insights/parameters/v6.0
+ :type parameters: Dict[str, Any]
:param gzip: Option to compress local file or file data for upload
:type gzip: bool
:param impersonation_chain: Optional service account to impersonate using short-term
@@ -77,6 +83,7 @@ class FacebookAdsReportToGcsOperator(BaseOperator):
"bucket_name",
"object_name",
"impersonation_chain",
+ "parameters",
)
def __init__(
@@ -85,7 +92,8 @@ class FacebookAdsReportToGcsOperator(BaseOperator):
bucket_name: str,
object_name: str,
fields: List[str],
- params: Dict[str, Any],
+ params: Dict[str, Any] = None,
+ parameters: Dict[str, Any] = None,
gzip: bool = False,
api_version: str = "v6.0",
gcp_conn_id: str = "google_cloud_default",
@@ -100,15 +108,26 @@ class FacebookAdsReportToGcsOperator(BaseOperator):
self.facebook_conn_id = facebook_conn_id
self.api_version = api_version
self.fields = fields
- self.params = params
+ self.parameters = parameters
self.gzip = gzip
self.impersonation_chain = impersonation_chain
+ if params is None and parameters is None:
+ raise AirflowException("Argument ['parameters'] is required")
+ if params and parameters is None:
+ # TODO: Remove in provider version 6.0
+ warnings.warn(
+ "Please use 'parameters' instead of 'params'",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+ self.parameters = params
+
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.params, fields=self.fields)
+ rows = service.bulk_facebook_report(params=self.parameters, fields=self.fields)
converted_rows = [dict(row) for row in rows]
self.log.info("Facebook Returned %s data points", len(converted_rows))
diff --git a/airflow/providers/google/marketing_platform/example_dags/example_display_video.py b/airflow/providers/google/marketing_platform/example_dags/example_display_video.py
index 8f24f08..0a3f6d5 100644
--- a/airflow/providers/google/marketing_platform/example_dags/example_display_video.py
+++ b/airflow/providers/google/marketing_platform/example_dags/example_display_video.py
@@ -71,7 +71,7 @@ REPORT = {
"schedule": {"frequency": "ONE_TIME"},
}
-PARAMS = {"dataRange": "LAST_14_DAYS", "timezoneCode": "America/New_York"}
+PARAMETERS = {"dataRange": "LAST_14_DAYS", "timezoneCode": "America/New_York"}
CREATE_SDF_DOWNLOAD_TASK_BODY_REQUEST: Dict = {
"version": SDF_VERSION,
@@ -94,7 +94,7 @@ with models.DAG(
# [START howto_google_display_video_runquery_report_operator]
run_report = GoogleDisplayVideo360RunReportOperator(
- report_id=report_id, params=PARAMS, task_id="run_report"
+ report_id=report_id, parameters=PARAMETERS, task_id="run_report"
)
# [END howto_google_display_video_runquery_report_operator]
diff --git a/airflow/providers/google/marketing_platform/operators/display_video.py b/airflow/providers/google/marketing_platform/operators/display_video.py
index 9f1f9f9..307df1b 100644
--- a/airflow/providers/google/marketing_platform/operators/display_video.py
+++ b/airflow/providers/google/marketing_platform/operators/display_video.py
@@ -21,6 +21,7 @@ import json
import shutil
import tempfile
import urllib.request
+import warnings
from typing import Any, Dict, List, Optional, Sequence, Union
from urllib.parse import urlparse
@@ -339,8 +340,12 @@ class GoogleDisplayVideo360RunReportOperator(BaseOperator):
:param report_id: Report ID to run.
:type report_id: str
:param params: Parameters for running a report as described here:
- https://developers.google.com/bid-manager/v1/queries/runquery
+ https://developers.google.com/bid-manager/v1/queries/runquery. Please note that this
+ keyword is deprecated, please use `parameters` keyword to pass the parameters.
:type params: Dict[str, Any]
+ :param parameters: Parameters for running a report as described here:
+ https://developers.google.com/bid-manager/v1/queries/runquery
+ :type parameters: Dict[str, Any]
:param api_version: The version of the api that will be requested for example 'v3'.
:type api_version: str
:param gcp_conn_id: The connection ID to use when fetching connection info.
@@ -362,7 +367,7 @@ class GoogleDisplayVideo360RunReportOperator(BaseOperator):
template_fields = (
"report_id",
- "params",
+ "parameters",
"impersonation_chain",
)
@@ -370,7 +375,8 @@ class GoogleDisplayVideo360RunReportOperator(BaseOperator):
self,
*,
report_id: str,
- params: Dict[str, Any],
+ params: Dict[str, Any] = None,
+ parameters: Dict[str, Any] = None,
api_version: str = "v1",
gcp_conn_id: str = "google_cloud_default",
delegate_to: Optional[str] = None,
@@ -379,12 +385,23 @@ class GoogleDisplayVideo360RunReportOperator(BaseOperator):
) -> None:
super().__init__(**kwargs)
self.report_id = report_id
- self.params = params
self.api_version = api_version
self.gcp_conn_id = gcp_conn_id
self.delegate_to = delegate_to
+ self.parameters = parameters
self.impersonation_chain = impersonation_chain
+ if params is None and parameters is None:
+ raise AirflowException("Argument ['parameters'] is required")
+ if params and parameters is None:
+ # TODO: Remove in provider version 6.0
+ warnings.warn(
+ "Please use 'parameters' instead of 'params'",
+ DeprecationWarning,
+ stacklevel=2,
+ )
+ self.parameters = params
+
def execute(self, context: dict) -> None:
hook = GoogleDisplayVideo360Hook(
gcp_conn_id=self.gcp_conn_id,
@@ -395,9 +412,9 @@ class GoogleDisplayVideo360RunReportOperator(BaseOperator):
self.log.info(
"Running report %s with the following params:\n %s",
self.report_id,
- self.params,
+ self.parameters,
)
- hook.run_query(query_id=self.report_id, params=self.params)
+ hook.run_query(query_id=self.report_id, params=self.parameters)
class GoogleDisplayVideo360DownloadLineItemsOperator(BaseOperator):
diff --git a/tests/providers/google/cloud/transfers/test_facebook_ads_to_gcs.py b/tests/providers/google/cloud/transfers/test_facebook_ads_to_gcs.py
index d43bfc5..4871bbf 100644
--- a/tests/providers/google/cloud/transfers/test_facebook_ads_to_gcs.py
+++ b/tests/providers/google/cloud/transfers/test_facebook_ads_to_gcs.py
@@ -31,7 +31,7 @@ FIELDS = [
"clicks",
"impressions",
]
-PARAMS = {"level": "ad", "date_preset": "yesterday"}
+PARAMETERS = {"level": "ad", "date_preset": "yesterday"}
FACEBOOK_RETURN_VALUE = [
{
"campaign_name": "abcd",
@@ -51,7 +51,7 @@ class TestFacebookAdsReportToGcsOperator:
op = FacebookAdsReportToGcsOperator(
facebook_conn_id=FACEBOOK_ADS_CONN_ID,
fields=FIELDS,
- params=PARAMS,
+ parameters=PARAMETERS,
object_name=GCS_OBJ_PATH,
bucket_name=GCS_BUCKET,
task_id="run_operator",
@@ -59,7 +59,9 @@ class TestFacebookAdsReportToGcsOperator:
)
op.execute({})
mock_ads_hook.assert_called_once_with(facebook_conn_id=FACEBOOK_ADS_CONN_ID, api_version=API_VERSION)
- mock_ads_hook.return_value.bulk_facebook_report.assert_called_once_with(params=PARAMS, fields=FIELDS)
+ mock_ads_hook.return_value.bulk_facebook_report.assert_called_once_with(
+ params=PARAMETERS, fields=FIELDS
+ )
mock_gcs_hook.assert_called_once_with(
gcp_conn_id=GCS_CONN_ID,
impersonation_chain=IMPERSONATION_CHAIN,
diff --git a/tests/providers/google/marketing_platform/operators/test_display_video.py b/tests/providers/google/marketing_platform/operators/test_display_video.py
index 98ee1e7..eb00b9d 100644
--- a/tests/providers/google/marketing_platform/operators/test_display_video.py
+++ b/tests/providers/google/marketing_platform/operators/test_display_video.py
@@ -169,10 +169,10 @@ class TestGoogleDisplayVideo360RunReportOperator(TestCase):
@mock.patch("airflow.providers.google.marketing_platform.operators.display_video.BaseOperator")
def test_execute(self, mock_base_op, hook_mock):
report_id = "QUERY_ID"
- params = {"param": "test"}
+ parameters = {"param": "test"}
op = GoogleDisplayVideo360RunReportOperator(
report_id=report_id,
- params=params,
+ parameters=parameters,
api_version=API_VERSION,
task_id="test_task",
)
@@ -183,7 +183,7 @@ class TestGoogleDisplayVideo360RunReportOperator(TestCase):
api_version=API_VERSION,
impersonation_chain=None,
)
- hook_mock.return_value.run_query.assert_called_once_with(query_id=report_id, params=params)
+ hook_mock.return_value.run_query.assert_called_once_with(query_id=report_id, params=parameters)
class TestGoogleDisplayVideo360DownloadLineItemsOperator(TestCase):