Merge pull request #6 from google/develop

Merge develop

Author: antoniolmm

megalist_dataflow/README.md

@@ -74,6 +74,7 @@ def expand(self, executions):
         return (
             executions
             | 'Load Data - Google Ads Customer Match Mobile Device Id' >> BatchesFromExecutions(DestinationType.ADS_CUSTOMER_MATCH_MOBILE_DEVICE_ID_UPLOAD)
+            | 'Hash Users - Google Ads Customer Match Contact Info' >> beam.Map(self._hasher.hash_users)
             | 'Upload - Google Ads Customer Match Mobile Device Id' >> beam.ParDo(
                 GoogleAdsCustomerMatchMobileUploaderDoFn(self._oauth_credentials, self._dataflow_options.developer_token))
         )
@@ -95,6 +96,7 @@ def expand(self, executions):
         return (
             executions
             | 'Load Data - Google Ads Customer Match User Id' >> BatchesFromExecutions(DestinationType.ADS_CUSTOMER_MATCH_USER_ID_UPLOAD)
+            | 'Hash Users - Google Ads Customer Match Contact Info' >> beam.Map(self._hasher.hash_users)
             | 'Upload - Google Ads Customer User Device Id' >> beam.ParDo(
                 GoogleAdsCustomerMatchUserIdUploaderDoFn(self._oauth_credentials, self._dataflow_options.developer_token))
         )

megalist_dataflow/main.py

@@ -79,7 +79,7 @@ def _hash_user(self, user, hasher):
 
         try:
             if 'user_id' in user:
-                hashed['userId'] = user['user_id']
+                hashed['userId'] = hasher.hash_field(user['user_id'])
                 del hashed['user_id']
         except:
             self.logger.error("Error hashing user_id for user: %s" % user)

megalist_dataflow/mappers/ads_user_list_pii_hashing_mapper.py

@@ -114,18 +114,16 @@ def process(self, batch: Batch, **kwargs) -> None:
 
     rows = self.get_filtered_rows(
         batch.elements, self.get_row_keys())
-    logging.getLogger(_DEFAULT_LOGGER).warning(
-        'Uploading %d rows to Google Ads', len(rows))
+        
     mutate_members_operation = {
         'operand': {
             'userListId': list_id,
             'membersList': rows
         },
         'operator': execution.destination.destination_metadata[1]
     }
+    
     utils.safe_call_api(self.call_api, logging, user_list_service, [mutate_members_operation])
-    logging.getLogger(_DEFAULT_LOGGER).warning(
-      'Uploaded %d rows to Google Ads', len(rows))
 
   def call_api(self, service, operations):
     service.mutateMembers(operations)