Account Subscription History Solution Recipe: Access historical subscription and suppression data for all profiles in a Klaviyo account with this Python script

Solution Recipes are tutorials to achieve specific objectives in Klaviyo. They can also help you master Klaviyo, learn new third-party technologies, and come up with creative ideas. They are written mainly for developers and technically-advanced users.

This is a guest post written by Shelly Saturné, who recently built this Solution Recipe as part of Klaviyo’s Solution Architect Mentor Program.

Introduction

On a day-to-day basis, Klaviyo strives to improve the experience of those using the platform. One way to do that is to provide Klaviyo users with accurate and timely information regarding a customer profile’s subscription and suppression status.

Today, when you navigate through the platform to view a customer profile, you find the following subscription details for both email and SMS:

• Status
• Method
• Date
• Added by

When a customer profile is suppressed, their profile page includes additional details:

• Red suppression tag
• Suppression details

Although this subscription details box is useful and accessible, it does not provide a timeline of all subscription and suppression events. The details box only shows the most current information relating to a profile’s subscription status.

Klaviyo users want a more holistic view of subscription status, including more detail about how a profile was suppressed, when and how it was unsuppressed, and more. They want access to all subscribe and suppress updates for a single profile in a centralized place.

This does not currently exist in the platform. There’s no log or historic page for subscription changes. Oftentimes, the Klaviyo success team must piece together details to provide Klaviyo users with the holistic story of how and when a profile was suppressed and later unsuppressed. That’s time the success team could be spending in a more productive way to add value to Klaviyo users.

Over the past 3 years, the number of business cases related to suppression and unsubscribe has increased drastically:

• 2021: 0.12% suppression tickets; 0.06% unsubscribe tickets
• 2022: 0.12% suppression tickets; 0.05% unsubscribe tickets
• 2023: 0.40% suppression tickets; 0.18% unsubscribe tickets

Clearly, Klaviyo users want a better understanding of the timeline of subscription and suppression events.

Challenge

To extract the subscription and suppression data as it is tracked on a profile, and display the data in a format that is human-readable and digestible.

Ingredients

• Klaviyo API
• Python IDE
• Postman
• A Klaviyo account
• Private API key

After reviewing the Klaviyo API, we selected the following endpoints to extract the necessary information:

• Profiles API
  • GET https://a.klaviyo.com/api/profiles/
  • Functions in the script to extract data from Profiles endpoint, manipulate the data to filter and return profile id, consent, updated date
• Events API
  • GET https://a.klaviyo.com/api/events/
  • A series of functions pulls metric events from profiles and manipulates the data to extract profile details
• Metrics API
  • GET https://a.klaviyo.com/api/metrics/
  • Multiple metric functions pull all metric and query for event-related data for a specified metric
    • Save as a .csv
      • All filtered data from the API are saved into separate .csv files, then merged into a single .csv file
    • Open .csv in Numbers application
      • Code-generated .csv file is viewable in the Numbers application through a simple right click on the file

Instructions

Step 1

Utilize the Profiles endpoint to render JSON data that contains the profile id, consent, updated date:

def get_profiles():

    baseurl = "https://a.klaviyo.com/api/profiles/?page[size]=100"

    headers = {
        "accept": "application/json",

        "revision": "2023-02-22",

        "Authorization": f"Klaviyo-API-Key {os.getenv('KLAVIYO_PRIVATE_KEY')}"
    }

    response = requests.get(baseurl, headers=headers)

    response_data = response.json()["data"]

    next_url = response.json()["links"].get("next", "")

    counter = 1

    while next_url:

        counter += 1

        print(f"processing page {counter}")

        response = requests.get(next_url, headers=headers)

        response_data += response.json()["data"]

        next_url = response.json()["links"].get("next", "")

    return response_data

Step 2

Step through the JSON data to extract the profile id, consent, and updated date, and add all of the data into an empty list:

def process_profile_data(retrieve_profile_data):

    filtered_profile_data = []

    for retrieve_profile in retrieve_profile_data:

        filtered_profile_data.append({

            "profile_id": retrieve_profile["id"],

            "subscription_updated_date": retrieve_profile.get("attributes",
 
              {}).get("subscriptions", {}).get("email",{}).get("marketing",               

              {}).get("timestamp", ""),                                                                                                  

            "profile_email_consent": retrieve_profile.get("attributes", 

              {}).get("subscriptions", {}).get("email", {}).get("marketing", 
                                                                                                                   
              {}).get("consent", ""),

            "included_profile_firstname": retrieve_profile.get("attributes",

              {}).get("first_name", {}),

            "included_profile_email": retrieve_profile.get("attributes",

              {}).get("email", {})

        })
    return filtered_profile_data

Step 3

Store the profile data from the above function into a new .csv file:

def save_filtered_profile_data_as_csv(filtered_profile_subscription_data):

    with open('subscription_events_profile_data.csv', 'w') as f:

        writer = csv.writer(f)

        writer.writerow(filtered_profile_subscription_data[0].keys())

        for filtered_profile in filtered_profile_subscription_data:

            writer.writerow(filtered_profile.values())

Step 4

Get every metric in the account using the Metrics endpoint:

def get_metric_data():

    url = "https://a.klaviyo.com/api/metrics"

    headers = {

        "accept": "application/json",

        "revision": "2023-02-22",

        "Authorization": f"Klaviyo-API-Key {os.getenv('KLAVIYO_PRIVATE_KEY')}"
    }

    response = requests.get(url, headers=headers)

    return response.json()["data"]

Step 5

Get the entire JSON for all metrics, filter through the JSON object in search of the metric name provided in the parameters, and return that metric’s id:

def process_metric_data_for_metric_id(retrieve_metric_data, metric_name):  

    metric_id = ""

    for metric in retrieve_metric_data:

        if metric.get("attributes", {}).get("name", "") == metric_name:  

            metric_id = metric["id"]

            break

    return metric_id

Step 6

Pass in metric id as a parameter and use it to find every event associated with the metric:

def get_metric_events(metric_id):

    if not metric_id:

        return []

    baseurl = "https://a.klaviyo.com/api/events/?filter=equals(metric_id,\"" + 

metric_id + "\")&fields[profile]=first_name,email&include=profiles"

    headers = {

        "accept": "application/json",

        "revision": "2023-02-22",

        "Authorization": f"Klaviyo-API-Key {os.getenv('KLAVIYO_PRIVATE_KEY')}"
    }


    response = requests.get(baseurl, headers=headers)

    response_data = response.json().get("data",[])

    next_url = response.json()["links"].get("next", "")

    counter = 1

    while next_url:

        counter += 1

        print(f"processing page {counter}")

        response = requests.get(next_url, headers=headers)

        response_data += response.json().get("data",[])

        next_url = response.json()["links"].get("next", "")

    return response_data

Step 7

Filter through the JSON data for the metric’s events to extract the profile id, subscribed timestamp, email consent, profile first name, and profile email:

def filter_specific_metric_data_for_field_data(retrieve_metric_response):

    if not retrieve_metric_response:

        return []

    filtered_metric_data = []

    print(retrieve_metric_response)

    retrieve_metric_data = retrieve_metric_response["data"]

    retrieve_metric_included = retrieve_metric_response["included"]

    for retrieve_metric in retrieve_metric_data:

        data_profile_id = retrieve_metric.get("attributes", {}).get("profile_id", {})

        for retrieve_profile in retrieve_metric_included:

            if data_profile_id == retrieve_profile.get("id"):

                filtered_metric_data.append({

                    "profile_id": retrieve_metric.get("attributes",

                      {}).get("profile_id", {}),

                    "subscription_updated_date": retrieve_metric.get("attributes", 
                  
                      {}).get("datetime", {}),

                    "profile_email_consent": "suppressed",

                    "included_profile_firstname": retrieve_profile.get("attributes",

                      {}).get("first_name", ""),

                    "included_profile_email": retrieve_profile.get("attributes",

                     {}).get("email", "")

                }) 

            break

    return filtered_metric_data

Step 8

Save the metric event data in another new .csv file:

def save_filtered_data_as_csv(filtered_specific_metric_event_data, filename):

    with open(filename, 'w') as f:

        writer = csv.writer(f)

        writer.writerow(filtered_specific_metric_event_data[0].keys())

        for filtered_specific_metric in filtered_specific_metric_event_data:

            writer.writerow(filtered_specific_metric.values())

Step 9

Combine the two .csv files—metric events and profile data—into one:

def merge_lists(list_of_lists):

    merged_lists = []

    for data_list in list_of_lists:

        merged_lists += data_list

    return merged_lists

Step 10

When the main function is called, all the other functions are also called to run and generate the .csv files with the historical subscription data:

def main():

    retrieved_metric_data = get_metric_data()
    

    # Step 1: Retrieve metric ids for all metrics that lead to consent or suppression
    retrieved_metric_unsubscribe_id =    
    process_metric_data_for_metric_id(retrieved_metric_data, "Unsubscribe")  
   
    retrieved_metric_subscribe_id = 
    process_metric_data_for_metric_id(retrieved_metric_data, "Subscribe")
    

    retrieved_metric_unsubscribe_from_list_id = 
    process_metric_data_for_metric_id(retrieved_metric_data,"Unsubscribe from List")                                                                   


    retrieved_metric_marked_spam_id = 
    process_metric_data_for_metric_id(retrieved_metric_data, "Marked Email as Spam")

    retrieved_metric_consented_sms_id =  
    process_metric_data_for_metric_id(retrieved_metric_data,"Consented to Receive 
    SMS")
   

    retrieved_metric_unsubscribed_from_sms_id = 
    process_metric_data_for_metric_id(retrieved_metric_data,"Unsubscribed from SMS")
   


    # Step 2: Get event data for each metric
    retrieved_filter_unsubscribe_metric_data_for_field_data = 
    get_metric_events(retrieved_metric_unsubscribe_id)
    

    retrieved_filter_subscribe_metric_data_for_field_data = 
    get_metric_events(retrieved_metric_subscribe_id)
    
    retrieved_filter_unsubscribe_from_list_metric_data_for_field_data =  
    get_metric_events(retrieved_metric_unsubscribe_from_list_id)
   
    retrieved_filter_marked_as_spam_metric_data_for_field_data = 
    get_metric_events(retrieved_metric_marked_spam_id)
    
    retrieved_filter_consented_sms_metric_data_for_field_data = 
    get_metric_events(retrieved_metric_consented_sms_id)
    
    retrieved_filter_unsubscribed_from_sms_metric_data_for_field_data =  
    get_metric_events(retrieved_metric_unsubscribed_from_sms_id)
    

    # Step 3: Filter event data for consent properties
    filtered_metric_unsubscribe_event_data =
    filter_specific_metric_data_for_field_data(  
    retrieved_filter_unsubscribe_metric_data_f or_field_data)
    

    filtered_metric_subscribe_event_data = 
    filter_specific_metric_data_for_field_data(
    retrieved_filter_subscribe_metric_data_for_field_data)
    
    filtered_metric_unsubscribe_from_list_event_data = 
    filter_specific_metric_data_for_field_data(
    retrieved_filter_unsubscribe_from_list_metric_data_for_field_data)
    

    filtered_metric_marked_as_spam_event_data = 
    filter_specific_metric_data_for_field_data(
    retrieved_filter_marked_as_spam_metric_data_for_field_data)
    
    filtered_metric_consented_sms_event_data =  
    filter_specific_metric_data_for_field_data(
    retrieved_filter_consented_sms_metric_data_for_field_data)
   

    filtered_metric_unsubscribed_from_sms_event_data = 
    filter_specific_metric_data_for_field_data(
    retrieved_filter_unsubscribed_from_sms_metric_data_for_field_data)
    


    # Step 4: Retrieves all profile data
    retrieve_profile_data = get_profiles()
    

    filtered_profile_subscription_data = process_profile_data(retrieve_profile_data)
    

    # Step 5: Combines the lists of all the filtered data
    clean_merged_file = merge_lists([
        filtered_profile_subscription_data,
        filtered_metric_unsubscribe_event_data,
        filtered_metric_subscribe_event_data,
        filtered_metric_unsubscribe_from_list_event_data,
        filtered_metric_marked_as_spam_event_data,
        filtered_metric_consented_sms_event_data,
        filtered_metric_unsubscribed_from_sms_event_data
    ])
    

    # Step 6: Saves data as a csv
    save_filtered_data_as_csv(clean_merged_file, "clean_merged_data.csv")


main()

Step 11

Download the files from Github and run the script for an account requesting an audit trail of subscription events. The files are accessible through this Github repository.

Download the published files for the subscription-suppression log history from Klaviyo’s Github workspace. Open the files in Pycharm, Jupyter Notebook, or any other Python IDE.

Ensure you have the following libraries installed in Pycharm or whichever development environment you prefer: os, csv, requests, and load_dotenv.

All files in the next image should appear in the IDE when you open the file package. If any of them are missing, repeat the download process to make sure you didn’t miss a step:

In the .env file, replace the existing API key with the private API key of the account you’re investigating:

Open the Subscription_log_history.py file and run it:

This should update the clean_merged_data file with the preferred account’s data:

Right click on the file name, clean_merged_data.csv, to open it in the Numbers application:

Impact

By generating a human-readable .csv file, this Solution Recipe equips all Klaviyo employees and users with the assurance and evidence they need to discuss a profile’s subscription history. It also empowers the customer success organization to efficiently handle cases pertaining to subscriptions and suppressions, while cultivating trust among Klaviyo-valued users. Above all, this Solution Recipe enables Klaviyo to consistently uphold its commitment to customer satisfaction by proactively anticipating user requirements and offering readily accessible solutions for these needs.

Learn more

Error handling

If you encounter any errors in the code, we recommend using online resources, such as Stackoverflow, Notion, or the Klaviyo Support team.

Rate limits

Keep in mind that there are rate limits for each endpoint in the code. You can address these limits by practicing pauses in between code runs to ensure that there is enough time between each attempt. Alternatively, you can customize the code to retry automatically when a rate limit occurs.

For the following endpoints, the rate limits are enforced at the following speeds:

• Profiles endpoint: burst of 75/s and steady of 700/m
• Events endpoint: burst of 350/s and steady of 3500/m
• Metrics endpoint: burst of 10/s and steady of 150/m

Want to learn more about Klaviyo’s developer experience and capabilities? Visit the developer portal.

About the Solutions Architect Mentorship Program

At Klaviyo, our solutions architects are a core aspect of the customer experience, both before and after the point of sale. But despite the solutions architect team’s consistent influence throughout the customer lifecycle, it can be difficult to fully articulate what they do and how they do it.

In order to share our learnings and evangelize the multifaceted nature of the role internally, we created the Solutions Architect Mentor Program. Anyone in the company is eligible to receive a mentorship from a Klaviyo solutions architect after a baseline technical evaluation.

Within the mentorship program, mentees can build a prototype solution with the guidance of their mentors and present it to their fellow mentees. We’re excited to showcase these solutions as part of the Klaviyo developer blog and celebrate our mentees’ resourcefulness and creativity.