feat: support the new private key

customerio/analytics/request.py

@@ -3,6 +3,7 @@
 from gzip import GzipFile
 import logging
 import json
+import re
 from dateutil.tz import tzutc
 from requests.auth import HTTPBasicAuth
 from requests import sessions
@@ -12,20 +13,28 @@
 
 _session = sessions.Session()
 
+_API_KEY_PREFIX = re.compile(r'^(ak|wk)_(us|eu)_')
+
 
 def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs):
     """Post the `kwargs` to the API"""
     log = logging.getLogger('customerio')
     body = kwargs
     body["sentAt"] = datetime.utcnow().replace(tzinfo=tzutc()).isoformat()
-    url = remove_trailing_slash(host or 'https://cdp.customer.io') + '/v1/batch'
-    auth = HTTPBasicAuth(write_key, '')
+    api_key = _API_KEY_PREFIX.match(write_key)
+    default_host = 'https://cdp.customer.io'
+    if api_key and api_key.group(2) == 'eu':
+        default_host = 'https://cdp-eu.customer.io'
+    url = remove_trailing_slash(host or default_host) + '/v1/batch'
+    auth = None if api_key else HTTPBasicAuth(write_key, '')
     data = json.dumps(body, cls=DatetimeSerializer)
     log.debug('making request: %s', data)
     headers = {
         'Content-Type': 'application/json',
         'User-Agent': 'analytics-python/' + VERSION
     }
+    if api_key:
+        headers['Authorization'] = 'Bearer ' + write_key
     if gzip:
         headers['Content-Encoding'] = 'gzip'
         buf = BytesIO()

customerio/analytics/test/request.py

@@ -84,3 +84,53 @@ def test_no_proxies_by_default(self):
             mock_post.assert_called_once()
             _, call_kwargs = mock_post.call_args
             self.assertNotIn('proxies', call_kwargs)
+
+    def _post_and_get_call(self, write_key, **kwargs):
+        mock_response = mock.Mock()
+        mock_response.status_code = 200
+        with mock.patch('customerio.analytics.request._session.post',
+                        return_value=mock_response) as mock_post:
+            post(write_key, batch=[{
+                'userId': 'userId',
+                'event': 'python event',
+                'type': 'track',
+            }], **kwargs)
+            mock_post.assert_called_once()
+            return mock_post.call_args
+
+    def test_secret_api_key_uses_bearer(self):
+        key = 'ak_us_' + 'a' * 32 + '_abcdef'
+        _, call_kwargs = self._post_and_get_call(key)
+        self.assertEqual(call_kwargs['headers']['Authorization'],
+                         'Bearer ' + key)
+        self.assertIsNone(call_kwargs['auth'])
+
+    def test_public_api_key_uses_bearer(self):
+        key = 'wk_us_' + 'a' * 32 + '_abcdef'
+        _, call_kwargs = self._post_and_get_call(key)
+        self.assertEqual(call_kwargs['headers']['Authorization'],
+                         'Bearer ' + key)
+        self.assertIsNone(call_kwargs['auth'])
+
+    def test_legacy_write_key_uses_basic(self):
+        call_args, call_kwargs = self._post_and_get_call('testsecret')
+        self.assertEqual(call_args[0], 'https://cdp.customer.io/v1/batch')
+        self.assertNotIn('Authorization', call_kwargs['headers'])
+        self.assertEqual(call_kwargs['auth'].username, 'testsecret')
+        self.assertEqual(call_kwargs['auth'].password, '')
+
+    def test_eu_api_key_uses_eu_host(self):
+        key = 'ak_eu_' + 'a' * 32 + '_abcdef'
+        call_args, _ = self._post_and_get_call(key)
+        self.assertEqual(call_args[0], 'https://cdp-eu.customer.io/v1/batch')
+
+    def test_us_api_key_uses_us_host(self):
+        key = 'ak_us_' + 'a' * 32 + '_abcdef'
+        call_args, _ = self._post_and_get_call(key)
+        self.assertEqual(call_args[0], 'https://cdp.customer.io/v1/batch')
+
+    def test_explicit_host_wins_over_api_key(self):
+        key = 'ak_eu_' + 'a' * 32 + '_abcdef'
+        call_args, _ = self._post_and_get_call(
+            key, host='https://example.com')
+        self.assertEqual(call_args[0], 'https://example.com/v1/batch')