Add GoogleCalendarToGCSOperator (#20769)

apache · Feb 16, 2022 · af2c047 · af2c047
1 parent 0ebd642
commit af2c047
Show file tree

Hide file tree

Showing 6 changed files with 459 additions and 0 deletions.
diff --git a/airflow/providers/google/cloud/example_dags/example_calendar_to_gcs.py b/airflow/providers/google/cloud/example_dags/example_calendar_to_gcs.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 os
+from datetime import datetime
+
+from airflow import models
+from airflow.providers.google.cloud.transfers.calendar_to_gcs import GoogleCalendarToGCSOperator
+
+BUCKET = os.environ.get("GCP_GCS_BUCKET", "test28397yeo")
+CALENDAR_ID = os.environ.get("CALENDAR_ID", "1234567890qwerty")
+API_VERSION = "v3"
+
+with models.DAG(
+ "example_calendar_to_gcs",
+ schedule_interval='@once', # Override to match your needs
+ start_date=datetime(2022, 1, 1),
+ catchup=False,
+ tags=["example"],
+) as dag:
+ # [START upload_calendar_to_gcs]
+ upload_calendar_to_gcs = GoogleCalendarToGCSOperator(
+ task_id="upload_calendar_to_gcs",
+ destination_bucket=BUCKET,
+ calendar_id=CALENDAR_ID,
+ api_version=API_VERSION,
+ )
+ # [END upload_calendar_to_gcs]
diff --git a/airflow/providers/google/cloud/transfers/calendar_to_gcs.py b/airflow/providers/google/cloud/transfers/calendar_to_gcs.py
@@ -0,0 +1,194 @@
+# 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 json
+from datetime import datetime
+from tempfile import NamedTemporaryFile
+from typing import Any, List, Optional, Sequence, Union
+
+from airflow.models import BaseOperator
+from airflow.providers.google.cloud.hooks.gcs import GCSHook
+from airflow.providers.google.suite.hooks.calendar import GoogleCalendarHook
+
+
+class GoogleCalendarToGCSOperator(BaseOperator):
+ """
+ Writes Google Calendar data into Google Cloud Storage.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the guide:
+ :ref:`howto/operator:GoogleCalendarToGCSOperator`
+
+ :param calendar_id: The Google Calendar ID to interact with.
+ :param i_cal_uid: Optional. Specifies event ID in the ``iCalendar`` format in the response.
+ :param max_attendees: Optional. If there are more than the specified number of attendees,
+ only the participant is returned.
+ :param max_results: Optional. Maximum number of events returned on one result page.
+ Incomplete pages can be detected by a non-empty ``nextPageToken`` field in the response.
+ By default the value is 250 events. The page size can never be larger than 2500 events
+ :param order_by: Optional. Acceptable values are ``"startTime"`` or "updated"
+ :param private_extended_property: Optional. Extended properties constraint specified as
+ ``propertyName=value``. Matches only private properties. This parameter might be repeated
+ multiple times to return events that match all given constraints.
+ :param text_search_query: Optional. Free text search.
+ :param shared_extended_property: Optional. Extended properties constraint specified as
+ ``propertyName=value``. Matches only shared properties. This parameter might be repeated
+ multiple times to return events that match all given constraints.
+ :param show_deleted: Optional. False by default
+ :param show_hidden_invitation: Optional. False by default
+ :param single_events: Optional. False by default
+ :param sync_token: Optional. Token obtained from the ``nextSyncToken`` field returned
+ :param time_max: Optional. Upper bound (exclusive) for an event's start time to filter by.
+ Default is no filter
+ :param time_min: Optional. Lower bound (exclusive) for an event's end time to filter by.
+ Default is no filter
+ :param time_zone: Optional. Time zone used in response. Default is calendars time zone.
+ :param updated_min: Optional. Lower bound for an event's last modification time
+ :param destination_bucket: The destination Google Cloud Storage bucket where the
+ report should be written to. (templated)
+ :param destination_path: The Google Cloud Storage URI array for the object created by the operator.
+ For example: ``path/to/my/files``.
+ :param gcp_conn_id: The connection ID to use when fetching connection info.
+ :param delegate_to: The account to impersonate using domain-wide delegation of authority,
+ if any. For this to work, the service account making the request must have
+ domain-wide delegation enabled.
+ :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.
+ If set as a string, the account must grant the originating account
+ the Service Account Token Creator IAM role.
+ If set as a sequence, the identities from the list must grant
+ Service Account Token Creator IAM role to the directly preceding identity, with first
+ account from the list granting this role to the originating account (templated).
+ """
+
+ template_fields = [
+ "calendar_id",
+ "destination_bucket",
+ "destination_path",
+ "impersonation_chain",
+ ]
+
+ def __init__(
+ self,
+ *,
+ destination_bucket: str,
+ api_version: str,
+ calendar_id: str = "primary",
+ i_cal_uid: Optional[str] = None,
+ max_attendees: Optional[int] = None,
+ max_results: Optional[int] = None,
+ order_by: Optional[str] = None,
+ private_extended_property: Optional[str] = None,
+ text_search_query: Optional[str] = None,
+ shared_extended_property: Optional[str] = None,
+ show_deleted: Optional[bool] = None,
+ show_hidden_invitation: Optional[bool] = None,
+ single_events: Optional[bool] = None,
+ sync_token: Optional[str] = None,
+ time_max: Optional[datetime] = None,
+ time_min: Optional[datetime] = None,
+ time_zone: Optional[str] = None,
+ updated_min: Optional[datetime] = None,
+ destination_path: Optional[str] = None,
+ gcp_conn_id: str = "google_cloud_default",
+ delegate_to: Optional[str] = None,
+ impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
+ **kwargs,
+ ) -> None:
+ super().__init__(**kwargs)
+ self.gcp_conn_id = gcp_conn_id
+ self.calendar_id = calendar_id
+ self.api_version = api_version
+ self.i_cal_uid = i_cal_uid
+ self.max_attendees = max_attendees
+ self.max_results = max_results
+ self.order_by = order_by
+ self.private_extended_property = private_extended_property
+ self.text_search_query = text_search_query
+ self.shared_extended_property = shared_extended_property
+ self.show_deleted = show_deleted
+ self.show_hidden_invitation = show_hidden_invitation
+ self.single_events = single_events
+ self.sync_token = sync_token
+ self.time_max = time_max
+ self.time_min = time_min
+ self.time_zone = time_zone
+ self.updated_min = updated_min
+ self.destination_bucket = destination_bucket
+ self.destination_path = destination_path
+ self.delegate_to = delegate_to
+ self.impersonation_chain = impersonation_chain
+
+ def _upload_data(
+ self,
+ events: List[Any],
+ ) -> str:
+ gcs_hook = GCSHook(
+ gcp_conn_id=self.gcp_conn_id,
+ delegate_to=self.delegate_to,
+ impersonation_chain=self.impersonation_chain,
+ )
+
+ # Construct destination file path
+ file_name = f"{self.calendar_id}.json".replace(" ", "_")
+ dest_file_name = (
+ f"{self.destination_path.strip('/')}/{file_name}" if self.destination_path else file_name
+ )
+
+ with NamedTemporaryFile("w+") as temp_file:
+ # Write data
+ json.dump(events, temp_file)
+ temp_file.flush()
+
+ # Upload to GCS
+ gcs_hook.upload(
+ bucket_name=self.destination_bucket,
+ object_name=dest_file_name,
+ filename=temp_file.name,
+ )
+ return dest_file_name
+
+ def execute(self, context):
+ calendar_hook = GoogleCalendarHook(
+ api_version=self.api_version,
+ gcp_conn_id=self.gcp_conn_id,
+ delegate_to=self.delegate_to,
+ impersonation_chain=self.impersonation_chain,
+ )
+
+ events = calendar_hook.get_events(
+ calendar_id=self.calendar_id,
+ i_cal_uid=self.i_cal_uid,
+ max_attendees=self.max_attendees,
+ max_results=self.max_results,
+ order_by=self.order_by,
+ private_extended_property=self.private_extended_property,
+ q=self.text_search_query,
+ shared_extended_property=self.shared_extended_property,
+ show_deleted=self.show_deleted,
+ show_hidden_invitation=self.show_hidden_invitation,
+ single_events=self.single_events,
+ sync_token=self.sync_token,
+ time_max=self.time_max,
+ time_min=self.time_min,
+ time_zone=self.time_zone,
+ updated_min=self.updated_min,
+ )
+ gcs_path_to_file = self._upload_data(events)
+
+ return gcs_path_to_file
diff --git a/airflow/providers/google/provider.yaml b/airflow/providers/google/provider.yaml
@@ -714,6 +714,10 @@ transfers:
  - source-integration-name: Apache Cassandra
  target-integration-name: Google Cloud Storage (GCS)
  python-module: airflow.providers.google.cloud.transfers.cassandra_to_gcs
+ - source-integration-name: Google Calendar
+ target-integration-name: Google Cloud Storage (GCS)
+ how-to-guide: /docs/apache-airflow-providers-google/operators/transfer/calendar_to_gcs.rst
+ python-module: airflow.providers.google.cloud.transfers.calendar_to_gcs
  - source-integration-name: Google Spreadsheet
  target-integration-name: Google Cloud Storage (GCS)
  how-to-guide: /docs/apache-airflow-providers-google/operators/transfer/sheets_to_gcs.rst

diff --git a/docs/apache-airflow-providers-google/operators/transfer/calendar_to_gcs.rst b/docs/apache-airflow-providers-google/operators/transfer/calendar_to_gcs.rst
@@ -0,0 +1,47 @@
+ .. 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 Calendar to Google Cloud Storage Transfer Operators
+==========================================================
+
+Google has a service `Google Cloud Storage <https://cloud.google.com/storage/>`__. This service is
+used to store large data from various applications.
+
+With `Google Calendar <https://www.google.com/calendar/about/>`__, you can quickly schedule
+meetings and events and get reminders about upcoming activities, so you always know what's next.
+
+Prerequisite Tasks
+^^^^^^^^^^^^^^^^^^
+
+.. include::/operators/_partials/prerequisite_tasks.rst
+
+.. _howto/operator:GoogleCalendarToGCSOperator:
+
+Upload data from Google Calendar to GCS
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+To upload data from Google Calendar to Google Cloud Storage you can use the
+:class:`~airflow.providers.google.cloud.transfers.calendar_to_gcs.GoogleCalendarToGCSOperator`.
+
+.. exampleinclude:: /../../airflow/providers/google/cloud/example_dags/example_calendar_to_gcs.py
+ :language: python
+ :dedent: 4
+ :start-after: [START upload_calendar_to_gcs]
+ :end-before: [END upload_calendar_to_gcs]
+
+You can use :ref:`Jinja templating <concepts:jinja-templating>` with
+:template-fields:`airflow.providers.google.cloud.transfers.calendar_to_gcs.GoogleCalendarToGCSOperator`.