Merge pull request #375 from treasure-data/import_iterable_to_td

[STP-287] Add scenario for importing from Iterable API

scenarios/import_iterable_to_treasuredata/README.md

@@ -0,0 +1,26 @@
+# Workflow: Scenario (Import from Iterable API into Treasure Data)
+
+This scenario shows how you can ingest data not supported by the Iterable connector using TD Workflow Custom Scripts with Iterable API. This example will ingest Iterable Campaigns that are in a Ready state into Treasure Data.
+
+# How to Run
+
+1. Update the destination database and table values in the `import_iterable_to_td.dig` file (`<database_name>` and `<table_name>`).
+
+2. Upload the workflow with TD CLI.
+```
+    $ td wf push import_iterable_to_td
+```
+3. Set the Iterable API Key and Treasure Data API Key as workflow secrets using the `td wf secrets` command.
+```
+    # Set Iterable API Key workflow secret
+    $ td wf secrets --project import_iterable_to_td --set iterable.apikey=<iterable_api_key>
+
+    # Set Treasure Data API Key workflow secret
+    $ td wf secrets --project import_iterable_to_td --set td.apikey=<treasuredata_master_api_key>
+```
+Finally, you can trigger the session manually.
+```
+    # Run
+    $ td wf start import_iterable_to_td import_iterable_to_td --session now
+```
+If you have any questions, contact to support@treasuredata.com.

scenarios/import_iterable_to_treasuredata/import_iterable_to_td.dig

@@ -0,0 +1,9 @@
++process:
+  py>: tasks.process
+  destination_db: <database_name>
+  destination_tbl: <table_name>
+  _env:
+    TD_API_KEY: ${secret:td.apikey}
+    ITERABLE_API_KEY: ${secret:iterable.apikey}
+  docker:
+    image: "digdag/digdag-python:3.9"

scenarios/import_iterable_to_treasuredata/tasks.py

@@ -0,0 +1,19 @@
+import os
+import sys
+import pandas as pd 
+import requests
+import pytd
+
+def process(destination_db, destination_tbl):
+  iterable_apikey = os.environ["ITERABLE_API_KEY"]
+  api_url = 'https://api.iterable.com/api/campaigns'
+
+  headers = {"Api-Key":iterable_apikey}
+  response = requests.get(api_url,headers=headers)
+  response_df = pd.DataFrame(response.json()['campaigns'])
+
+  ready_campaigns = response_df.query("campaignState == 'Ready'")
+
+  td_apikey = os.environ["TD_API_KEY"]
+  client = pytd.Client(apikey=td_apikey,database=destination_db)
+  client.load_table_from_dataframe(ready_campaigns, destination_tbl, if_exists='overwrite')