Add conversion tag and conversion event models (#69) (#70)

* Add conversions models
* Lint

integration_tests/ads/test_conversion_events.py

@@ -0,0 +1,91 @@
+"""
+Test Conversion Model
+"""
+import os as _os
+from integration_tests.base_test import BaseTestCase
+from integration_tests.config import DEFAULT_AD_ACCOUNT_ID
+
+from pinterest.client import PinterestSDKClient
+from pinterest.ads.conversion_events import Conversion
+
+class TestSendConversionEvent(BaseTestCase):
+    """
+    Test send Conversion Event
+    """
+
+    def test_send_conversion_success(self):
+        """
+        Test send ConversionEvent successfully
+        """
+        client = PinterestSDKClient.create_client_with_token(_os.environ.get('CONVERSION_ACCESS_TOKEN'))
+
+        NUMBER_OF_CONVERSION_EVENTS = 2
+        raw_user_data = dict(
+            em = ["964bbaf162703657e787eb4455197c8b35c18940c75980b0285619fe9b8acec8"] #random hash256
+        )
+        conversion_events = [
+            Conversion.create_conversion_event(
+                event_name = "add_to_cart",
+                action_source = "app_ios",
+                event_time = 1670026573,
+                event_id = "eventId0001",
+                user_data = raw_user_data,
+            )
+            for _ in range(NUMBER_OF_CONVERSION_EVENTS)
+        ]
+
+        response = Conversion.send_conversion_events(
+            client = client,
+            ad_account_id = DEFAULT_AD_ACCOUNT_ID,
+            conversion_events = conversion_events,
+            test = True,
+        )
+
+        assert response
+        assert response.num_events_received == 2
+        assert response.num_events_processed == 2
+        assert len(response.events) == 2
+
+        assert response.events[0].status == "processed"
+        assert response.events[0].error_message == ""
+        assert response.events[0].warning_message == ""
+
+        assert response.events[1].status == "processed"
+        assert response.events[1].error_message == ""
+        assert response.events[1].warning_message == ""
+
+    def test_send_conversion_fail(self):
+        """
+        Test send ConversionEvent fail with non-hashed email
+        """
+        client = PinterestSDKClient.create_client_with_token(_os.environ.get('CONVERSION_ACCESS_TOKEN'))
+
+        NUMBER_OF_CONVERSION_EVENTS = 2
+        raw_user_data = dict(
+            em = ["[email protected]"]
+        )
+        conversion_events = [
+            Conversion.create_conversion_event(
+                event_name = "add_to_cart",
+                action_source = "app_ios",
+                event_time = 1670026573,
+                event_id = "eventId0001",
+                user_data = raw_user_data,
+            )
+            for _ in range(NUMBER_OF_CONVERSION_EVENTS)
+        ]
+
+        response = Conversion.send_conversion_events(
+            client = client,
+            ad_account_id = DEFAULT_AD_ACCOUNT_ID,
+            conversion_events = conversion_events,
+            test = True,
+        )
+
+        assert response
+        assert response.num_events_received == 2
+        assert response.num_events_processed == 2
+        assert len(response.events) == 2
+
+        assert response.events[0].warning_message == "'em' is not in sha256 hashed format."
+        assert response.events[1].warning_message == "'em' is not in sha256 hashed format."

integration_tests/ads/test_conversion_tags.py

@@ -0,0 +1,99 @@
+"""
+Test Conversion Tag Model
+"""
+
+from integration_tests.base_test import BaseTestCase
+from integration_tests.config import DEFAULT_AD_ACCOUNT_ID
+
+from pinterest.ads.conversion_tags import ConversionTag
+
+class TestCreateConversionTag(BaseTestCase):
+    """
+    Test creating Conversion Tag
+    """
+
+    def test_create_conversion_tag_success(self):
+        """
+        Test creating a new Conversion Tag successfully
+        """
+        conversion_tag = ConversionTag.create(
+            ad_account_id = DEFAULT_AD_ACCOUNT_ID,
+            name = "Test Conversion Tag"
+        )
+
+        assert conversion_tag
+        assert getattr(conversion_tag, "_id")
+        assert getattr(conversion_tag, "_name") == "Test Conversion Tag"
+
+class TestGetConversionTag(BaseTestCase):
+    """
+    Test get Conversion Tag
+    """
+
+    def test_get_conversion_tag_success(self):
+        """
+        Test get conversion tag from existing conversion tag
+        """
+        exiting_conversion_tag_id = self.conversion_tag_utils.get_conversion_tag_id()
+        conversion_tag = ConversionTag(
+            client = self.test_client,
+            ad_account_id = DEFAULT_AD_ACCOUNT_ID,
+            conversion_tag_id = exiting_conversion_tag_id,
+            )
+
+        assert conversion_tag
+        assert getattr(conversion_tag, "_id")
+        assert getattr(conversion_tag, "_name") == getattr(self.conversion_tag_utils.get_conversion_tag(), "_name")
+
+class TestGetListConversionTag(BaseTestCase):
+    """
+    Test get list of ConversionTags
+    """
+    def test_get_list_success(self):
+        """
+        Test get list successfully
+        """
+        # Create new account so the integration test does not get slow as number of conversion
+        # tags increasing while testing
+        ad_account_id = getattr(self.ad_account_utils.create_new_ad_account(), "_id")
+
+        NUMBER_OF_NEW_CONVERSION_TAG = 3
+        for _ in range(NUMBER_OF_NEW_CONVERSION_TAG):
+            self.conversion_tag_utils.create_new_conversion_tag(
+                name = "SDK_TEST_CONVERSION_TAG",
+                ad_account_id = ad_account_id,
+            )
+
+        conversion_tags = ConversionTag.get_all(ad_account_id = ad_account_id)
+
+        assert len(conversion_tags) == NUMBER_OF_NEW_CONVERSION_TAG
+
+class TestGetPageVsitConversionTag(BaseTestCase):
+    """
+    Test get page visit conversion tag events
+    """
+    def test_get_page_visit_success(self):
+        """
+        Test get page visit converion tag events for an Ad Account
+        """
+        conversion_tag_events, bookmark = ConversionTag.get_page_visit_conversion_tag_events(
+            ad_account_id = DEFAULT_AD_ACCOUNT_ID
+        )
+
+        assert not conversion_tag_events
+        assert not bookmark
+
+class TestGetOcpmEligibleConversionTag(BaseTestCase):
+    """
+    Test get ocpm eligible conversion tag events
+    """
+    def test_get_ocpm_eligible_conversion_tags(self):
+        """
+        Test get ocpm eligible conversion tag events for an Ad Account
+        """
+        property, conversion_tag_events = ConversionTag.get_ocpm_eligible_conversion_tag_events(
+            ad_account_id = DEFAULT_AD_ACCOUNT_ID
+        )
+
+        assert not property
+        assert not conversion_tag_events

pinterest/ads/conversion_events.py

@@ -0,0 +1,116 @@
+"""
+Conversion Event Class for Pinterest Python SDK
+"""
+from __future__ import annotations
+
+from openapi_generated.pinterest_client.api.conversion_events_api import ConversionEventsApi
+from openapi_generated.pinterest_client.model.conversion_events import ConversionEvents
+from openapi_generated.pinterest_client.model.conversion_events_data import ConversionEventsData
+from openapi_generated.pinterest_client.model.conversion_api_response_events import ConversionApiResponseEvents
+from openapi_generated.pinterest_client.model.conversion_events_user_data import ConversionEventsUserData
+
+from pinterest.client import PinterestSDKClient
+from pinterest.utils.base_model import PinterestBaseModel
+
+class Conversion(PinterestBaseModel):
+    # pylint: disable=too-many-locals
+    """
+    Conversion Event Model used to send conversion events to Pinterest API
+    """
+
+    @classmethod
+    def create_conversion_event(
+        cls,
+        event_name : str,
+        action_source : str,
+        event_time : int,
+        event_id : str,
+        user_data : dict,
+        event_source_url : str = None,
+        partner_name : str = None,
+        app_id : str = None,
+        app_name : str = None,
+        app_version : str = None,
+        device_brand : str = None,
+        device_carrier : str = None,
+        device_model : str = None,
+        device_type : str = None,
+        os_version : str = None,
+        language : str = None,
+        **kwargs
+    ) -> ConversionEventsData:
+        """ Create Conversion Event Data to be sent
+
+        Args:
+            event_name (str): The type of the user event, Enum: "add_to_cart", "checkout", "custom",
+                "lead", "page_visit", "search", "signup", "view_category", "watch_video"
+            action_source (str): The source indicating where the conversion event occurred, Enum:
+                "app_adroid", "app_ios", "web", "offline"
+            event_time (int): The time when the event happened. Unix timestamp in seconds
+            event_id (str): The unique id string that identifies this event and can be used for deduping
+                between events ingested via both the conversion API and Pinterest tracking
+            user_data (dict): Object containing customer information data. Note, it is required at least
+                one of 1) em, 2) hashed_maids or 3) pair client_ip_address + client_user_agent.
+            event_source_url (str, optional): URL of the web conversion event
+            partner_name (str, optional): The third party partner name responsible to send the event to
+                Conversion API on behalf of the adverstiser. Only send this field if Pinterest has worked
+                directly with you to define a value for partner_name.
+            app_id (str, optional): The app store app ID.
+            app_name (str, optional): Name of the app.
+            app_version (str, optional): Version of the app.
+            device_brand (str, optional): Brand of the user device.
+            device_carrier (str, optional): User device's model carrier.
+            device_model (str, optional): Model of the user device.
+            device_type (str, optional): Type of the user device.
+            os_version (str, optional): Version of the device operating system.
+            language (str, optional): Two-character ISO-639-1 language code indicating the user's language.
+
+        Returns:
+            ConversionEventsData: ConversionEventData to be sent
+        """
+        return ConversionEventsData(
+            event_name = event_name,
+            action_source = action_source,
+            event_time = event_time,
+            event_id = event_id,
+            user_data = ConversionEventsUserData(**user_data),
+            event_source_url = event_source_url,
+            partner_name = partner_name,
+            app_id = app_id,
+            app_name = app_name,
+            app_version = app_version,
+            device_brand = device_brand,
+            device_carrier = device_carrier,
+            device_model = device_model,
+            device_type = device_type,
+            os_version = os_version,
+            language = language,
+            **kwargs,
+        )
+
+    @classmethod
+    def send_conversion_events(
+        cls,
+        ad_account_id : str,
+        conversion_events : list[ConversionEventsData],
+        test : bool = False,
+        client : PinterestSDKClient = None,
+        **kwargs,
+    )-> tuple(int, int, list[ConversionApiResponseEvents]):
+        """
+        Send conversion events to Pinterest API for Conversions
+
+        Note: Highly recommend to use create_client_with_token (with Conversion Access Token) to create new client
+        for this functionality.
+        """
+
+        response = ConversionEventsApi(api_client=cls._get_client(client)).events_create(
+            ad_account_id = str(ad_account_id),
+            conversion_events = ConversionEvents(
+                data = conversion_events
+                ),
+            test = test,
+            **kwargs,
+        )
+
+        return response