Add analytics support (#128)

Author: ferhatelmas

stream/client.py

@@ -133,9 +133,9 @@ def get_default_header(self):
 
     def get_full_url(self, service_name, relative_url):
         if self.api_location:
-            hostname = "%s-%s.%s" % (
+            hostname = "%s%s.%s" % (
                 self.api_location,
-                service_name,
+                "" if service_name == "analytics" else f"-{service_name}",
                 self.base_domain_name,
             )
         elif service_name:
@@ -472,7 +472,7 @@ def create_redirect_url(self, target_url, user_id, events):
         params = dict(auth_type="jwt", authorization=auth_token, url=target_url)
         params["api_key"] = self.api_key
         params["events"] = json.dumps(events)
-        url = self.base_analytics_url + "redirect/"
+        url = f"{self.base_analytics_url}redirect/"
         # we get the url from the prepare request, this skips issues with
         # python's urlencode implementation
         request = Request("GET", url, params=params)
@@ -481,6 +481,76 @@ def create_redirect_url(self, target_url, user_id, events):
         Request("GET", target_url).prepare()
         return prepared_request.url
 
+    def track_engagements(self, engagements):
+        """
+        Creates a list of engagements
+
+        ;param engagements: Slice of engagements to create.
+
+        eg.
+        [
+            {
+                "content": "1",
+                "label": "click",
+                "features": [
+                    {"group": "topic", "value": "js"},
+                    {"group": "user", "value": "tommaso"},
+                ],
+                "user_data": "tommaso",
+            },
+            {
+                "content": "2",
+                "label": "click",
+                "features": [
+                    {"group": "topic", "value": "go"},
+                    {"group": "user", "value": "tommaso"},
+                ],
+                "user_data": {"id": "486892", "alias": "Julian"},
+            },
+            {
+                "content": "3",
+                "label": "click",
+                "features": [{"group": "topic", "value": "go"}],
+                "user_data": {"id": "tommaso", "alias": "tommaso"},
+            },
+        ]
+        """
+
+        auth_token = self.create_jwt_token("*", "*", feed_id="*")
+        self.post(
+            "engagement/",
+            auth_token,
+            data={"content_list": engagements},
+            service_name="analytics",
+        )
+
+    def track_impressions(self, impressions):
+        """
+        Creates a list of impressions
+
+        ;param impressions: Slice of impressions to create.
+
+        eg.
+        [
+            {
+                "content_list": ["1", "2", "3"],
+                "features": [
+                    {"group": "topic", "value": "js"},
+                    {"group": "user", "value": "tommaso"},
+                ],
+                "user_data": {"id": "tommaso", "alias": "tommaso"},
+            },
+            {
+                "content_list": ["2", "3", "5"],
+                "features": [{"group": "topic", "value": "js"}],
+                "user_data": {"id": "486892", "alias": "Julian"},
+            },
+        ]
+        """
+
+        auth_token = self.create_jwt_token("*", "*", feed_id="*")
+        self.post("impression/", auth_token, data=impressions, service_name="analytics")
+
     def og(self, target_url):
         """
         Retrieve open graph information from a URL which you can

stream/tests/test_client.py

@@ -126,6 +126,20 @@ def test_collections_url(self):
                 feed_url, "https://qa-api.stream-io-api.com/api/v1.0/meta/"
             )
 
+    def test_analytics_url(self):
+        feed_url = client.get_full_url(
+            relative_url="engagement/", service_name="analytics"
+        )
+
+        if self.local_tests:
+            self.assertEqual(
+                feed_url, "http://localhost:8000/analytics/v1.0/engagement/"
+            )
+        else:
+            self.assertEqual(
+                feed_url, "https://qa.stream-io-api.com/analytics/v1.0/engagement/"
+            )
+
     def test_personalization_url(self):
         feed_url = client.get_full_url(
             relative_url="recommended", service_name="personalization"
@@ -1616,6 +1630,53 @@ def test_feed_enrichment_reaction_counts(self):
         enriched_response = f.get(reactions={"counts": True})
         self.assertEqual(enriched_response["results"][0]["reaction_counts"]["like"], 1)
 
+    def test_track_engagements(self):
+        engagements = [
+            {
+                "content": "1",
+                "label": "click",
+                "features": [
+                    {"group": "topic", "value": "js"},
+                    {"group": "user", "value": "tommaso"},
+                ],
+                "user_data": "tommaso",
+            },
+            {
+                "content": "2",
+                "label": "click",
+                "features": [
+                    {"group": "topic", "value": "go"},
+                    {"group": "user", "value": "tommaso"},
+                ],
+                "user_data": {"id": "486892", "alias": "Julian"},
+            },
+            {
+                "content": "3",
+                "label": "click",
+                "features": [{"group": "topic", "value": "go"}],
+                "user_data": {"id": "tommaso", "alias": "tommaso"},
+            },
+        ]
+        client.track_engagements(engagements)
+
+    def test_track_impressions(self):
+        impressions = [
+            {
+                "content_list": ["1", "2", "3"],
+                "features": [
+                    {"group": "topic", "value": "js"},
+                    {"group": "user", "value": "tommaso"},
+                ],
+                "user_data": {"id": "tommaso", "alias": "tommaso"},
+            },
+            {
+                "content_list": ["2", "3", "5"],
+                "features": [{"group": "topic", "value": "js"}],
+                "user_data": {"id": "486892", "alias": "Julian"},
+            },
+        ]
+        client.track_impressions(impressions)
+
     def test_og(self):
         response = client.og("https://google.com")
         self.assertTrue("title" in response)