[AIRFLOW-6724] Add Google Analytics 360 Accounts Retrieve Operator (#7630)

Co-authored-by: michalslowikowski00 <michal.slowikowski@polidea.com>

Author: michalslowikowski00

airflow/providers/google/marketing_platform/example_dags/example_analytics.py

@@ -0,0 +1,36 @@
+# 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.
+"""
+Example Airflow DAG that shows how to use Google Analytics 360.
+"""
+
+from airflow import models
+from airflow.providers.google.marketing_platform.operators.analytics import (
+    GoogleAnalyticsListAccountsOperator,
+)
+from airflow.utils import dates
+
+default_args = {"start_date": dates.days_ago(1)}
+
+with models.DAG(
+    "example_google_analytics",
+    default_args=default_args,
+    schedule_interval=None,  # Override to match your needs
+) as dag:
+    # [START howto_marketing_platform_list_accounts_operator]
+    list_account = GoogleAnalyticsListAccountsOperator(task_id="list_account")
+    # [END howto_marketing_platform_list_accounts_operator]

airflow/providers/google/marketing_platform/hooks/analytics.py

@@ -0,0 +1,75 @@
+#
+# 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.
+from typing import Any, Dict, List
+
+from googleapiclient.discovery import Resource, build
+
+from airflow.providers.google.cloud.hooks.base import CloudBaseHook
+
+
+class GoogleAnalyticsHook(CloudBaseHook):
+    """
+    Hook for Google Analytics 360.
+    """
+
+    def __init__(
+        self,
+        api_version: str = "v3",
+        gcp_connection_id: str = "google cloud default",
+        *args,
+        **kwargs
+    ):
+        super().__init__(*args, **kwargs)
+        self.api_version = api_version
+        self.gcp_connection_is = gcp_connection_id
+        self._conn = None
+
+    def get_conn(self) -> Resource:
+        """
+        Retrieves connection to Google Analytics 360.
+        """
+        if not self._conn:
+            http_authorized = self._authorize()
+            self._conn = build(
+                "analytics",
+                self.api_version,
+                http=http_authorized,
+                cache_discovery=False,
+            )
+        return self._conn
+
+    def list_accounts(self) -> List[Dict[str, Any]]:
+        """
+        Lists accounts list from Google Analytics 360.
+        """
+
+        self.log.info("Retrieving accounts list...")
+        result = []  # type: List[Dict]
+        conn = self.get_conn()
+        accounts = conn.management().accounts()  # pylint: disable=no-member
+        while True:
+            # start index has value 1
+            request = accounts.list(start_index=len(result) + 1)
+            response = request.execute(num_retries=self.num_retries)
+            result.extend(response.get('items', []))
+            # result is the number of fetched accounts from Analytics
+            # when all accounts will be add to the result
+            # the loop will be break
+            if response["totalResults"] <= len(result):
+                break
+        return result

airflow/providers/google/marketing_platform/operators/analytics.py

@@ -0,0 +1,64 @@
+#
+# 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.
+"""
+This module contains Google Analytics 360 operators.
+"""
+
+from airflow.models import BaseOperator
+from airflow.providers.google.marketing_platform.hooks.analytics import GoogleAnalyticsHook
+from airflow.utils.decorators import apply_defaults
+
+
+class GoogleAnalyticsListAccountsOperator(BaseOperator):
+    """
+    Lists all accounts to which the user has access.
+
+    .. seealso::
+        Check official API docs:
+        https://developers.google.com/analytics/devguides/config/mgmt/v3/mgmtReference/management/accounts/list
+        and for python client
+        http://googleapis.github.io/google-api-python-client/docs/dyn/analytics_v3.management.accounts.html#list
+
+    .. seealso::
+        For more information on how to use this operator, take a look at the guide:
+        :ref:`howto/operator:GoogleAnalyticsListAccountsOperator`
+
+    :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.
+    :type gcp_conn_id: str
+    """
+
+    template_fields = ("api_version", "gcp_connection_id",)
+
+    @apply_defaults
+    def __init__(self,
+                 api_version: str = "v3",
+                 gcp_connection_id: str = "google_cloud_default",
+                 *args,
+                 **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self.api_version = api_version
+        self.gcp_connection_id = gcp_connection_id
+
+    def execute(self, context):
+        hook = GoogleAnalyticsHook(api_version=self.api_version,
+                                   gcp_connection_id=self.gcp_connection_id)
+        result = hook.list_accounts()
+        return result

docs/howto/operator/gcp/analytics.rst

@@ -0,0 +1,50 @@
+ .. 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.
+
+Google Analytics 360 Operators
+==============================
+
+Google Analytics 360 operators allow you to lists all accounts to which the user has access.
+For more information about the Google Analytics 360 API check
+`official documentation <https://developers.google.com/analytics/devguides/config/mgmt/v3>`__.
+
+
+.. contents::
+  :depth: 1
+  :local:
+
+Prerequisite Tasks
+^^^^^^^^^^^^^^^^^^
+
+.. include:: _partials/prerequisite_tasks.rst
+
+.. _howto/operator:GoogleAnalyticsListAccountsOperator:
+
+List the Accounts
+^^^^^^^^^^^^^^^^^
+
+To list accounts from Analytics you can use the
+:class:`~airflow.providers.google.marketing_platform.operators.analytics.GoogleAnalyticsListAccountsOperator`.
+
+.. exampleinclude:: ../../../../airflow/providers/google/marketing_platform/example_dags/example_analytics.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_marketing_platform_list_accounts_operator]
+    :end-before: [END howto_marketing_platform_list_accounts_operator]
+
+You can use :ref:`Jinja templating <jinja-templating>` with
+:template-fields:`airflow.providers.google.marketing_platform.operators.analytics.GoogleAnalyticsListAccountsOperator`

docs/operators-and-hooks-ref.rst

@@ -554,6 +554,12 @@ These integrations allow you to perform various operations within the Google Clo
      - Operators
      - Sensors
 
+   * - `Analytics360 <https://analytics.google.com/>`__
+     - :doc:`How to use <howto/operator/gcp/analytics>`
+     - :mod:`airflow.providers.google.marketing_platform.hooks.analytics`
+     - :mod:`airflow.providers.google.marketing_platform.operators.analytics`
+     -
+
    * - `AutoML <https://cloud.google.com/automl/>`__
      - :doc:`How to use <howto/operator/gcp/automl>`
      - :mod:`airflow.providers.google.cloud.hooks.automl`

tests/providers/google/marketing_platform/hooks/test_analytics.py

@@ -0,0 +1,76 @@
+#
+# 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.
+import unittest
+from unittest import mock
+
+from airflow.providers.google.marketing_platform.hooks.analytics import GoogleAnalyticsHook
+from tests.providers.google.cloud.utils.base_gcp_mock import mock_base_gcp_hook_default_project_id
+
+API_VERSION = "v3"
+GCP_CONN_ID = "google_cloud_default"
+
+
+class TestGoogleAnalyticsHook(unittest.TestCase):
+    def setUp(self):
+        with mock.patch(
+            "airflow.providers.google.cloud.hooks.base.CloudBaseHook.__init__",
+            new=mock_base_gcp_hook_default_project_id,
+        ):
+            self.hook = GoogleAnalyticsHook(API_VERSION, GCP_CONN_ID)
+
+    @mock.patch(
+        "airflow.providers.google.marketing_platform.hooks."
+        "analytics.GoogleAnalyticsHook._authorize"
+    )
+    @mock.patch("airflow.providers.google.marketing_platform.hooks.analytics.build")
+    def test_gen_conn(self, mock_build, mock_authorize):
+        result = self.hook.get_conn()
+        mock_build.assert_called_once_with(
+            "analytics",
+            API_VERSION,
+            http=mock_authorize.return_value,
+            cache_discovery=False,
+        )
+        self.assertEqual(mock_build.return_value, result)
+
+    @mock.patch(
+        "airflow.providers.google.marketing_platform.hooks."
+        "analytics.GoogleAnalyticsHook.get_conn"
+    )
+    def test_list_accounts(self, get_conn_mock):
+        mock_accounts = get_conn_mock.return_value.management.return_value.accounts
+        mock_list = mock_accounts.return_value.list
+        mock_execute = mock_list.return_value.execute
+        mock_execute.return_value = {"items": ["a", "b"], "totalResults": 2}
+        list_accounts = self.hook.list_accounts()
+        self.assertEqual(list_accounts, ["a", "b"])
+
+    @mock.patch(
+        "airflow.providers.google.marketing_platform.hooks."
+        "analytics.GoogleAnalyticsHook.get_conn"
+    )
+    def test_list_accounts_for_multiple_pages(self, get_conn_mock):
+        mock_accounts = get_conn_mock.return_value.management.return_value.accounts
+        mock_list = mock_accounts.return_value.list
+        mock_execute = mock_list.return_value.execute
+        mock_execute.side_effect = [
+            {"items": ["a"], "totalResults": 2},
+            {"items": ["b"], "totalResults": 2},
+        ]
+        list_accounts = self.hook.list_accounts()
+        self.assertEqual(list_accounts, ["a", "b"])

tests/providers/google/marketing_platform/operators/test_analytics.py

@@ -0,0 +1,43 @@
+# 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.
+
+import unittest
+from unittest import mock
+
+from airflow.providers.google.marketing_platform.operators.analytics import (
+    GoogleAnalyticsListAccountsOperator,
+)
+
+API_VERSION = "api_version"
+GCP_CONN_ID = "google_cloud_default"
+
+
+class TestGoogleAnalyticsListAccountsOperator(unittest.TestCase):
+    @mock.patch(
+        "airflow.providers.google.marketing_platform.operators."
+        "analytics.GoogleAnalyticsHook"
+    )
+    def test_execute(self, hook_mock):
+
+        op = GoogleAnalyticsListAccountsOperator(
+            api_version=API_VERSION,
+            gcp_connection_id=GCP_CONN_ID,
+            task_id="test_task",
+        )
+        op.execute(context=None)
+        hook_mock.assert_called_once()
+        hook_mock.return_value.list_accounts.assert_called_once()