Stop attempting to extract marketing data from nested key (#5851)

Author: oliverjwroberts

datahub/investment_lead/tasks/ingest_eyb_marketing.py

@@ -1,5 +1,7 @@
 import logging
 
+from django.db.models import Q
+
 from datahub.investment_lead.models import EYBLead
 from datahub.investment_lead.serializers import CreateEYBLeadMarketingSerializer
 from datahub.investment_lead.tasks.ingest_eyb_common import (
@@ -54,7 +56,27 @@ def _get_hashed_uuid(self, obj):
         return None if obj is None else obj.get('hashed_uuid', None)
 
     def _record_has_no_changes(self, record):
-        hashed_uuid = self._get_hashed_uuid(record.get('object'))
+        hashed_uuid = self._get_hashed_uuid(record)
         if hashed_uuid and EYBLead.objects.filter(marketing_hashed_uuid=hashed_uuid).exists():
             return True
         return False
+
+    def json_to_model(self, jsn):
+        serializer = self.serializer_class(data=jsn)
+        hashed_uuid = self._get_hashed_uuid(jsn)
+        if serializer.is_valid():
+            queryset = EYBLead.objects.filter(
+                Q(user_hashed_uuid=hashed_uuid)
+                | Q(triage_hashed_uuid=hashed_uuid)
+                | Q(marketing_hashed_uuid=hashed_uuid),
+            )
+            instance, created = queryset.update_or_create(defaults=serializer.validated_data)
+            if created:
+                self.created_hashed_uuids.append(hashed_uuid)
+            else:
+                self.updated_hashed_uuids.append(hashed_uuid)
+        else:
+            self.errors.append({
+                'index': hashed_uuid,
+                'errors': serializer.errors,
+            })

datahub/investment_lead/test/test_tasks/test_ingest_eyb_marketing.py

@@ -141,7 +141,7 @@ def test_marketing_file_is_ingested(self, caplog, test_marketing_file_path):
         """
         initial_eyb_lead_count = EYBLead.objects.count()
         initial_ingested_count = IngestedFile.objects.count()
-        file_contents = file_contents_faker(default_faker='marketing')
+        file_contents = file_contents_faker(default_faker='marketing', nested=False)
         setup_s3_bucket(BUCKET, [test_marketing_file_path], [file_contents])
         with caplog.at_level(logging.INFO):
             ingest_eyb_marketing_data(BUCKET, test_marketing_file_path)
@@ -164,7 +164,7 @@ def test_marketing_data_ingestion_does_not_update_existing(self, test_marketing_
                 'content': updated_value,
             }),
         ]
-        file_contents = file_contents_faker(records)
+        file_contents = file_contents_faker(records, nested=False)
         setup_s3_bucket(BUCKET, [test_marketing_file_path], [file_contents])
         ingest_eyb_marketing_data(BUCKET, test_marketing_file_path)
         assert EYBLead.objects.count() == 1
@@ -182,7 +182,7 @@ def test_marketing_data_ingestion_does_not_fail_with_empty_records(
         EYBLeadFactory(marketing_hashed_uuid=hashed_uuid, utm_content=initial_value)
         assert EYBLead.objects.count() == 1
         records = []
-        file_contents = file_contents_faker(records)
+        file_contents = file_contents_faker(records, nested=False)
         setup_s3_bucket(BUCKET, [test_marketing_file_path], [file_contents])
         ingest_eyb_marketing_data(BUCKET, test_marketing_file_path)
         assert EYBLead.objects.count() == 1
@@ -223,7 +223,7 @@ def test_incoming_marketing_records_trigger_correct_logging(self, caplog):
         ]
 
         file_path = f'{MARKETING_PREFIX}1.jsonl.gz'
-        file_contents = file_contents_faker(records)
+        file_contents = file_contents_faker(records, nested=False)
         setup_s3_bucket(BUCKET, [file_path], [file_contents])
         with caplog.at_level(logging.INFO):
             ingest_eyb_marketing_data(BUCKET, file_path)

datahub/investment_lead/test/test_tasks/utils.py

@@ -18,7 +18,9 @@ def setup_s3_client():
     return boto3.client('s3', REGION)
 
 
-def file_contents_faker(records: list[dict] = None, default_faker: str = 'triage') -> bytes:
+def file_contents_faker(
+    records: list[dict] = None, default_faker: str = 'triage', nested: bool = True,
+) -> bytes:
     """Create fake file contents with the specified list of records.
 
     Returns compressed lines of JSON encoded with UTF-8.
@@ -35,6 +37,7 @@ def file_contents_faker(records: list[dict] = None, default_faker: str = 'triage
             records = [eyb_lead_marketing_record_faker()]
     json_lines = [
         json.dumps({'object': record}, default=str)
+        if nested else json.dumps(record, default=str)
         for record in records
     ]
     compressed_content = gzip.compress('\n'.join(json_lines).encode('utf-8'))