michalslowikowski00 commented on a change in pull request #7743: [AIRFLOW-7075]
Operators for storing information from GCS into GA
URL: https://github.com/apache/airflow/pull/7743#discussion_r396956097
##########
File path: airflow/providers/google/marketing_platform/hooks/analytics.py
##########
@@ -121,21 +127,119 @@ def list_ad_words_links(
"""
self.log.info("Retrieving ad words list...")
- result = [] # type: List[Dict]
conn = self.get_conn()
- ads_links = conn.management().webPropertyAdWordsLinks() # pylint:
disable=no-member
- while True:
- # start index has value 1
- request = ads_links.list(
- accountId=account_id,
- webPropertyId=web_property_id,
- start_index=len(result) + 1,
- )
- response = request.execute(num_retries=self.num_retries)
- result.extend(response.get("items", []))
- # result is the number of fetched links from Analytics
- # when all links will be added to the result
- # the loop will break
- if response["totalResults"] <= len(result):
- break
+ ads_links = (
+ conn.management().webPropertyAdWordsLinks() # pylint:
disable=no-member
+ )
+ list_args = {"accountId": account_id, "webPropertyId": web_property_id}
+ result = self._paginate(ads_links, list_args)
+ return result
+
+ def upload_data(
+ self,
+ file_location: str,
+ account_id: str,
+ web_property_id: str,
+ custom_data_source_id: str,
+ resumable_upload: bool = False,
+ ) -> None:
+
+ """
+ Uploads file to GA via the Data Import API
+
+ :param file_location: The path and name of the file to upload.
+ :type file_location: str
+ :param account_id: The GA account Id to which the data upload belongs.
+ :type account_id: str
+ :param web_property_id: UA-string associated with the upload.
+ :type web_property_id: str
+ :param custom_data_source_id: Custom Data Source Id to which this data
import belongs.
+ :type custom_data_source_id: str
+ :param resumable_upload: flag to upload the file in a resumable
fashion, using a
+ series of at least two requests
+ :type resumable_upload: bool
+ """
+
+ media = MediaFileUpload(
+ file_location,
+ mimetype="application/octet-stream",
+ resumable=resumable_upload,
+ )
+
+ self.log.info(
+ "Uploading file to GA file for accountId: %s, webPropertyId:%s and
customDataSourceId:%s ",
+ account_id,
+ web_property_id,
+ custom_data_source_id,
+ )
+
+ self.get_conn().management().uploads().uploadData( # pylint:
disable=no-member
+ accountId=account_id,
+ webPropertyId=web_property_id,
+ customDataSourceId=custom_data_source_id,
+ media_body=media,
+ ).execute()
+
+ def delete_upload_data(
+ self,
+ account_id: str,
+ web_property_id: str,
+ custom_data_source_id: str,
+ delete_request_body: Dict[str, Any],
+ ) -> None:
+ """
+ Deletes the uploaded data for a given account/property/dataset
+
+ :param account_id: The GA account Id to which the data upload belongs.
+ :type account_id: str
+ :param web_property_id: UA-string associated with the upload.
+ :type web_property_id: str
+ :param custom_data_source_id: Custom Data Source Id to which this data
import belongs.
+ :type custom_data_source_id: str
+ :param delete_request_body: Dict of customDataImportUids to delete
Review comment:
;)
```suggestion
:param delete_request_body: Dict of customDataImportUids to delete.
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
