Merge pull request #8 from google/ga4_mp_user_id

caiotomazelli · web-flow · commit 7598763f3738 · 2021-03-15T15:44:59.000Z
Adds user_id support for GA4 MP. Improve checks for App and Web required parameters.
diff --git a/.gitignore b/.gitignore
@@ -10,3 +10,4 @@ generate_csv.sh
 .terraform
 .idea
 .venv
+.vscode
diff --git a/megalist_dataflow/uploaders/google_analytics/google_analytics_4_measurement_protocol.py b/megalist_dataflow/uploaders/google_analytics/google_analytics_4_measurement_protocol.py
@@ -1,13 +1,13 @@
 # Copyright 2021 Google LLC
 #
-# Licensed under the Apache License, Version 2.0 (the "License");
+# Licensed under the Apache License, Version 2.0 (the 'License');
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     https://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
+# distributed under the License is distributed on an 'AS IS' BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
@@ -28,7 +28,7 @@
 class GoogleAnalytics4MeasurementProtocolUploaderDoFn(beam.DoFn):
   def __init__(self):
     super().__init__()
-    self.API_URL = "https://www.google-analytics.com/mp/collect"
+    self.API_URL = 'https://www.google-analytics.com/mp/collect'
 
   def start_bundle(self):
     pass
@@ -41,7 +41,7 @@ def _str2bool(s: str) -> bool:
   def _exactly_one_of(a: Any, b: Any) -> bool:
     return (a and not b) or (not a and b)
 
-  @utils.safe_process(logger=logging.getLogger("megalista.GoogleAnalytics4MeasurementProtocolUploader"))
+  @utils.safe_process(logger=logging.getLogger('megalista.GoogleAnalytics4MeasurementProtocolUploader'))
   def process(self, batch: Batch, **kwargs):
     execution = batch.execution
 
@@ -60,52 +60,62 @@ def process(self, batch: Batch, **kwargs):
      
     if not self._exactly_one_of(firebase_app_id, measurement_id):
           raise ValueError(
-            f"GA4 MP should be called either with a firebase_app_id (for apps) or a measurement_id (for web)")      
+            'GA4 MP should be called either with a firebase_app_id (for apps) or a measurement_id (for web)')      
 
     if not self._exactly_one_of(is_event, is_user_property):
           raise ValueError(
-            f"GA4 MP should be called either for sending events or a user properties")        
+            'GA4 MP should be called either for sending events or a user properties')        
     
     payload: Dict[str, Any] = {
-      "nonPersonalizedAds": non_personalized_ads
+      'nonPersonalizedAds': non_personalized_ads
     }
 
     accepted_elements = []
 
     for row in batch.elements:
       app_instance_id = row.get('app_instance_id')
       client_id = row.get('client_id')
+      user_id = row.get('user_id')
 
       if not self._exactly_one_of(app_instance_id, client_id):
-        logging.getLogger("megalista.GoogleAnalytics4MeasurementProtocolUploader").error(
-          f"GA4 MP should be called either with an app_instance_id (for apps) or a client_id (for web)")
+        raise ValueError(
+          'GA4 MP should be called either with an app_instance_id (for apps) or a client_id (for web)')
     
       if is_event:
-        params = {k: v for k, v in row.items() if k not in ("name", "app_instance_id", "client_id", "uuid")}
-        payload["events"] = [{"name": row["name"], "params": params}]
+        params = {k: v for k, v in row.items() if k not in ('name', 'app_instance_id', 'client_id', 'uuid', 'user_id')}
+        payload['events'] = [{'name': row['name'], 'params': params}]
 
       if is_user_property: 
-        payload["userProperties"] = {k: {"value": v} for k, v in row.items() if k not in ("app_instance_id", "client_id", "uuid")}
-        payload["events"] = {"name": "user_property_addition_event", "params": {}}
+        payload['userProperties'] = {k: {'value': v} for k, v in row.items() if k not in ('app_instance_id', 'client_id', 'uuid', 'user_id')}
+        payload['events'] = {'name': 'user_property_addition_event', 'params': {}}
 
-      url_container = [f"{self.API_URL}?api_secret={api_secret}"]
+      url_container = [f'{self.API_URL}?api_secret={api_secret}']
 
       if firebase_app_id:
-        url_container.append(f"&firebase_app_id={firebase_app_id}")
-        payload["app_instance_id"] = app_instance_id
-
+        url_container.append(f'&firebase_app_id={firebase_app_id}')
+        if not app_instance_id:
+          raise ValueError(
+            'GA4 MP needs an app_instance_id parameter when used for an App Stream.')
+        payload['app_instance_id'] = app_instance_id
+        
       if measurement_id:
-        url_container.append(f"&measurement_id={measurement_id}")
-        payload["client_id"] = client_id
+        url_container.append(f'&measurement_id={measurement_id}')
+        if not client_id:
+          raise ValueError(
+            'GA4 MP needs a client_id parameter when used for a Web Stream.')
+        payload['client_id'] = client_id
+
+      if user_id:
+        payload['user_id'] = user_id
 
       url = ''.join(url_container)
       response = requests.post(url,data=json.dumps(payload))
       if response.status_code != 204:
-        logging.getLogger("megalista.GoogleAnalytics4MeasurementProtocolUploader").error(
-          f"Error calling GA4 MP {response.status_code}: {response.raw}")
+        logging.getLogger('megalista.GoogleAnalytics4MeasurementProtocolUploader').error(
+          f'Error calling GA4 MP {response.status_code}: {response.raw}')
       else:
         accepted_elements.append(row)
 
-    logging.getLogger("megalista.GoogleAnalytics4MeasurementProtocolUploader").info(
-      f"Successfully uploaded {len(accepted_elements)}/{len(batch.elements)} events.")
+    logging.getLogger('megalista.GoogleAnalytics4MeasurementProtocolUploader').info(
+      f'Successfully uploaded {len(accepted_elements)}/{len(batch.elements)} events.')
     yield Batch(execution, accepted_elements)
diff --git a/megalist_dataflow/uploaders/google_analytics/google_analytics_4_measurement_protocol_test.py b/megalist_dataflow/uploaders/google_analytics/google_analytics_4_measurement_protocol_test.py
@@ -1,13 +1,13 @@
 # Copyright 2021 Google LLC
 #
-# Licensed under the Apache License, Version 2.0 (the "License");
+# Licensed under the Apache License, Version 2.0 (the 'License');
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     https://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
+# distributed under the License is distributed on an 'AS IS' BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
@@ -46,7 +46,7 @@ def test_exception_event_and_user_property(uploader, caplog):
             ])
         source = Source('orig1', SourceType.BIG_QUERY, [])
         execution = Execution(_account_config, source, destination)
-        with pytest.raises(ValueError, match="GA4 MP should be called either for sending events"):
+        with pytest.raises(ValueError, match='GA4 MP should be called either for sending events'):
             next(uploader.process(Batch(execution, [])))
 
 
@@ -64,7 +64,7 @@ def test_exception_no_event_nor_user_property(uploader, caplog):
             ])
         source = Source('orig1', SourceType.BIG_QUERY, [])
         execution = Execution(_account_config, source, destination)
-        with pytest.raises(ValueError, match="GA4 MP should be called either for sending events"):
+        with pytest.raises(ValueError, match='GA4 MP should be called either for sending events'):
             next(uploader.process(Batch(execution, [])))
 
 
@@ -82,7 +82,7 @@ def test_exception_app_and_web(uploader, caplog):
             ])
         source = Source('orig1', SourceType.BIG_QUERY, [])
         execution = Execution(_account_config, source, destination)
-        with pytest.raises(ValueError, match="GA4 MP should be called either with a firebase_app_id"):
+        with pytest.raises(ValueError, match='GA4 MP should be called either with a firebase_app_id'):
             next(uploader.process(Batch(execution, [{
                 'name': 'event_name',
             }])))
@@ -102,12 +102,55 @@ def test_exception_no_id(uploader, caplog):
             ])
         source = Source('orig1', SourceType.BIG_QUERY, [])
         execution = Execution(_account_config, source, destination)
-        with pytest.raises(ValueError, match="GA4 MP should be called either with a firebase_app_id"):
+        with pytest.raises(ValueError, match='GA4 MP should be called either with a firebase_app_id'):
             next(uploader.process(Batch(execution, [{
                 'name': 'event_name',
                 'value': '123'
             }])))
 
+def test_exception_app_event_without_app_instance_id(uploader, caplog):
+    with requests_mock.Mocker() as m:
+        m.post(requests_mock.ANY, status_code=204)
+        destination = Destination(
+            'dest1', DestinationType.GA_4_MEASUREMENT_PROTOCOL, [
+                'api_secret',
+                'True',
+                'False',
+                '',
+                'some_id',
+                ''
+            ])
+        source = Source('orig1', SourceType.BIG_QUERY, [])
+        execution = Execution(_account_config, source, destination)
+        with pytest.raises(ValueError, match='GA4 MP needs an app_instance_id parameter when used for an App Stream.'):
+            next(uploader.process(Batch(execution, [{
+                'client_id': '123',
+                'name': 'event_name',
+                'value': '42',
+                'important_event': 'False'
+            }])))
+
+def test_exception_web_event_without_client_id(uploader, caplog):
+    with requests_mock.Mocker() as m:
+        m.post(requests_mock.ANY, status_code=204)
+        destination = Destination(
+            'dest1', DestinationType.GA_4_MEASUREMENT_PROTOCOL, [
+                'api_secret',
+                'True',
+                'False',
+                '',
+                '',
+                'some_id'
+            ])
+        source = Source('orig1', SourceType.BIG_QUERY, [])
+        execution = Execution(_account_config, source, destination)
+        with pytest.raises(ValueError, match='GA4 MP needs a client_id parameter when used for a Web Stream.'):
+            next(uploader.process(Batch(execution, [{
+                'app_instance_id': '123',
+                'name': 'event_name',
+                'value': '42',
+                'important_event': 'False'
+            }])))
 
 def test_succesful_app_event_call(uploader, caplog):
     with requests_mock.Mocker() as m:
@@ -124,14 +167,39 @@ def test_succesful_app_event_call(uploader, caplog):
         source = Source('orig1', SourceType.BIG_QUERY, [])
         execution = Execution(_account_config, source, destination)
         next(uploader.process(Batch(execution, [{
-            'firebase_app_id': '123',
+            'app_instance_id': '123',
             'name': 'event_name',
             'value': '42',
             'important_event': 'False'
         }])))
 
         assert m.call_count == 1
-        assert m.last_request.json()["events"][0]["params"]["value"] == '42'
+        assert m.last_request.json()['events'][0]['params']['value'] == '42'
+
+
+def test_succesful_app_event_call_with_user_id(uploader, caplog):
+    with requests_mock.Mocker() as m:
+        m.post(requests_mock.ANY, status_code=204)
+        destination = Destination(
+            'dest1', DestinationType.GA_4_MEASUREMENT_PROTOCOL, [
+                'api_secret',
+                'True',
+                'False',
+                '',
+                'some_id',
+                ''
+            ])
+        source = Source('orig1', SourceType.BIG_QUERY, [])
+        execution = Execution(_account_config, source, destination)
+        next(uploader.process(Batch(execution, [{
+            'app_instance_id': '123',
+            'name': 'event_name',
+            'value': '42',
+            'user_id': 'Id42'
+        }])))
+
+        assert m.call_count == 1
+        assert m.last_request.json()['user_id'] == 'Id42'
 
 
 def test_succesful_web_user_property_call(uploader, caplog):
@@ -149,13 +217,41 @@ def test_succesful_web_user_property_call(uploader, caplog):
         source = Source('orig1', SourceType.BIG_QUERY, [])
         execution = Execution(_account_config, source, destination)
         next(uploader.process(Batch(execution, [{
-            'user_ltv': '42'
+            'user_ltv': '42',
+            'client_id': 'some_id'
         },
         {
-            'user_will_churn': 'Maybe'
+            'user_will_churn': 'Maybe',
+            'client_id': 'some_id'
         }
         ])))
 
         assert m.call_count == 2
         assert m.last_request.json(
-        )["userProperties"]["user_will_churn"]["value"] == 'Maybe'
+        )['userProperties']['user_will_churn']['value'] == 'Maybe'
+
+def test_succesful_web_user_property_call_with_user_id(uploader, caplog):
+    with requests_mock.Mocker() as m:
+        m.post(requests_mock.ANY, status_code=204)
+        destination = Destination(
+            'dest1', DestinationType.GA_4_MEASUREMENT_PROTOCOL, [
+                'api_secret',
+                'False',
+                'True',
+                '',
+                '',
+                'some_id'
+            ])
+        source = Source('orig1', SourceType.BIG_QUERY, [])
+        execution = Execution(_account_config, source, destination)
+        next(uploader.process(Batch(execution, [{
+            'user_ltv': '42',
+            'user_id': 'Id42',
+            'client_id': 'someId'
+        }
+        ])))
+
+        assert m.call_count == 1
+        assert m.last_request.json(
+        )['user_id'] == 'Id42'
+

-Original file line number
+Diff line change
 .terraform
 .idea
 .venv
 +.vscode