Add IdentityMapClient

Author: caroline-ttd

examples/sample_identity_map_client.py

@@ -0,0 +1,38 @@
+import sys
+
+from uid2_client import IdentityMapClient, IdentityMapInput
+
+
+# this sample client takes email addresses or phone numbers as input and generates an IdentityMapResponse object
+# which contains raw uid or the reason why it is unmapped
+
+def _usage():
+    print('Usage: python3 sample_sharing_client.py <base_url> <auth_key> <secret_key> <email_list>', file=sys.stderr)
+    sys.exit(1)
+
+
+if len(sys.argv) <= 4:
+    _usage()
+
+base_url = sys.argv[1]
+auth_key = sys.argv[2]
+secret_key = sys.argv[3]
+email_list = sys.argv[4:]
+first_email = sys.argv[4]
+
+client = IdentityMapClient(base_url, auth_key, secret_key)
+
+identity_map_response = client.generate_identity_map(IdentityMapInput.from_emails(email_list))
+
+mapped_identities = identity_map_response.mapped_identities
+unmapped_identities = identity_map_response.unmapped_identities
+
+mapped_identity = mapped_identities.get(first_email)
+if mapped_identity is not None:
+    raw_uid = mapped_identity.get_raw_id()
+    print('raw_uid =', raw_uid)
+else:
+    unmapped_identity = unmapped_identities.get(first_email)
+    reason = unmapped_identity.get_reason()
+    print('reason =', reason)
+

uid2_client/__init__.py

@@ -24,5 +24,6 @@
 from .encryption_data_response import *
 from .refresh_response import *
 from .uid2_token_generator import *
-
-
+from .identity_map_client import *
+from .identity_map_input import *
+from .identity_map_response import *

uid2_client/identity_map_client.py

@@ -0,0 +1,40 @@
+import base64
+import datetime as dt
+from datetime import timezone
+
+from .identity_map_response import IdentityMapResponse
+
+from uid2_client import auth_headers, make_v2_request, post, parse_v2_response
+
+
+class IdentityMapClient:
+    """Client for interacting with UID2 Identity Map services
+
+        You will need to have the base URL of the endpoint and a client API key
+        and secret to consume web services.
+
+        Methods:
+            generate_identity_map: Generate identity map
+    """
+
+    def __init__(self, base_url, auth_key, secret_key):
+        """Create a new IdentityMapClient client.
+
+        Args:
+            base_url (str): base URL for all requests to UID2 services (e.g. 'https://prod.uidapi.com')
+            auth_key (str): authorization key for consuming the UID2 services
+            secret_key (str): secret key for consuming the UID2 services
+
+        Note:
+            Your authorization key will determine which UID2 services you are allowed to use.
+        """
+        self._base_url = base_url
+        self._auth_key = auth_key
+        self._secret_key = base64.b64decode(secret_key)
+
+    def generate_identity_map(self, identity_map_input):
+        req, nonce = make_v2_request(self._secret_key, dt.datetime.now(tz=timezone.utc),
+                                     identity_map_input.get_identity_map_input_as_json_string().encode())
+        resp = post(self._base_url, '/v2/identity/map', headers=auth_headers(self._auth_key), data=req)
+        resp_body = parse_v2_response(self._secret_key, resp.read(), nonce)
+        return IdentityMapResponse(resp_body, identity_map_input)

uid2_client/identity_map_input.py

@@ -0,0 +1,59 @@
+import json
+
+from uid2_client import IdentityType, normalize_email_string, get_base64_encoded_hash, is_phone_number_normalized
+
+
+class IdentityMapInput:
+    """input for IdentityMapClient, such as email addresses or phone numbers"""
+
+    def __init__(self, identity_type, emails_or_phones, already_hashed):
+        self.hashed_dii_to_raw_diis = {}
+        self.hashed_normalized_emails = []
+        self.hashed_normalized_phones = []
+        if identity_type == IdentityType.Email:
+            for email in emails_or_phones:
+                if already_hashed:
+                    self.hashed_normalized_emails.append(email)
+                else:
+                    normalized_email = normalize_email_string(email)
+                    if normalized_email is None:
+                        raise ValueError("invalid email address")
+                    hashed_normalized_email = get_base64_encoded_hash(normalized_email)
+                    self.hashed_normalized_emails.append(hashed_normalized_email)
+                    self._add_hashed_to_raw_dii_mapping(hashed_normalized_email, email)
+        else:  # phone
+            for phone in emails_or_phones:
+                if already_hashed:
+                    self.hashed_normalized_phones.append(phone)
+                else:
+                    if not is_phone_number_normalized(phone):
+                        raise ValueError("phone number is not normalized: " + phone)
+                    hashed_normalized_phone = get_base64_encoded_hash(phone)
+                    self._add_hashed_to_raw_dii_mapping(hashed_normalized_phone, phone)
+                    self.hashed_normalized_phones.append(hashed_normalized_phone)
+
+    @staticmethod
+    def from_emails(emails):
+        return IdentityMapInput(IdentityType.Email, emails, False)
+
+    @staticmethod
+    def from_phones(phones):
+        return IdentityMapInput(IdentityType.Phone, phones, False)
+
+    @staticmethod
+    def from_hashed_emails(hashed_emails):
+        return IdentityMapInput(IdentityType.Email, hashed_emails, True)
+
+    @staticmethod
+    def from_hashed_phones(hashed_phones):
+        return IdentityMapInput(IdentityType.Phone, hashed_phones, True)
+
+    def _add_hashed_to_raw_dii_mapping(self, hashed_dii, raw_dii):
+        self.hashed_dii_to_raw_diis.setdefault(hashed_dii, []).append(raw_dii)
+
+    def get_identity_map_input_as_json_string(self):
+        json_object = {
+            "email_hash": self.hashed_normalized_emails,
+            "phone_hash": self.hashed_normalized_phones
+        }
+        return json.dumps({k: v for k, v in json_object.items() if v is not None and len(v) > 0})

uid2_client/identity_map_response.py

@@ -0,0 +1,71 @@
+import json
+
+
+class IdentityMapResponse:
+    def __init__(self, response, identity_map_input):
+        self.mapped_identities = {}
+        self.unmapped_identities = {}
+        response_json = json.loads(response)
+        self.status = response_json["status"]
+
+        if not self.is_success():
+            raise ValueError("Got unexpected identity map status: " + self.status)
+
+        body = self._get_body_as_json(response_json)
+
+        for identity in body.get("mapped", []):
+            raw_diis = self._get_raw_diis(identity, identity_map_input)
+            mapped_identity = MappedIdentity.from_json(identity)
+            for raw_dii in raw_diis:
+                self.mapped_identities[raw_dii] = mapped_identity
+
+        for identity in body.get("unmapped", []):
+            raw_diis = self._get_raw_diis(identity, identity_map_input)
+            unmapped_identity = UnmappedIdentity.from_json(identity)
+            for raw_dii in raw_diis:
+                self.unmapped_identities[raw_dii] = unmapped_identity
+
+    @staticmethod
+    def _get_body_as_json(json_response):
+        return json_response["body"]
+
+    @staticmethod
+    def _get_raw_diis(identity, identity_map_input):
+        identifier = identity["identifier"]
+        return identity_map_input.hashed_dii_to_raw_diis[identifier]
+
+    def is_success(self):
+        return self.status == "success"
+
+
+class MappedIdentity:
+    def __init__(self, raw_uid, bucket_id):
+        self.raw_uid = raw_uid
+        self.bucket_id = bucket_id
+
+    def get_raw_id(self):
+        return self.raw_uid
+
+    def get_bucket_id(self):
+        return self.bucket_id
+
+    @staticmethod
+    def from_json(json_obj):
+        return MappedIdentity(
+            json_obj.get("advertising_id"),
+            json_obj.get("bucket_id")
+        )
+
+
+class UnmappedIdentity:
+    def __init__(self, reason):
+        self.reason = reason
+
+    def get_reason(self):
+        return self.reason
+
+    @staticmethod
+    def from_json(json_obj):
+        return UnmappedIdentity(
+            json_obj.get("reason")
+        )