Customer.io Python

This module is tested with Python 3.10 through 3.14. If you're new to Customer.io, we recommend that you integrate with our Data Pipelines Python library instead.

Installing

pip install customerio

Usage

from customerio import CustomerIO, Regions

cio = CustomerIO(site_id, api_key, region=Regions.US)
cio.identify(id="5", email="customer@example.com", name="Bob", plan="premium")
cio.track(customer_id="5", name="purchased")
cio.track(customer_id="5", name="purchased", data={"price": 23.45})

Instantiating customer.io object

Create an instance of the client with your Customer.io credentials.

from customerio import CustomerIO, Regions

cio = CustomerIO(site_id, api_key, region=Regions.US)

region is optional and takes one of two values—Regions.US or Regions.EU. If you do not specify your region, we assume that your account is based in the US (Regions.US). If your account is based in the EU and you do not provide the correct region (Regions.EU), we'll route requests to our EU data centers accordingly, however this may cause data to be logged in the US.

Create or update a Customer.io customer profile

cio.identify(id="5", email="customer@example.com", name="Bob", plan="premium")

Only the id field is used to identify the customer here. Using an existing id with a different email (or any other attribute) will update/overwrite any pre-existing values for that field.

You can pass any keyword arguments to the identify method. These kwargs will be converted to custom attributes.

See original REST documentation here

Track a custom event

cio.track(customer_id="5", name="purchased")

Track a custom event with custom data values

cio.track(customer_id="5", name="purchased", data={"price": 23.45, "product": "widget"})

Pass custom event attributes to track in the data dict.

See original REST documentation here

Track a custom event with an event id or timestamp

cio.track(
    customer_id="5",
    name="purchased",
    data={"price": 23.45, "product": "widget"},
    id="01HB4HBDKTFWYZCK01DMRSWRFD",
    timestamp=1561231234,
)

Pass id to provide a unique event identifier for deduplication. Pass timestamp to set the event time. These fields are sent as top-level event fields, not as custom attributes in data.

Backfill a custom event

from datetime import datetime, timedelta

customer_id = "5"
event_type = "purchase"

# Backfill an event one hour in the past
event_date = datetime.utcnow() - timedelta(hours=1)
cio.backfill(customer_id, event_type, event_date, price=23.45, coupon=True)

event_timestamp = 1408482633
cio.backfill(customer_id, event_type, event_timestamp, price=34.56)

event_timestamp = "1408482680"
cio.backfill(customer_id, event_type, event_timestamp, price=45.67)

Event timestamp may be passed as a datetime.datetime object, an integer or a string UNIX timestamp

Keyword arguments to backfill are converted to custom event attributes.

See original REST documentation here

Track an anonymous event

cio.track_anonymous(
    anonymous_id="anon-event", name="purchased", data={"price": 23.45, "product": "widget"}
)

An anonymous event is an event associated with a person you haven't identified. The event requires an anonymous_id representing the unknown person and an event name. When you identify a person, you can set their anonymous_id attribute. If event merging is turned on in your workspace, and the attribute matches the anonymous_id in one or more events that were logged within the last 30 days, we associate those events with the person.

Like track, track_anonymous accepts custom event attributes in data and optional top-level id and timestamp fields.

Anonymous invite events

If you previously sent invite events, you can achieve the same functionality by sending an anonymous event with the anonymous identifier set to None. To send anonymous invites, your event must include a recipient attribute.

cio.track_anonymous(
    anonymous_id=None,
    name="invite",
    data={"first_name": "alex", "recipient": "alex.person@example.com"},
)

Delete a customer profile

cio.delete(customer_id="5")

Deletes the customer profile for a specified customer.

This method returns nothing. Attempts to delete non-existent customers will not raise any errors.

See original REST documentation here

You can pass any keyword arguments to the identify method. These kwargs will be converted to custom attributes.

Merge duplicate customer profiles

When you merge two people, you pick a primary person and merge a secondary, duplicate person into it. The primary person remains after the merge and the secondary is deleted. This process is permanent: you cannot recover the secondary person.

For each person, you'll set the type of identifier you want to use to identify a person—one of id, email, or cio_id—and then you'll provide the corresponding identifier.

## Please import identifier types
cio.merge_customers(
    primary_id_type=ID,
    primary_id="cool.person@company.com",
    secondary_id_type=EMAIL,
    secondary_id="cperson@gmail.com",
)

Add a device

cio.add_device(customer_id="1", device_id="device_hash", platform="ios")

Adds the device device_hash with the platform ios for a specified customer.

Supported platforms are ios and android.

Optionally, last_used can be passed in to specify the last touch of the device. Otherwise, this attribute is set by the API.

cio.add_device(customer_id="1", device_id='device_hash', platform='ios', last_used=1514764800})

This method returns nothing.

Delete a device

cio.delete_device(customer_id="1", device_id="device_hash")

Deletes the specified device for a specified customer.

This method returns nothing. Attempts to delete non-existent devices will not raise any errors.

Suppress a customer

cio.suppress(customer_id="1")

Suppresses the specified customer. They will be deleted from Customer.io, and we will ignore all further attempts to identify or track activity for the suppressed customer ID

See REST documentation here

Unsuppress a customer

cio.unsuppress(customer_id="1")

Unsuppresses the specified customer. We will remove the supplied id from our suppression list and start accepting new identify and track calls for the customer as normal

See REST documentation here

Send Transactional Messages

To use the Transactional API, instantiate the Customer.io object using an app key and create a request object for your message type.

Email

SendEmailRequest requires:

• transactional_message_id: the ID of the transactional message you want to send, or the body, _from, and subject of a new message.
• to: the email address of your recipients
• an identifiers object containing the email and/or id of your recipient. If the person you reference by email or ID does not exist, Customer.io creates them.
• a message_data object containing properties that you want reference in your message using Liquid.
• You can also send attachments with your message. Use attach to encode attachments.

Use send_email referencing your request to send a transactional message. Learn more about transactional messages and SendEmailRequest properties.

from customerio import APIClient, Regions, SendEmailRequest

client = APIClient("your API key", region=Regions.US)

request = SendEmailRequest(
    to="person@example.com",
    _from="override.sender@example.com",
    transactional_message_id="3",
    message_data={
        "name": "person",
        "items": [
            {
                "name": "shoes",
                "price": "59.99",
            },
        ],
    },
    identifiers={
        "email": "person@example.com",
    },
)

with open("receipt.pdf", "rb") as f:
    request.attach("receipt.pdf", f.read())

response = client.send_email(request)
print(response)

Push

SendPushRequest requires:

• transactional_message_id: the ID of the transactional push message you want to send.
• an identifiers object containing the id or email of your recipient. If the profile does not exist, Customer.io will create it.

Use send_push referencing your request to send a transactional message. Learn more about transactional messages and SendPushRequest properties.

from customerio import APIClient, Regions, SendPushRequest

client = APIClient("your API key", region=Regions.US)

request = SendPushRequest(
    transactional_message_id="3",
    message_data={
        "name": "person",
        "items": [
            {
                "name": "shoes",
                "price": "59.99",
            },
        ],
    },
    identifiers={
        "id": "2",
    },
)

response = client.send_push(request)
print(response)

WhatsApp

SendWhatsAppRequest requires:

• transactional_message_id: the ID of the transactional WhatsApp message you want to send.
• an identifiers object containing the id or email of your recipient. If the profile does not exist, Customer.io will create it.

to is the WhatsApp number in E.164 format.

Use send_whatsapp referencing your request to send a transactional message. Learn more about transactional messages and SendWhatsAppRequest properties.

from customerio import APIClient, Regions, SendWhatsAppRequest

client = APIClient("your API key", region=Regions.US)

request = SendWhatsAppRequest(
    transactional_message_id="3",
    to="+15551234567",
    message_data={
        "name": "person",
    },
    identifiers={
        "id": "2",
    },
)

response = client.send_whatsapp(request)
print(response)

Notes

• The Customer.io Python SDK depends on the Requests library which includes urllib3 as a transitive dependency. The Requests library leverages connection pooling defined in urllib3. urllib3 only attempts to retry invocations of HTTP methods which are understood to be idempotent (See: Retry.DEFAULT_ALLOWED_METHODS). Since the POST method is not considered to be idempotent, any invocations which require POST are not retried.

• It is possible to have the Customer.io Python SDK effectively disable connection pooling by passing a named initialization parameter use_connection_pooling to either the APIClient class or CustomerIO class. Setting this parameter to False (default: True) causes the Session to be initialized and discarded after each request. If you are experiencing integration issues where the cause is reported as Connection Reset by Peer, this may correct the problem. It will, however, impose a slight performance penalty as the TCP connection set-up and tear-down will now occur for each request.

Usage Example Disabling Connection Pooling

from customerio import CustomerIO, Regions

cio = CustomerIO(site_id, api_key, region=Regions.US, use_connection_pooling=False)

Running tests

Changes to the library can be tested by running make test from the parent directory.

Thanks!

• Dimitriy Narkevich for creating the library.
• EZL for contributing customer deletes and improving README
• Noemi Millman for adding custom JSON encoder
• Jason Kraus for event backfilling
• Nicolas Paris for better handling of NaN values