Adds SSI tests

Author: Mr-Lopes

.github/workflows/python-app.yml

@@ -16,10 +16,10 @@ jobs:
 
     steps:
     - uses: actions/checkout@v2
-    - name: Set up Python 3.8
+    - name: Set up Python 3.11
       uses: actions/setup-python@v2
       with:
-        python-version: 3.8
+        python-version: 3.11
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip

megalista_dataflow/data_sources/data_schemas.py

@@ -125,7 +125,8 @@
             {'name': 'mailing_address_zip_name',
                 'required': False, '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': 'currency_code', 'required': True, 'data_type': 'string'}
         ],
         'groups': [
             ['email', 'phone', 'mailing_address_first_name']

megalista_dataflow/megalista_metadata

@@ -90,7 +90,7 @@
         },
         {
             "name": "collect_usage_stats",
-            "label": "Should show code lines in log messages?",
+            "label": "Share crash/usage stats that helps us suporting the solution? True or False",
             "help_text": "Share crash/usage stats that helps us suporting the solution? True or False",
             "is_optional": "true"
         }

megalista_dataflow/uploaders/google_ads/conversions/google_ads_ssi_uploader.py

@@ -25,56 +25,83 @@
 
 
 class GoogleAdsSSIUploaderDoFn(MegalistaUploader):
-
     def __init__(self, oauth_credentials, developer_token, error_handler: ErrorHandler):
         super().__init__(error_handler)
         self.oauth_credentials = oauth_credentials
         self.developer_token = developer_token
         self.active = developer_token is not None
 
     def _get_offline_user_data_job_service(self, customer_id):
-        return utils.get_ads_service('OfflineUserDataJobService', ADS_API_VERSION,
-                                         self.oauth_credentials,
-                                         self.developer_token.get(),
-                                         customer_id)
+        return utils.get_ads_service(
+            "OfflineUserDataJobService",
+            ADS_API_VERSION,
+            self.oauth_credentials,
+            self.developer_token.get(),
+            customer_id,
+        )
 
     @staticmethod
     def _assert_conversion_metadata_is_present(execution: Execution):
         metadata = execution.destination.destination_metadata
         if len(metadata) < 3:
             raise ValueError(
-                f'Missing destination information. Received {len(metadata)} entry(ies)')
+                f"Missing destination information. Received {len(metadata)} entry(ies)"
+            )
 
-    @utils.safe_process(
-        logger=logging.getLogger('megalista.GoogleAdsSSIUploader'))
+    @utils.safe_process(logger=logging.getLogger("megalista.GoogleAdsSSIUploader"))
     def process(self, batch: Batch, **kwargs):
         execution = batch.execution
         self._assert_conversion_metadata_is_present(execution)
 
-        currency_code = self._get_currency_code(execution.destination)
-        customer_id = self._get_customer_id(execution.account_config, execution.destination)
-        login_customer_id = self._get_login_customer_id(execution.account_config, execution.destination)
-        user_data_consent = self._get_user_data_consent(execution.account_config, execution.destination)
-        ad_personalization = self._get_ad_personalization_consent(execution.account_config, execution.destination)
+        custom_key = self._get_custom_key(execution.destination)
+        customer_id = self._get_customer_id(
+            execution.account_config, execution.destination
+        )
+        login_customer_id = self._get_login_customer_id(
+            execution.account_config, execution.destination
+        )
+        user_data_consent = self._get_user_data_consent(
+            execution.account_config, execution.destination
+        )
+        ad_personalization = self._get_ad_personalization_consent(
+            execution.account_config, execution.destination
+        )
 
         offline_user_data_job_service = self._get_offline_user_data_job_service(
-            login_customer_id)
-        conversion_action_resource_name = self._get_resource_name(customer_id,
-                                                                  login_customer_id,  
-                                                                  execution.destination.destination_metadata[0])
-        self._do_upload(execution,
-                        offline_user_data_job_service,
-                        customer_id,
-                        currency_code,
-                        conversion_action_resource_name,
-                        user_data_consent,
-                        ad_personalization,
-                        batch.elements)
+            login_customer_id
+        )
+
+        ads_service = self._get_ads_service(login_customer_id)
+
+        conversion_action_resource_name = self._get_resource_name(
+            ads_service, customer_id, execution.destination.destination_metadata[0]
+        )
+
+        self._do_upload(
+            execution,
+            offline_user_data_job_service,
+            customer_id,
+            custom_key,
+            conversion_action_resource_name,
+            user_data_consent,
+            ad_personalization,
+            batch.elements,
+        )
 
         return [execution]
 
-    def _do_upload(self, execution, offline_user_data_job_service, customer_id, currency_code, conversion_action_resource_name, user_data_consent, ad_personalization, rows):
-        logger = logging.getLogger('megalista.GoogleAdsSSIUploader')
+    def _do_upload(
+        self,
+        execution,
+        offline_user_data_job_service,
+        customer_id,
+        custom_key,
+        conversion_action_resource_name,
+        user_data_consent,
+        ad_personalization,
+        rows,
+    ):
+        logger = logging.getLogger("megalista.GoogleAdsSSIUploader")
 
         # Upload is divided into 3 parts:
         # 1. Creates Job
@@ -83,108 +110,153 @@ def _do_upload(self, execution, offline_user_data_job_service, customer_id, curr
 
         # 1. Creates Job
         job_creation_payload = {
-            'type_': 'STORE_SALES_UPLOAD_FIRST_PARTY',
-            'external_id':  int(datetime.datetime.now().timestamp()*10e3),
-            'store_sales_metadata': {
-                'loyalty_fraction': 1.0,
-                'transaction_upload_fraction': 1.0
-            }
+            "type_": "STORE_SALES_UPLOAD_FIRST_PARTY",
+            "external_id": int(datetime.datetime.now().timestamp() * 10e3),
+            "store_sales_metadata": {
+                "loyalty_fraction": 1.0,
+                "transaction_upload_fraction": 1.0,
+                **({"custom_key": custom_key} if custom_key else {}),
+            },
         }
 
-        job_resource_name = offline_user_data_job_service.create_offline_user_data_job(customer_id = customer_id, job = job_creation_payload).resource_name
-        
+        job_resource_name = offline_user_data_job_service.create_offline_user_data_job(
+            customer_id=customer_id, job=job_creation_payload
+        ).resource_name
+
         # Sets consent info if any
         consent = {}
         if user_data_consent:
-            consent['ad_user_data'] = user_data_consent
-        if  ad_personalization:
-            consent['ad_personalization'] = ad_personalization
+            consent["ad_user_data"] = user_data_consent
+        if ad_personalization:
+            consent["ad_personalization"] = ad_personalization
 
         # 2. Creates operations (data insertion)
         data_insertion_payload = {
-            'resource_name': job_resource_name,
-            'enable_partial_failure': False,
-            'operations': [{
-                'create': {
-                    'user_identifiers': [{k: v} for (k, v) in conversion.items() if k not in ('amount', 'time')],
-                    'transaction_attribute': {
-                        'conversion_action': conversion_action_resource_name,
-                        'currency_code': currency_code,
-                        'transaction_amount_micros': conversion['amount'],
-                        'transaction_date_time': utils.format_date(conversion['time'])
-                    },
-                    **({'consent':consent} if consent else {})
+            "resource_name": job_resource_name,
+            "enable_partial_failure": False,
+            "operations": [
+                {
+                    "create": {
+                        "user_identifiers": [
+                            {k: v}
+                            for (k, v) in conversion.items()
+                            if k not in ("amount", "time", "currency_code", "custom_value")
+                        ],
+                        "transaction_attribute": {
+                            "conversion_action": conversion_action_resource_name,
+                            "currency_code": conversion["currency_code"],
+                            "transaction_amount_micros": int(conversion["amount"]),
+                            "transaction_date_time": utils.format_date(
+                                conversion["time"]
+                            ),
+                            **(
+                                {"custom_value": conversion["custom_value"]}
+                                if "custom_value" in conversion and custom_key
+                                else {}
+                            ),
+                        },
+                        **({"consent": consent} if consent else {}),
+                    }
                 }
-            } for conversion in rows]
+                for conversion in rows
+            ],
         }
 
-        data_insertion_response = offline_user_data_job_service.add_offline_user_data_job_operations(request = data_insertion_payload)
+        data_insertion_response = (
+            offline_user_data_job_service.add_offline_user_data_job_operations(
+                request=data_insertion_payload
+            )
+        )
 
-        error_message = utils.print_partial_error_messages(logger, 'uploading ssi',
-                                                           data_insertion_response)
+        error_message = utils.print_partial_error_messages(
+            logger, "uploading ssi", data_insertion_response
+        )
         if error_message:
             self._add_error(execution, error_message)
 
         # 3. Runs the Job
-        offline_user_data_job_service.run_offline_user_data_job(resource_name = job_resource_name)
+        offline_user_data_job_service.run_offline_user_data_job(
+            resource_name=job_resource_name
+        )
 
-    def _get_customer_id(self, account_config: AccountConfig, destination: Destination) -> str:
+    def _get_customer_id(
+        self, account_config: AccountConfig, destination: Destination
+    ) -> str:
         """
-          If the customer_id is present on the destination, returns it, otherwise defaults to the account_config info.
+        If the customer_id is present on the destination, returns it, otherwise defaults to the account_config info.
         """
-        if len(destination.destination_metadata) >= 2 and len(destination.destination_metadata[1]) > 0:
+        if (
+            len(destination.destination_metadata) >= 2
+            and len(destination.destination_metadata[1]) > 0
+        ):
             return Utils.filter_text_only_numbers(destination.destination_metadata[1])
         return account_config.google_ads_account_id
 
-    def _get_currency_code(self, destination: Destination) -> str:
+    def _get_custom_key(self, destination: Destination) -> Optional[str]:
         """
-          If the currency_code is present on the destination, return it, otherwise default to BRL.
+        If the custom_key is present on the destination, return it, otherwise returns none.
         """
-        if len(destination.destination_metadata) >= 5 and len(destination.destination_metadata[3]) > 0:
+        if (
+            len(destination.destination_metadata) >= 5
+            and destination.destination_metadata[3] is not None
+        ):
             return destination.destination_metadata[3]
-        return 'BRL'
-    
-    def _get_user_data_consent(self, account_config: AccountConfig, destination: Destination) -> Optional[str]:
+        return None
+
+    def _get_user_data_consent(
+        self, account_config: AccountConfig, destination: Destination
+    ) -> Optional[str]:
         """
-          Specifies whether user consent was obtained for the data you are uploading.
-          https://www.google.com/about/company/user-consent-policy
+        Specifies whether user consent was obtained for the data you are uploading.
+        https://www.google.com/about/company/user-consent-policy
         """
-        if len(destination.destination_metadata) >= 5 and len(destination.destination_metadata[4]) > 0:
+        if (
+            len(destination.destination_metadata) >= 5
+            and len(destination.destination_metadata[4]) > 0
+        ):
             return destination.destination_metadata[4]
         return None
-    
-    def _get_ad_personalization_consent(self, account_config: AccountConfig, destination: Destination) -> Optional[str]:
+
+    def _get_ad_personalization_consent(
+        self, account_config: AccountConfig, destination: Destination
+    ) -> Optional[str]:
         """
-          Specifies whether user consent was obtained for the data you are uploading.
-          https://www.google.com/about/company/user-consent-policy
+        Specifies whether user consent was obtained for the data you are uploading.
+        https://www.google.com/about/company/user-consent-policy
         """
-        if len(destination.destination_metadata) >= 6 and len(destination.destination_metadata[5]) > 0:
+        if (
+            len(destination.destination_metadata) >= 6
+            and len(destination.destination_metadata[5]) > 0
+        ):
             return destination.destination_metadata[5]
         return None
 
-    def _get_login_customer_id(self, account_config: AccountConfig, destination: Destination) -> str:
+    def _get_login_customer_id(
+        self, account_config: AccountConfig, destination: Destination
+    ) -> str:
         """
-          If the customer_id in account_config is a mcc, then login with the mcc account id, otherwise use the customer id.
+        If the customer_id in account_config is a mcc, then login with the mcc account id, otherwise use the customer id.
         """
         if account_config._mcc:
             return account_config.google_ads_account_id
-        
+
         return self._get_customer_id(account_config, destination)
 
     def _get_ads_service(self, customer_id: str):
-      return utils.get_ads_service('GoogleAdsService', ADS_API_VERSION,
-                                       self.oauth_credentials,
-                                       self.developer_token.get(),
-                                       customer_id)
-
-    def _get_resource_name(self, customer_id: str, login_customer_id: str, name: str):
-        resource_name = None
-        service = self._get_ads_service(login_customer_id)
+        return utils.get_ads_service(
+            "GoogleAdsService",
+            ADS_API_VERSION,
+            self.oauth_credentials,
+            self.developer_token.get(),
+            customer_id,
+        )
+
+    def _get_resource_name(self, ads_service, customer_id: str, name: str):
         query = f"SELECT conversion_action.resource_name FROM conversion_action WHERE conversion_action.name = '{name}'"
-        response_query = service.search_stream(customer_id=customer_id, query=query)
+        response_query = ads_service.search_stream(customer_id=customer_id, query=query)
         for batch in response_query:
-          for row in batch.results:
-            resource_name = row.conversion_action.resource_name
-        return resource_name
-
-
+            for row in batch.results:
+                return row.conversion_action.resource_name
+        raise Exception(
+            f'Conversion "{name}" could not be found on account {customer_id}'
+        )