adding new columns to process external attribution for oci

megalista_dataflow/data_sources/data_schemas.py

@@ -49,7 +49,9 @@
         'columns': [
             {'name': 'gclid', 'required': True, 'data_type': 'string'},
             {'name': 'time', 'required': True, 'data_type': 'string'},
-            {'name': 'amount', 'required': True, 'data_type': 'string'}
+            {'name': 'amount', 'required': True, 'data_type': 'string'},
+            {'name': 'external_attribution_credit', 'required': False, 'data_type': 'string'},
+            {'name': 'external_attribution_model', 'required': False, 'data_type': 'string'}
         ],
         'groups': []
     },

megalista_dataflow/uploaders/google_ads/conversions/google_ads_offline_conversions_uploader.py

@@ -106,12 +106,19 @@ def process(self, batch: Batch, **kwargs):
 
   def _do_upload(self, oc_service, execution, conversion_resource_name, customer_id, rows):
     logging.getLogger(_DEFAULT_LOGGER).info(f'Uploading {len(rows)} offline conversions on {conversion_resource_name} to Google Ads.')
-    conversions = [{
-          'conversion_action': conversion_resource_name,
-          'conversion_date_time': utils.format_date(conversion['time']),
-          'conversion_value': float(str(conversion['amount'])),
-          'gclid': conversion['gclid']
-    } for conversion in rows]
+    conversions = []
+    for row in rows:
+      conversion = {}
+      conversion['conversion_action'] = conversion_resource_name
+      conversion['conversion_date_time'] = utils.format_date(row['time'])
+      conversion['conversion_value'] = float(str(row['amount']))
+      conversion['gclid'] = row['gclid']
+      if 'external_attribution_credit' in row and 'external_attribution_model' in row:
+        conversion['external_attribution_data'] = {
+          'external_attribution_credit': float(str(row['external_attribution_credit'])),
+          'external_attribution_model': row['external_attribution_model']
+        }
+      conversions.append(conversion)
 
     upload_data = {
       'customer_id': customer_id,

megalista_dataflow/uploaders/google_ads/conversions/google_ads_offline_conversions_uploader_test.py

@@ -303,7 +303,8 @@ def test_conversion_upload_and_error_notification(mocker, uploader, error_notifi
   element1 = {
     'time': time1,
     'amount': '123',
-    'gclid': '456'
+    'gclid': '456',
+
   }
   element2 = {
     'time': time2,
@@ -359,3 +360,94 @@ def test_conversion_upload_and_error_notification(mocker, uploader, error_notifi
   assert error_notifier.were_errors_sent is True
   assert error_notifier.destination_type is DestinationType.ADS_OFFLINE_CONVERSION
   assert error_notifier.errors_sent == {execution: f'Error on uploading offline conversions: {error_message}.'}
+
+
+def test_conversion_upload_and_error_notification_with_external_attribution(mocker, uploader, error_notifier):
+  """
+  Scenario where some gclids are uploaded but some give errors
+  """
+
+  # arrange
+  conversion_resource_name = 'user_list_resouce'
+  arrange_conversion_resource_name_api_call(mocker, uploader, conversion_resource_name)
+
+  mocker.patch.object(uploader, '_get_oc_service')
+  conversion_name = 'user_list'
+  destination = Destination(
+    'dest1', DestinationType.ADS_OFFLINE_CONVERSION, ['user_list'])
+  source = Source('orig1', SourceType.BIG_QUERY, ['dt1', 'buyers'])
+  execution = Execution(_account_config, source, destination)
+
+  time1 = '2020-04-09T14:13:55.0005'
+  time1_result = '2020-04-09 14:13:55-03:00'
+
+  time2 = '2020-04-09T13:13:55.0005'
+  time2_result = '2020-04-09 13:13:55-03:00'
+
+  element1 = {
+    'time': time1,
+    'amount': '123',
+    'gclid': '456',
+    'external_attribution_credit': 0.6,
+    'external_attribution_model': 'teste_attribution'
+  }
+  element2 = {
+    'time': time2,
+    'amount': '234',
+    'gclid': '567'
+  }
+  batch = Batch(execution, [element1, element2])
+
+  # gclid '456' returns as successful by the API, while gclid '567' does not.
+  # in this scenario, it's expected that both are present in the API call,
+  # but since gclid '567' is not returned as successful by the API, an error is sent through error_notifier
+
+  error_message = 'Offline Conversion uploading failures'
+
+  gclid_result_mock1 = MagicMock()
+  gclid_result_mock1.gclid = '456'
+  gclid_result_mock1.external_attribution_data.external_attribution_credit = 0.6
+  gclid_result_mock1.external_attribution_data.external_attribution_model = 'teste_attribution'
+
+  upload_return_mock = MagicMock()
+  upload_return_mock.results = [gclid_result_mock1]
+  upload_return_mock.partial_failure_error.message = error_message
+  uploader._get_oc_service.return_value.upload_click_conversions.return_value = upload_return_mock
+
+  # act
+  successful_uploaded_gclids_batch = uploader.process(batch)[0]
+  uploader.finish_bundle()
+
+  # assert
+  assert len(successful_uploaded_gclids_batch.elements) == 1
+  assert successful_uploaded_gclids_batch.elements[0] == element1
+
+  uploader._get_ads_service.return_value.search_stream.assert_called_once_with(
+    customer_id='12345567890',
+    query=f"SELECT conversion_action.resource_name FROM conversion_action WHERE conversion_action.name = '{conversion_name}'"
+  )
+
+  uploader._get_oc_service.return_value.upload_click_conversions.assert_called_once_with(request={
+    'customer_id': '12345567890',
+    'partial_failure': True,
+    'validate_only': False,
+    'conversions': [{
+      'conversion_action': conversion_resource_name,
+      'conversion_date_time': time1_result,
+      'conversion_value': 123,
+      'gclid': '456',
+      'external_attribution_data': {
+        'external_attribution_credit': 0.6,
+        'external_attribution_model': 'teste_attribution'
+      }
+    }, {
+      'conversion_action': conversion_resource_name,
+      'conversion_date_time': time2_result,
+      'conversion_value': 234,
+      'gclid': '567'
+    }]
+  })
+
+  assert error_notifier.were_errors_sent is True
+  assert error_notifier.destination_type is DestinationType.ADS_OFFLINE_CONVERSION
+  assert error_notifier.errors_sent == {execution: f'Error on uploading offline conversions: {error_message}.'}