Skip to main content


The issue I am facing involves Klaviyo's rate limits, which restrict the number of API calls I can make within specific time frames. I am sending around 22 requests per run (with 10 images), and depending on my tier, I might hit these limits. For example, i dont know if im on the XS tier, but i think i am and you can only make 15 requests per minute. If I exceed this, the API will throttle or block your requests, causing errors. I need to know how to either adjusting my program to stay within these limits or figure out how to get a higher rate limit. Basically my program down below is paired with a flask website that takes in around 10 images to be placed in a new template. That means there are 10 Post requests and 10 Get requests and 1 request to create the template. Help me figure out how to fix.

from flask import Flask, render_template, request
from flask_wtf import FlaskForm
from wtforms import SubmitField, StringField, MultipleFileField
from werkzeug.utils import secure_filename
import os
import json
import requests
from wtforms.validators import InputRequired

app = Flask(__name__)
app.config['SECRET_KEY'] = 'supersecretkey'
app.config['UPLOAD_FOLDER'] = 'uploads'

class UploadFileForm(FlaskForm):
    file = MultipleFileField("File", validators=[InputRequired()])
    apiKey = StringField("String", validators=[InputRequired()])
    tempName = StringField("String", validators=[InputRequired()])
    SendToKlaviyo = SubmitField('Send To Klaviyo')
    SendToHome = SubmitField('Go Back Home')


@app.route('/', methods=['GET',"POST"])
@app.route('/FluidCreatives', methods=['GET',"POST"])
def FluidCreatives():
    numOfPhotos = 0
    form = UploadFileForm()
    clear_upload_folder()
    if 'SendToKlaviyo' in request.form:
        # Check if any files are uploaded (including folder files)
        photos = form.file.data
        for file in photos:
                # If you want to process only image files, check the file extension
                if file.filename.lower().endswith(('png', 'jpg', 'jpeg')):
                    numOfPhotos += 1 
                    print(f"Image file: {file.filename}")
                    # Optionally, save the image to the server
                    file.save(os.path.join(os.path.abspath(os.path.dirname(__file__)),app.config['UPLOAD_FOLDER'],secure_filename(file.filename))) # Then save the file
                else:
                   print(f"Not an image: {file.filename}")
        apiKey = str(form.apiKey.data)
        tempName = str(form.tempName.data)
        responseCode = klaviyoConnect(apiKey, tempName, app.config['UPLOAD_FOLDER'], numOfPhotos)
        if responseCode == 201:
            return render_template('success.html', form=form)
        else:
            return render_template('fail.html', form=form)
    return render_template('index.html', form=form)

def clear_upload_folder():
    """ Clears the upload folder before processing new files. """
    upload_folder = app.config['UPLOAD_FOLDER']
    if os.path.exists(upload_folder):
        # Remove all files in the folder
        for filename in os.listdir(upload_folder):
            file_path = os.path.join(upload_folder, filename)
            try:
                if os.path.isfile(file_path):
                    os.unlink(file_path)  # Remove file
                elif os.path.isdir(file_path):
                    shutil.rmtree(file_path)  # Remove directory and its contents
            except Exception as e:
                print(f"Error removing file {file_path}: {e}")

def klaviyoConnect(api_key: str, template_name: str, folder_path: str, numberOfPhotos: int):
    # Klaviyo API Configuration
    upload_url = "https://a.klaviyo.com/api/image-upload"
    get_image_url = "https://a.klaviyo.com/api/images/"
    template_url = "https://a.klaviyo.com/api/templates/"
    headers = {
        "accept": "application/json",
        "revision": "2024-10-15",
        "Authorization": f"Klaviyo-API-Key {api_key}",
    }

    # Supported image file extensions
    supported_extensions = {".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp"}

    # Get a sorted list of all image files in the folder
    image_files = sorted(
        [f for f in os.listdir(folder_path) if os.path.splitext(f)[1].lower() in supported_extensions]
    )

    # List to store uploaded image URLs
    uploaded_image_urls = []

    # Step 1: Upload Images
    for image_file in image_files:
        file_path = os.path.join(folder_path, image_file)
        print(f"Uploading: {image_file}")

        # Open the image file and prepare the payload
        with open(file_path, "rb") as file:
            files = {"file": file}
            data = {"hidden": "false", "name": image_file}

            # Send the POST request to upload the image
            response = requests.post(upload_url, data=data, files=files, headers=headers)

        # Check the response and fetch the image URL using the image ID
        if response.status_code == 201:
            response_data = response.json()
            image_id = response_data.get("data", {}).get("id")
            if image_id:
                # Fetch the image URL using the image ID
                image_response = requests.get(f"{get_image_url}{image_id}", headers=headers)
                if image_response.status_code == 200:
                    image_url = image_response.json().get("data", {}).get("attributes", {}).get("image_url")
                    if image_url:
                        uploaded_image_urls.append(image_url)
                        print(f"Successfully uploaded: {image_file} -> {image_url}")
                    else:
                        print(f"Failed to get URL for: {image_file}")
                else:
                    print(f"Failed to fetch image details for ID: {image_id}. Status Code: {image_response.status_code}")
            else:
                print(f"Failed to get image ID for: {image_file}")
        else:
            print(f"Failed to upload: {image_file}. Status Code: {response.status_code}")
            print(response.text)

    # Step 2: Create Drag-and-Drop Template with Uploaded Images
    if len(uploaded_image_urls) == numberOfPhotos:
        print("Creating drag-and-drop template with the uploaded images...")

        # Create HTML content with all the uploaded image URLs using the image block structure
        html_content = f"""
        <!DOCTYPE html>
<html>
<body style="word-spacing:normal;background-color:#f7f7f7;">
    <table align="center" width="600" style="border-spacing: 0; margin: 0 auto; padding: 0; width: 600px;">
        {"".join(f'''
        <tr style="margin: 0; padding: 0;">
            <td style="margin: 0; padding: 0;" data-klaviyo-region="true" data-klaviyo-region-width-pixels="600">
                <div class="klaviyo-block klaviyo-image-block" style="margin: 0; padding: 0;">
                    <img src="{url}" alt="Image {i+1}" style="width: 600px; height: auto; display: block; margin: 0; padding: 0;">
                </div>
            </td>
        </tr>
        ''' for i, url in enumerate(uploaded_image_urls))}
    </table>
</body>
</html>
        """

        # JSON payload for creating the template
        payload = {
            "data": {
                "type": "template",
                "attributes": {
                    "name": f"{template_name}",
                    "editor_type": "USER_DRAGGABLE",  # Set as user-draggable
                    "html": html_content,
                    "text": "This is a fallback plain text version of the email.",
                },
            }
        }

        # Send the POST request to create the template
        response = requests.post(template_url, headers={**headers, "content-type": "application/vnd.api+json"}, data=json.dumps(payload))

        # Print the response
        if response.status_code == 201:
            print("Drag-and-drop template created successfully!")
            template_id = response.json().get("data", {}).get("id")
            print(f"Template ID: {template_id}")
            return 201
        else:
            print("Failed to create the template.")
            print(f"Status Code: {response.status_code}")
            print(f"Response: {response.json()}")
            return response.status_code
    else:
        print("Forgetting an Image")
        return 0

if __name__ == "__main__":
    app.run(port=7000, debug=True)

 

To manage Klaviyo's API rate limits effectively, consider the following strategies:

  1. Understand Rate Limits: Klaviyo enforces rate limits per account to maintain system stability. These limits vary by endpoint and are categorized into tiers such as XS, S, M, L, and XL, each with specific burst (per second) and steady (per minute) thresholds. For instance, the XS tier allows 1 request per second and 15 per minute.

    Klaviyo Developers

  2. Optimize API Calls: Given your application's 22 requests per run, you may exceed the XS tier's limits. To mitigate this, implement a queuing system that spaces out requests to stay within the allowed rate. Introducing a delay between requests can help prevent hitting the rate limit. For example, adding a 17-millisecond pause between requests can help avoid rate limits.

    Klaviyo Community

  3. Handle Rate Limit Responses: Monitor API responses for HTTP status code 429, which indicates rate limiting. When this occurs, check the Retry-After header in the response to determine how long to wait before retrying the request. Implementing an exponential backoff strategy with randomized intervals can also help manage retries effectively.

    Klaviyo Developers

  4. Review API Usage: Assess your application's API usage patterns to identify opportunities for reducing the number of requests. Batching multiple operations into a single request where possible or caching frequent data can decrease the demand on the API.

  5. Consult Klaviyo Documentation: Refer to Klaviyo's official API documentation for detailed information on rate limits and best practices for error handling.

    Klaviyo Developers

By implementing these strategies, you can adjust your application's behavior to comply with Klaviyo's API rate limits, ensuring smoother operation and reducing the likelihood of encountering rate limit errors.

Key Strategies:

  1. Optimize API Calls:

    • Batch Requests: Combine multiple operations into fewer requests whenever possible.
    • Cache Responses: Store frequently retrieved data to reduce redundant calls.

  2. Implement Throttling:

    • Use a queue or rate limiter in your application to ensure API calls stay within the allowed thresholds.
    • Add dynamic delays (e.g., based on the Retry-After header) between calls to prevent hitting limits.

  3. Handle Rate Limit Errors (HTTP 429):

    • Retry with Backoff: When you hit a limit, implement exponential backoff and respect the Retry-After header for retry timing.

  4. Monitor and Scale:

    • Regularly monitor your API usage against limits.
    • If you consistently hit limits, consider reaching out to Klaviyo for account-specific rate limit adjustments or optimized solutions.

Practical Example:

If your app performs 22 requests per run and exceeds limits:

  • Introduce a delay (e.g., 50ms between requests) or process in smaller batches.
  • Adjust your logic to handle retries based on API responses dynamically.

Tools and Libraries:

  • Use libraries like axios-retry or retry-axios for automatic retries.
  • Implement rate-limiting libraries such as Bottleneck in Node.js.

These approaches ensure smoother operation without breaching Klaviyo's limits.

Terms & Conditions

Links & Resources

Platforms

Partners

Useful Things

About

  • Terms and Privacy |
    © 2024 Klaviyo All rights reserved. Klaviyo and the Klaviyo logo are trademarks or registered trademarks of Klaviyo, Inc. or its affiliates.