Source LinkedIn Ads: add conversions stream

airbyte-integrations/connectors/source-linkedin-ads/source_linkedin_ads/schemas/conversions.json

@@ -0,0 +1,72 @@
+{
+  "$schema": "https://json-schema.org/draft-07/schema#",
+  "type": ["null", "object"],
+  "title": "Conversions",
+  "properties": {
+    "attributionType": {
+      "type": ["null", "string"]
+    },
+    "account": {
+      "type": ["null", "string"]
+    },
+    "campaigns": {
+      "type": ["null", "array"],
+      "items": {
+        "type": ["null", "string"]
+      }
+    },
+    "created": {
+      "type": ["null", "integer"]
+    },
+    "enabled": {
+      "type": ["null", "boolean"]
+    },
+    "id": {
+      "type": ["null", "integer"]
+    },
+    "imagePixelTag": {
+      "type": ["null", "string"]
+    },
+    "name": {
+      "type": ["null", "string"]
+    },
+    "type": {
+      "type": ["null", "boolean"]
+    },
+    "latestFirstPartyCallbackAt": {
+      "type": ["null", "integer"]
+    },
+    "postClickAttributionWindowSize": {
+      "type": ["null", "integer"]
+    },
+    "viewThroughAttributionWindowSize": {
+      "type": ["null", "integer"]
+    },
+    "lastCallbackAt": {
+      "type": ["null", "integer"]
+    },
+    "lastModified": {
+      "type": ["null", "integer"]
+    },
+    "value": {
+      "type": ["null", "object"],
+      "properties": {
+        "amount": {
+          "type": ["null", "string"]
+        },
+        "currencyCode": {
+          "type": ["null", "string"]
+        }
+      }
+    },
+    "associatedCampaigns": {
+      "type": ["null", "array"]
+    },
+    "urlMatchRuleExpression": {
+      "type": ["null", "array"]
+    },
+    "urlRules": {
+      "type": ["null", "array"]
+    }
+  }
+}

airbyte-integrations/connectors/source-linkedin-ads/source_linkedin_ads/source.py

@@ -29,6 +29,7 @@
     CampaignGroups,
     Campaigns,
     Creatives,
+    Conversions,
 )
 
 logger = logging.getLogger("airbyte")
@@ -105,6 +106,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             CampaignGroups(config=config),
             Campaigns(config=config),
             Creatives(config=config),
+            Conversions(config=config),
         ]
 
         return streams + self.get_custom_ad_analytics_reports(config)

airbyte-integrations/connectors/source-linkedin-ads/source_linkedin_ads/streams.py

@@ -354,6 +354,44 @@ def get_updated_state(self, current_stream_state: MutableMapping[str, Any], late
         return {self.cursor_field: max(latest_record.get(self.cursor_field), int(current_stream_state.get(self.cursor_field)))}
 
 
+class Conversions(LinkedInAdsStreamSlicing):
+    """
+    Get Conversions data using `account_id` slicing.
+    https://learn.microsoft.com/en-us/linkedin/marketing/integrations/ads-reporting/conversion-tracking?view=li-lms-2023-05&tabs=curl#find-conversions-by-ad-account
+    """
+
+    endpoint = "conversions"
+    search_param = "account"
+
+    def path(
+        self,
+        *,
+        stream_state: Mapping[str, Any] = None,
+        stream_slice: Mapping[str, Any] = None,
+        next_page_token: Mapping[str, Any] = None,
+    ) -> str:
+        return f"{stream_slice.get('account_id')}/{self.endpoint}"
+
+    def request_headers(
+        self, stream_state: Mapping[str, Any], stream_slice: Mapping[str, Any] = None, next_page_token: Mapping[str, Any] = None
+    ) -> Mapping[str, Any]:
+        headers = super().request_headers(stream_state, stream_slice, next_page_token)
+        headers.update({"X-Restli-Protocol-Version": "2.0.0"})
+        return headers
+
+    def request_params(
+        self,
+        stream_state: Mapping[str, Any],
+        stream_slice: Mapping[str, Any] = None,
+        next_page_token: Mapping[str, Any] = None,
+    ) -> MutableMapping[str, Any]:
+        params = super().request_params(stream_state, stream_slice, next_page_token)
+        params["q"] = self.search_param
+        params["accounts"] = f"urn:li:sponsoredAccount:{stream_slice.get('account_id')}"
+
+        return urlencode(params, safe="():,%")
+
+
 class LinkedInAdsAnalyticsStream(IncrementalLinkedinAdsStream, ABC):
     """
     AdAnalytics Streams more info: