Adding end date support for streams for backfill

tap_marketo/__init__.py

@@ -18,7 +18,6 @@
 
 REQUIRED_CONFIG_KEYS = [
     "start_date",
-
     # Log in to Marketo
     # Go to Admin, select Integration->Web Services
     # Endpoint url matches https://123-ABC-456.mktorest.com/rest

tap_marketo/client.py

@@ -113,7 +113,10 @@ def __init__(self, endpoint, client_id, client_secret,
 
         self._session = requests.Session()
         self._use_corona = None
+        self.end_date = kwargs.get("end_date", None)
 
+    def get_end_date(self):
+        return self.end_date
     @property
     def use_corona(self):
         if getattr(self, "_use_corona", None) is None:
@@ -185,8 +188,9 @@ def _request(self, method, url, endpoint_name=None, stream=False, **kwargs):
         endpoint_name = endpoint_name or url
         url = self.get_url(url)
         headers = kwargs.pop("headers", {})
+        params = kwargs.pop("params", {})
         headers.update(self.headers)
-        req = requests.Request(method, url, headers=headers, **kwargs).prepare()
+        req = requests.Request(method, url, headers=headers, params=params, **kwargs).prepare()
         singer.log_info("%s: %s", method, req.url)
         with singer.metrics.http_request_timer(endpoint_name):
             resp = self._session.send(req, stream=stream, timeout=self.request_timeout)
@@ -234,6 +238,14 @@ def request(self, method, url, endpoint_name=None, **kwargs):
 
             return resp
 
+    def get_paging_token(self, sinceDatetime):
+        endpoint = "rest/v1/activities/pagingtoken.json"
+        singer.log_info("Getting paging token for date %s", sinceDatetime)
+
+        params = {"sinceDatetime": sinceDatetime}
+        data = self.request("GET", endpoint, endpoint_name="paging_token", params=params)
+        return data.get("nextPageToken")
+
     def create_export(self, stream_type, fields, query):
         # http://developers.marketo.com/rest-api/bulk-extract/#creating_a_job
         payload = {

tap_marketo/discover.py

@@ -21,7 +21,7 @@
 LISTS_AUTOMATIC_INCLUSION = frozenset(["id", "name", "createdAt", "updatedAt"])
 PROGRAMS_AUTOMATIC_INCLUSION = frozenset(["id", "createdAt", "updatedAt"])
 CAMPAIGNS_AUTOMATIC_INCLUSION = frozenset(["id", "createdAt", "updatedAt"])
-
+DELETED_LEADS_AUTOMATIC_INCLUSION = frozenset(["marketoGUID", "leadId", "activityDate"])
 LEAD_REQUIRED_FIELDS = frozenset(["id", "updatedAt", "createdAt"])
 
 def clean_string(string):
@@ -104,11 +104,11 @@ def get_activity_type_stream(activity):
         primary = clean_string(activity["primaryAttribute"]["name"])
         mdata = metadata.write(mdata, (), 'marketo.primary-attribute-name', primary)
 
-
     if "attributes" in activity:
         for attr in activity["attributes"]:
             attr_name = clean_string(attr["name"])
-            field_schema, mdata = get_schema_for_type(attr["dataType"], breadcrumb=('properties', attr_name), mdata=mdata, null=True)
+            field_schema, mdata = get_schema_for_type(attr["dataType"], breadcrumb=('properties', attr_name),
+                                                      mdata=mdata, null=True)
             if field_schema:
                 properties[attr_name] = field_schema
 
@@ -141,6 +141,11 @@ def discover_activities(client):
 
 
 def discover_leads(client):
+    root = os.path.dirname(os.path.realpath(__file__))
+    path = os.path.join(root, 'schemas/{}.json'.format('leads'))
+    if os.path.isfile(path):
+        discovered_schema = discover_catalog('leads', LEAD_REQUIRED_FIELDS)
+        return discovered_schema
     # http://developers.marketo.com/rest-api/lead-database/leads/#describe
     endpoint = "rest/v1/leads/describe.json"
     data = client.request("GET", endpoint, endpoint_name="leads_discover")
@@ -181,6 +186,35 @@ def discover_leads(client):
     }
 
 
+def discover_activity_deleted_leads(client):
+
+    # Cannot use on-the-fly schema discovery for deleted leads because paging_token is needed
+    # which cannot be fetched without bookmark's state. Bookmark's state depends on discovery itself
+    root = os.path.dirname(os.path.realpath(__file__))
+    path = os.path.join(root, 'schemas/{}.json'.format('deleted_leads'))
+    if os.path.isfile(path):
+        singer.log_info("deleted_leads.json file exists, returning discovered schema")
+        discovered_schema = discover_catalog('deleted_leads', DELETED_LEADS_AUTOMATIC_INCLUSION)
+        return discovered_schema
+    else:
+        return {
+            "tap_stream_id": "deleted_leads",
+            "stream": "deleted_leads",
+            "key_properties": ["marketoGUID"],
+            "metadata": metadata.to_list(metadata.new()),
+            "schema": {
+                "type": "object",
+                "additionalProperties": False,
+                "properties": {
+                    "marketoGUID": {"type": "string"},
+                    "leadId": {"type": "integer"},
+                    "activityDate": {"type": "string", "format": "date-time"},
+                }
+            }
+        }
+
+
+
 def discover_catalog(name, automatic_inclusion, **kwargs):
     unsupported = kwargs.get("unsupported", frozenset([]))
     stream_automatic_inclusion = kwargs.get("stream_automatic_inclusion", False)
@@ -214,6 +248,7 @@ def discover(client):
     streams.append(discover_leads(client))
     streams.append(discover_catalog("activity_types", ACTIVITY_TYPES_AUTOMATIC_INCLUSION, unsupported=ACTIVITY_TYPES_UNSUPPORTED, stream_automatic_inclusion=True))
     streams.extend(discover_activities(client))
+    streams.append(discover_activity_deleted_leads(client))
     streams.append(discover_catalog("campaigns", CAMPAIGNS_AUTOMATIC_INCLUSION))
     streams.append(discover_catalog("lists", LISTS_AUTOMATIC_INCLUSION))
     streams.append(discover_catalog("programs", PROGRAMS_AUTOMATIC_INCLUSION))

tap_marketo/schemas/deleted_leads.json

@@ -0,0 +1,44 @@
+{
+      "tap_stream_id": "deleted_leads",
+      "stream": "deleted_leads",
+      "key_properties": [
+        "marketoGUID"
+      ],
+      "schema": {
+        "type": "object",
+        "additionalProperties": false,
+        "properties": {
+          "marketoGUID": {
+            "type": [
+              "null",
+              "string"
+            ]
+          },
+          "leadId": {
+            "type": [
+              "null",
+              "integer"
+            ]
+          },
+          "activityDate": {
+            "type": [
+              "null",
+              "string"
+            ],
+            "format": "date-time"
+          },
+          "activityTypeId": {
+            "type": [
+              "null",
+              "integer"
+            ]
+          },
+          "campaignId": {
+            "type": [
+              "null",
+              "integer"
+            ]
+          }
+        }
+      }
+    }