Klaviyo API.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "f0a821a4-b4e5-4284-92ce-5259f3ec2984",
   "metadata": {},
   "source": [
    "# Klaviyo API\n",
    "\n",
    "Our goal is to create a flattened table of email campaign data, with their audiences, tags, and applicable metrics.\\\n",
    "There are a few transformations below outside of the calls, mostly related to creating variables like end dates, pivoting output for merging into the main df, and string cleanup for misc characters.\n",
    "\n",
    "### The API Sum Of Parts\n",
    "\n",
    "* Campaigns\n",
    "  * [Get Campaigns](https://developers.klaviyo.com/en/reference/get_campaigns)\n",
    "* Audiences (Segments & Lists)\n",
    "  * [Get Segments](https://developers.klaviyo.com/en/reference/get_segments)\n",
    "  * [Get Lists](https://developers.klaviyo.com/en/reference/get_lists)\n",
    "* Tags\n",
    "  * [Get Tags](https://developers.klaviyo.com/en/reference/get_campaign_tags)\n",
    "  * [Get Tag Groups](https://developers.klaviyo.com/en/reference/get_tag_groups)\n",
    "  * [Get Campaign Relationship Tags](https://developers.klaviyo.com/en/reference/get_campaign_relationships_tags)\n",
    "      * You can alternatively use [Get Campaign Tags](https://developers.klaviyo.com/en/reference/get_campaign_tags), but I wanted to avoid string splitting.\n",
    "* Metrics\n",
    "  * [Get Metrics](https://developers.klaviyo.com/en/reference/get_metrics)\n",
    "  * [Query Metric Aggregates](https://developers.klaviyo.com/en/reference/query_metric_aggregates)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a219389-ee3d-4ecd-9425-f427b6f71ce9",
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "import time\n",
    "import pandas as pd\n",
    "from datetime import datetime, timedelta\n",
    "from config import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "29aff672-49b2-4260-8ed1-ad2fe56fa704",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set target database/brand.\n",
    "database = \"CT\"\n",
    "\n",
    "# Assigning private_key based on db value.\n",
    "if database == \"CT\":\n",
    "    private_key = ct_private_key\n",
    "    connection_string = ct_connection_string\n",
    "elif database == \"KP\":\n",
    "    private_key = kp_private_key\n",
    "    connection_string = kp_connection_string\n",
    "elif database == \"ST\":\n",
    "    private_key = st_private_key\n",
    "    connection_string = st_connection_string\n",
    "elif database == \"WC\":\n",
    "    private_key = wc_private_key\n",
    "    connection_string = wc_connection_string"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cad0ff7f-eb2a-44a1-a21a-96f683948ad2",
   "metadata": {},
   "source": [
    "### Campaigns \n",
    "\n",
    "Not all data is neatly appended like in the documentation, it's just links for subsequent API calls to get these fields.\\\n",
    "We will only need to pull in: campaign_id, campaign_name, status, audiences, send_strategy, scheduled, sent.\n",
    "\n",
    "We are also adding a column called end_date, 14 days from sent, to mark when the attribution window closes.\n",
    "\n",
    "_Note: A/B tests will group all metrics for all sub-mailings. You will need to make a specialized call to break these apart into their own campaigns._\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a061f989-425f-4f08-b07c-51dc863f779b",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Pull all in all email campaign ids.\n",
    "\n",
    "url = \"https://a.klaviyo.com/api/campaigns/?filter=equals(messages.channel,'email')&include=campaign-messages\" # Need to use &include= to get campaign text details. \n",
    "headers = {\n",
    "    \"accept\": \"application/json\",\n",
    "    \"revision\": \"2023-10-15\",\n",
    "    \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "}\n",
    "\n",
    "campaigns = [] # Campaign data.\n",
    "included_items = [] # Campaign text details.\n",
    "\n",
    "while url:\n",
    "    response = requests.get(url, headers=headers)\n",
    "\n",
    "    if response.status_code == 200:\n",
    "        data = response.json()\n",
    "        campaigns += [campaign for campaign in data['data'] if campaign['type'] == 'campaign']\n",
    "        included_items += data.get('included', [])\n",
    "        url = data['links'].get('next')  # Get the next page URL\n",
    "    else:\n",
    "        print('Failed to retrieve campaigns: ', response.status_code)\n",
    "        break\n",
    "\n",
    "# Output is heavily nested, and contains a lot of useless meta data.\n",
    "# Just going to cherry pick the values we need into a flattened table.\n",
    "df_campaigns = pd.DataFrame([{\n",
    "    'campaign_id': campaign['id'],\n",
    "    'campaign_name': campaign['attributes']['name'],\n",
    "    'status': campaign['attributes']['status'],\n",
    "    'audiences': campaign['attributes']['audiences']['included'],\n",
    "    'send_strategy': campaign['attributes']['send_strategy']['method'],\n",
    "    'scheduled': campaign['attributes']['scheduled_at'], # Used for metric capture window.\n",
    "    'sent': campaign['attributes']['send_time'], # Effectively useless depending on smart send times.\n",
    "    'subject': next((item['attributes']['content']['subject'] for item in included_items if item['type'] == 'campaign-message' and any(rel['id'] == item['id'] for rel in campaign['relationships']['campaign-messages']['data'])), None),\n",
    "    'preview_text': next((item['attributes']['content']['preview_text'] for item in included_items if item['type'] == 'campaign-message' and any(rel['id'] == item['id'] for rel in campaign['relationships']['campaign-messages']['data'])), None),\n",
    "} for campaign in campaigns])\n",
    "\n",
    "# Adding an end date to mark when Klaviyo attribution window ends.\n",
    "# Will be referenced later when calling for aggregate metrics.\n",
    "df_campaigns['end_date'] = pd.to_datetime(df_campaigns['sent'])\n",
    "df_campaigns['end_date'] = df_campaigns['end_date'].apply(lambda x: (x + pd.Timedelta(days=14)).isoformat() if pd.notna(x) else None)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85404e2c-dd43-4d1a-8b38-b6b8913cf722",
   "metadata": {},
   "source": [
    "### Limiting Campaigns\n",
    "\n",
    "We are pulling in _a lot_ of camaigns, so we are going to limit to the last 20 days from scheduled.\\\n",
    "There are nuances to the differences in scheduled/sent when it comes to \"smart send\", so we'll use this one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f3526798-d494-47ce-893b-cfa751d25bfd",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#Doing a bit of clean up. There are a lot of subsequent API calls, limiting to recent campaigns speeds this up.\n",
    "\n",
    "# Limit to only sent emails. No need to pull in email drafts.\n",
    "df_campaigns = df_campaigns[df_campaigns['status'] != 'Draft']\n",
    "\n",
    "# Get the date string for x days ago.\n",
    "date_str = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%dT%H:%M:%S')\n",
    "\n",
    "# Filter rows where 'scheduled' >= date_str.\n",
    "df_campaigns = df_campaigns[df_campaigns['sent'] >= date_str]\n",
    "df_campaigns.reset_index(drop=True, inplace=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "10bc50bf-70d3-455a-84d2-64ff45e8c22b",
   "metadata": {},
   "source": [
    "### Audiences\n",
    "\n",
    "The column 'audiences' in campaigns is a combination of segments and lists.\\\n",
    "They are functionally the same thing, but you need to make two calls to get the names of each id, break apart the strings in campaigns, and append to campaign ids.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0237d0f6-afff-428f-ad3b-251214030155",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pull all segments.\n",
    "url = \"https://a.klaviyo.com/api/segments/\"\n",
    "headers = {\n",
    "    \"accept\": \"application/json\",\n",
    "    \"revision\": \"2023-10-15\",\n",
    "    \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "}\n",
    "\n",
    "# API only allows 10 at a time, need to paginate.\n",
    "\n",
    "segments = []\n",
    "\n",
    "while url:\n",
    "    response = requests.get(url, headers=headers)\n",
    "\n",
    "    if response.status_code == 200:\n",
    "        data = response.json()\n",
    "        segments += [segment for segment in data['data'] if segment['type'] == 'segment']\n",
    "        url = data['links'].get('next')  # Get the next page URL\n",
    "    else:\n",
    "        print('Failed to retrieve segments: ', response.status_code)\n",
    "        break\n",
    "\n",
    "# Converting to DataFrame\n",
    "df_segments = pd.DataFrame([{\n",
    "    'audience_id': segment['id'],\n",
    "    'audience_name': segment['attributes']['name'],\n",
    "} for segment in segments])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5c0140b6-b96d-46e3-b924-29f50400b7cb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pull all lists.\n",
    "url = \"https://a.klaviyo.com/api/lists/\"\n",
    "headers = {\n",
    "    \"accept\": \"application/json\",\n",
    "    \"revision\": \"2023-10-15\",\n",
    "    \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "}\n",
    "\n",
    "# API only allows 10 at a time, need to paginate.\n",
    "\n",
    "lists = []\n",
    "\n",
    "while url:\n",
    "    response = requests.get(url, headers=headers)\n",
    "\n",
    "    if response.status_code == 200:\n",
    "        data = response.json()\n",
    "        lists += [list for list in data['data'] if list['type'] == 'list']\n",
    "        url = data['links'].get('next')  # Get the next page URL\n",
    "    else:\n",
    "        print('Failed to retrieve lists: ', response.status_code)\n",
    "        break\n",
    "\n",
    "# Converting to DataFrame\n",
    "df_lists = pd.DataFrame([{\n",
    "    'audience_id': list['id'],\n",
    "    'audience_name': list['attributes']['name'],\n",
    "} for list in lists])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7bb3e121-cb68-4973-88de-6cc752f109d1",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Combine lists and segments into one data frame.\n",
    "df_audience = pd.concat([df_segments, df_lists], ignore_index=True)\n",
    "\n",
    "# Cleaning the 'audiences' column in df_campaigns\n",
    "df_campaigns['audiences'] = df_campaigns['audiences'].astype(str).str.replace(r'[\\[\\]\\'\"]', '', regex=True)\n",
    "\n",
    "# Splitting the audience IDs into lists, handling None, empty or invalid formats in df_campaigns\n",
    "df_campaigns['audiences'] = df_campaigns['audiences'].apply(lambda x: x.split(',') if x and x.strip() else [])\n",
    "\n",
    "# Creating a dictionary from df_audience for mapping\n",
    "audience_id_name_map = dict(zip(df_audience['audience_id'], df_audience['audience_name']))\n",
    "\n",
    "# Function to replace IDs with names and concatenate in df_campaigns\n",
    "def replace_ids_with_names(ids):\n",
    "    if not ids:\n",
    "        return None\n",
    "    names = [audience_id_name_map.get(id.strip()) for id in ids if id.strip() in audience_id_name_map]\n",
    "    return ', '.join(filter(None, names))  # filter(None, names) removes any None values\n",
    "\n",
    "# Applying the function to the 'audiences' column in df_campaigns\n",
    "df_campaigns['audiences'] = df_campaigns['audiences'].apply(replace_ids_with_names)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "989de3f9-f2c1-4334-b9c6-d151107f3e1f",
   "metadata": {},
   "source": [
    "### Tags\n",
    "\n",
    "You need to make a call to get tag names, the tag group names, and then create a table merging them all together.\\\n",
    "Once this is done, we are making a call to Get Campaign Relationship Tags one-by-one for all campaigns, this should return associated tags.\\\n",
    "I think there might be an easier way to achieve this, but I opted to go this route."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eaeb48e2-b085-4cb5-a905-b4a306680c7c",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Pull all tags\n",
    "url = \"https://a.klaviyo.com/api/tags/\"\n",
    "headers = {\n",
    "    \"accept\": \"application/json\",\n",
    "    \"revision\": \"2023-10-15\",\n",
    "    \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "}\n",
    "\n",
    "response = requests.get(url, headers=headers)\n",
    "\n",
    "# Request test\n",
    "if response.status_code == 200:\n",
    "    response = response.json()\n",
    "else:\n",
    "    print('Failed to retrieve campaigns: ', response.status_code) \n",
    "    \n",
    "# Extracting metric data\n",
    "tags = [tag for tag in response['data'] if tag['type'] == 'tag']\n",
    "\n",
    "# Converting to DataFrame\n",
    "df_tags = pd.DataFrame([{\n",
    "    'tag_id': tag['id'],\n",
    "    'tag_name': tag['attributes']['name'],\n",
    "    'tag_group_id': tag['relationships']['tag-group']['data']['id']\n",
    "} for tag in tags])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "284e6478-291c-48b8-8733-09cde1698144",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pull all tag groups\n",
    "\n",
    "url = \"https://a.klaviyo.com/api/tag-groups/\"\n",
    "headers = {\n",
    "    \"accept\": \"application/json\",\n",
    "    \"revision\": \"2023-10-15\",\n",
    "    \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "}\n",
    "\n",
    "response = requests.get(url, headers=headers)\n",
    "\n",
    "# Request test\n",
    "if response.status_code == 200:\n",
    "    response = response.json()\n",
    "else:\n",
    "    print('Failed to retrieve tag groups: ', response.status_code) \n",
    "    \n",
    "# Extracting metric data\n",
    "tag_groups = [tag_group for tag_group in response['data'] if tag_group['type'] == 'tag-group']\n",
    "\n",
    "# Converting to DataFrame\n",
    "df_tag_groups = pd.DataFrame([{\n",
    "    'tag_group_id': tag_group['id'],\n",
    "    'tag_group_name': tag_group['attributes']['name']\n",
    "} for tag_group in tag_groups])\n",
    "\n",
    "# Pulling in only tag groups we want.\n",
    "df_tag_groups = df_tag_groups[df_tag_groups['tag_group_name'].isin(['Content', 'Format', 'Program/Audience', 'Sales'])]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9291c050-d032-4c81-9f01-e56d8d7499e0",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Merge tags with tag groupings.\n",
    "df_tag_table = df_tags.merge(df_tag_groups, on='tag_group_id', how='left')\n",
    "# Drop null/old tags that don't have a group assigned.\n",
    "df_tag_table.dropna(subset=['tag_group_name'], inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3990222d-706d-48f1-a4c8-4e66cc917178",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to call API with entire row and process response\n",
    "def call_api_for_tags(campaign_row, private_key):\n",
    "    # Extracting values from the row\n",
    "    campaign_id = campaign_row['campaign_id']\n",
    "    campaign_name = campaign_row['campaign_name']\n",
    "\n",
    "    # API call setup\n",
    "    url = \"https://a.klaviyo.com/api/campaigns/\" + campaign_id + \"/relationships/tags/\"\n",
    "    headers = {\n",
    "        \"accept\": \"application/json\",\n",
    "        \"revision\": \"2023-10-15\",\n",
    "        \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "    }\n",
    "\n",
    "    # Making the API call\n",
    "    response = requests.get(url, headers=headers)\n",
    "    if response.status_code == 200:\n",
    "        response = response.json()\n",
    "        campaign_tags = [tag for tag in response['data'] if tag['type'] == 'tag']\n",
    "        return pd.DataFrame([{\n",
    "            'campaign_id': campaign_id,\n",
    "            'campaign_name': campaign_name,\n",
    "            'tag_id': tag['id']\n",
    "        } for tag in campaign_tags])\n",
    "    else:\n",
    "        print('Failed to retrieve tags: ', response.status_code)\n",
    "        return pd.DataFrame()\n",
    "\n",
    "# DataFrame to store results\n",
    "df_campaign_tags = pd.DataFrame()\n",
    "\n",
    "# Iterate over each row in df_campaigns\n",
    "for index, row in df_campaigns.iterrows():\n",
    "    df_temp = call_api_for_tags(row, private_key)\n",
    "    df_campaign_tags = pd.concat([df_campaign_tags, df_temp], ignore_index=True)\n",
    "    \n",
    "    # Rate limiting: Burst 3/s, Steady 60/m; appears to be a lie.\n",
    "    time.sleep(1)\n",
    "\n",
    "# Merge tags with tag groupings.\n",
    "df_tag_table = df_campaign_tags.merge(df_tag_table, on='tag_id', how='left')\n",
    "\n",
    "# Grouping by 'campaign_id' and 'tag_group_name', then joining the 'tag_name's\n",
    "grouped = df_tag_table.groupby(['campaign_id', 'tag_group_name'])['tag_name'].apply(','.join).reset_index()\n",
    "\n",
    "# Pivoting the table to get the desired format\n",
    "pivot = grouped.pivot(index='campaign_id', columns='tag_group_name', values='tag_name').reset_index()\n",
    "\n",
    "# Merging with df_campaigns\n",
    "df_campaigns = df_campaigns.merge(pivot, on='campaign_id', how='left')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cc2a8be8-0f07-41d2-9680-769a61fc0a30",
   "metadata": {},
   "source": [
    "### Metrics\n",
    "\n",
    "Here we are pulling in all of the metric name and ids.\\\n",
    "Each brand/account has a different id, and is often polluted with old tags, so we are specifiying the standardized ones we use.\\\n",
    "Once the metrics are put into a table, we are recursively calling all for metrics for each campaign in the main table, pivoting the output, and merging into the main campaigns table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2e633374-979e-4db7-87ee-bf246a70a434",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Pull in unique metric IDs.\n",
    "\n",
    "url = \"https://a.klaviyo.com/api/metrics/\"\n",
    "headers = {\n",
    "    \"accept\": \"application/json\",\n",
    "    \"revision\": \"2023-10-15\",\n",
    "    \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "}\n",
    "\n",
    "response = requests.get(url, headers=headers)\n",
    "    \n",
    "# Request test\n",
    "if response.status_code == 200:\n",
    "    response = response.json()\n",
    "else:\n",
    "    print('Failed to retrieve metrics: ', response.status_code) \n",
    "    \n",
    "# Output is heavily nested, and contains a lot of useless meta data.\n",
    "# Just going to cherry pick the values we need into a flattened table.\n",
    "metrics = [item for item in response['data'] if item['type'] == 'metric']\n",
    "\n",
    "# Converting to DataFrame\n",
    "df_metrics = pd.DataFrame([{\n",
    "    'metric_id': metric['id'],\n",
    "    'metric_name': metric['attributes']['name']\n",
    "} for metric in metrics])\n",
    "\n",
    "# ID's below are: received email, opened email, clicked email, unsubscribed.\n",
    "# Each brand has different/unique metric_id's, need to use the default names to filter what we want.\n",
    "df_metrics = df_metrics[df_metrics['metric_name'].isin(['Received Email', 'Opened Email', 'Clicked Email', 'Clicked email to unsubscribe'])]\n",
    "\n",
    "#Klaviyo changed the name of this column to allow different kinds of unsubscribed.\n",
    "df_metrics['metric_name'] = df_metrics['metric_name'].replace('Clicked email to unsubscribe', 'Unsubscribed')\n",
    "\n",
    "df_metrics.reset_index(drop=True, inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a4ee3465-95cf-4102-abd1-33646d615f48",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Klaviyo requires that you call every metric for every campaign.\n",
    "# Going to have to call every row in df_campaign, and cycle through every df_metric id.\n",
    "\n",
    "def call_api_for_metrics(campaign_row, metric_row, private_key):\n",
    "    \n",
    "    campaign_id = campaign_row['campaign_id']\n",
    "    campaign_name = campaign_row['campaign_name']\n",
    "    scheduled = campaign_row['scheduled']\n",
    "    end_date = campaign_row['end_date']\n",
    "\n",
    "    metric_id = metric_row['metric_id']\n",
    "    metric_name = metric_row['metric_name']\n",
    "\n",
    "    # API URL\n",
    "    url = \"https://a.klaviyo.com/api/metric-aggregates/\"\n",
    "\n",
    "    # Preparing the payload\n",
    "    payload = { \"data\": {\n",
    "            \"type\": \"metric-aggregate\",\n",
    "            \"attributes\": {\n",
    "                \"metric_id\": metric_id,\n",
    "                \"measurements\": [\"unique\"], # Can use: sum_value, count, unique\n",
    "                \"interval\": \"month\", # Can use: hour, day, week, month\n",
    "                \"by\": [\"$message\"],\n",
    "                \"filter\": [\n",
    "                    \"greater-or-equal(datetime,\" + scheduled + \")\",  \n",
    "                    \"less-than(datetime,\" + end_date + \")\",  \n",
    "                    \"equals($message,\\\"\" + campaign_id+ \"\\\")\"],\n",
    "                \"timezone\": \"US/Pacific\"\n",
    "            }\n",
    "        } }\n",
    "\n",
    "    # Headers for the API request\n",
    "    headers = {\n",
    "        \"accept\": \"application/json\",\n",
    "        \"revision\": \"2023-10-15\",\n",
    "        \"content-type\": \"application/json\",\n",
    "        \"Authorization\": \"Klaviyo-API-Key \" + private_key\n",
    "    }\n",
    "\n",
    "    # Making the API request\n",
    "    response = requests.post(url, json=payload, headers=headers)\n",
    "\n",
    "    # Check if the request was successful\n",
    "    if response.status_code == 200:\n",
    "        response_data = response.json()\n",
    "    else:\n",
    "        print('Failed to retrieve data: ', response.status_code)\n",
    "        return pd.DataFrame()\n",
    "\n",
    "    # Extracting metric data\n",
    "    metric_aggregate = response_data['data']\n",
    "\n",
    "    # Preparing data for DataFrame\n",
    "    data_for_df = []\n",
    "    dates = metric_aggregate['attributes']['dates']\n",
    "    measurements = metric_aggregate['attributes']['data'][0]['measurements']['unique']\n",
    "\n",
    "    for date, measurement in zip(dates, measurements):\n",
    "        data_for_df.append({\n",
    "            'campaign_id': campaign_id,\n",
    "            'campaign_name': campaign_name,\n",
    "            'metric_id': metric_id,\n",
    "            'metric_name': metric_name,\n",
    "            'date': date,\n",
    "            'aggregate': measurement\n",
    "        })\n",
    "\n",
    "    return pd.DataFrame(data_for_df)\n",
    "\n",
    "def main(df_campaigns, df_metrics, private_key):\n",
    "    final_output = pd.DataFrame()\n",
    "\n",
    "    for campaign_index, campaign_row in df_campaigns.iterrows():\n",
    "        for metric_index, metric_row in df_metrics.iterrows():\n",
    "            result = call_api_for_metrics(campaign_row, metric_row, private_key)\n",
    "            final_output = pd.concat([final_output, result])\n",
    "\n",
    "            # Rate limiting: Burst 10/s, Steady 150/m; Actual seems to be 60-75/m.\n",
    "            time.sleep(1)\n",
    "\n",
    "    return final_output\n",
    "\n",
    "# Call main()\n",
    "metric_output = main(df_campaigns, df_metrics, private_key)\n",
    "\n",
    "# Remove date columns, sum aggregate values to get a readable table.\n",
    "metric_output = metric_output.groupby(['campaign_id', 'campaign_name', 'metric_id', 'metric_name'])['aggregate'].sum().reset_index()\n",
    "\n",
    "# Pivot the output, merge into df_campaigns.\n",
    "pivot = metric_output.pivot(index='campaign_id', columns='metric_name', values='aggregate')\n",
    "df_campaigns = df_campaigns.merge(pivot, on='campaign_id', how='left')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dd3b70c9-e9f8-4408-b0be-f7d2803fb793",
   "metadata": {},
   "source": [
    "### Storage\n",
    "\n",
    "df_campaigns has now been transformed, and is ready to store however you see fit.\n",
    "\n",
    "This obviously is not the best way to go about upserting fields into our campaigns table,\\\n",
    "...but is substantially easier."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "98c96727-e7aa-4065-ba6b-d829d3bcf491",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# We are just going to pull all existing records, merge with new records, all in a dataframe.\n",
    "import pyodbc\n",
    "from sqlalchemy.engine import URL\n",
    "connection_url = URL.create(\"mssql+pyodbc\", query={\"odbc_connect\": connection_string})\n",
    "\n",
    "from sqlalchemy import create_engine\n",
    "engine = create_engine(connection_url)\n",
    "\n",
    "# Select all records from existing table.\n",
    "query = (\"SELECT * FROM \" + database + \".klaviyo.Campaigns\")\n",
    " \n",
    "# Run Query.\n",
    "df_historical = pd.read_sql(query, engine)\n",
    "\n",
    "# Merge new with old\n",
    "df_merged = pd.concat([df_historical, df_campaigns]).drop_duplicates(subset='campaign_id', keep='last').reset_index(drop=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ea7e0e08-9169-4065-9452-438b9a471750",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Replace and insert into the Campaigns table.\n",
    "from sqlalchemy.types import NVARCHAR\n",
    "cursor = engine.connect()\n",
    "\n",
    "default_length = 255  # You can adjust this as needed\n",
    "df_merged.to_sql(name='Campaigns', con=engine, schema='klaviyo', if_exists='replace', index=False, \n",
    "                    dtype={column_name: NVARCHAR(default_length) for column_name in df_merged.columns})"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}