async-customerio 2.10.0

Async CustomerIO Client - a Python client to interact with CustomerIO in an async fashion.

async-customerio is a lightweight asynchronous client to interact with CustomerIO

• Free software: MIT license
• Requires: Python 3.10+

Features

• Fully async
• Interface preserved as Official Python Client customerio has
• Send push notification
• Send messages (email, SMS, push, inbox)
• App API support (customers, and more to come)

Installation

pip install async-customerio

Getting started

import asyncio

from async_customerio import AsyncCustomerIO, Regions


async def main():
    site_id = "Some-id-gotten-from-CustomerIO"
    api_key = "Some-key-gotten-from-CustomerIO"
    cio = AsyncCustomerIO(site_id, api_key, region=Regions.US)
    await cio.identify(
        id=5,
        email="[e-mail]",
        first_name="John",
        last_name="Doh",
        subscription_plan="premium",
    )
    await cio.track(
        customer_id=5, name="product.purchased", product_sku="XYZ-12345", price=23.45
    )


if __name__ == "__main__":
    asyncio.run(main())

Instantiating AsyncCustomerIO object

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

from async_customerio import AsyncCustomerIO, Regions


cio = AsyncCustomerIO(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.

Custom User-Agent

By default every request is sent with the User-Agent header set to async-customerio/<version>. You can override it via the user_agent parameter:

cio = AsyncCustomerIO(site_id, api_key, user_agent="my-app/1.0")

The same parameter is available on AsyncAPIClient.

Track API v2

The v2 Track API is accessed via the .v2 property on the AsyncCustomerIO instance. It provides typed convenience methods for all person and object operations, sharing the same connection and credentials as the v1 client.

Person operations

import asyncio

from async_customerio import AsyncCustomerIO, Regions


async def main():
    async with AsyncCustomerIO(site_id="site", api_key="key", region=Regions.US) as cio:
        # Identify (create or update) a person
        await cio.v2.identify_person(identifiers={"id": 123}, name="Jane", plan="premium")

        # Track an event
        await cio.v2.track_person_event(identifiers={"id": 123}, name="purchase", amount=49.99)

        # Page view / screen view (mobile)
        await cio.v2.person_pageview(identifiers={"id": 123}, name="/pricing")
        await cio.v2.person_screen(identifiers={"id": 123}, name="home_screen")

        # Device management
        await cio.v2.add_person_device(identifiers={"id": 123}, device_id="tok_abc", platform="ios")
        await cio.v2.delete_person_device(identifiers={"id": 123}, device_id="tok_abc")

        # Suppress / unsuppress
        await cio.v2.suppress_person(identifiers={"id": 123})
        await cio.v2.unsuppress_person(identifiers={"id": 123})

        # Merge two person profiles (secondary is deleted)
        await cio.v2.merge_persons(primary={"id": 123}, secondary={"email": "[e-mail]"})

        # Delete a person
        await cio.v2.delete_person(identifiers={"id": 123})


if __name__ == "__main__":
    asyncio.run(main())

Object operations

async with AsyncCustomerIO(site_id="site", api_key="key") as cio:
    # Identify (create or update) an object
    await cio.v2.identify_object(
        identifiers={"object_type_id": "1", "object_id": "acme"},
        name="Acme Corp",
        industry="Software",
    )

    # Track an event on an object
    await cio.v2.track_object_event(
        identifiers={"object_type_id": "1", "object_id": "acme"},
        name="plan_changed",
    )

    # Delete an object
    await cio.v2.delete_object(identifiers={"object_type_id": "1", "object_id": "acme"})

Relationships

async with AsyncCustomerIO(site_id="site", api_key="key") as cio:
    # Relate a person to an object
    await cio.v2.add_person_relationships(
        identifiers={"id": 123},
        relationships=[{"identifiers": {"object_type_id": "1", "object_id": "acme"}}],
    )

    # Relate an object to people
    await cio.v2.add_object_relationships(
        identifiers={"object_type_id": "1", "object_id": "acme"},
        relationships=[{"identifiers": {"id": 123}}, {"identifiers": {"id": 456}}],
    )

    # Remove relationships
    await cio.v2.delete_person_relationships(
        identifiers={"id": 123},
        relationships=[{"identifiers": {"object_type_id": "1", "object_id": "acme"}}],
    )

Batch operations

from async_customerio.track.v2 import Actions

async with AsyncCustomerIO(site_id="site", api_key="key") as cio:
    batch = [
        {
            "type": "person",
            "action": Actions.identify.value,
            "identifiers": {"id": 123},
            "attributes": {"name": "Jane"},
        },
        {
            "type": "object",
            "action": Actions.identify.value,
            "identifiers": {"object_type_id": "1", "object_id": "acme"},
            "attributes": {"name": "Acme Corp"},
        },
    ]
    await cio.v2.send_batch(batch)

Notes

• All v2 methods validate required parameters and raise AsyncCustomerIOError for missing identifiers, names, etc.
• The API enforces size limits: each item <= 32 KB, whole batch < 500 KB.
• HTTP 200 and 207 are treated as success (methods return None). HTTP 400+ raises AsyncCustomerIOError.
• The legacy cio.send_entity() and cio.send_batch() methods still work for backwards compatibility but delegate to the .v2 class internally.

Sending transactional inbox messages

import asyncio

from async_customerio import AsyncAPIClient, SendInboxMessageRequest, Regions


async def main():
    api = AsyncAPIClient(key="your-app-api-key", region=Regions.US)
    request = SendInboxMessageRequest(
        transactional_message_id="3",
        identifiers={"id": "user_123"},
        message_data={"name": "Jane", "order_id": "1234"},
    )
    response = await api.send_inbox_message(request)
    print(response)


if __name__ == "__main__":
    asyncio.run(main())

App API — Customers

The AsyncAPIClient provides access to the Customer.io App API. Customer endpoints are accessed via the .customers namespace:

import asyncio

from async_customerio import AsyncAPIClient, Regions


async def main():
    async with AsyncAPIClient(key="your-app-api-key", region=Regions.US) as client:
        # Look up customers by email
        result = await client.customers.get_by_email("[e-mail]")

        # Search with filters and pagination
        result = await client.customers.search(
            filter={"and": [{"segment": {"id": 1}}]},
            limit=10,
        )

        # Get customers with attributes by IDs
        result = await client.customers.get_by_ids([1, 2, 3])

        # Look up a single customer's attributes, segments, messages, etc.
        attrs = await client.customers.get_attributes(42)
        segments = await client.customers.get_segments(42)
        messages = await client.customers.get_messages(42, start_ts=1700000000, limit=5)
        activities = await client.customers.get_activities(42, type="event")
        relationships = await client.customers.get_relationships(42)
        prefs = await client.customers.get_subscription_preferences(42)

        # Use id_type to reference by email or cio_id instead of id
        attrs = await client.customers.get_attributes("[e-mail]", id_type="email")


if __name__ == "__main__":
    asyncio.run(main())

App API — Segments

async with AsyncAPIClient(key="your-app-api-key") as client:
    # List all segments
    result = await client.segments.list()

    # Create a manual segment
    result = await client.segments.create("VIP Users", description="High-value customers")

    # Get a segment, its customer count, and dependencies
    segment = await client.segments.get(segment_id=1)
    count = await client.segments.get_customer_count(segment_id=1)
    deps = await client.segments.get_used_by(segment_id=1)

    # List customers in a segment with pagination
    members = await client.segments.get_membership(segment_id=1, limit=50)

    # Delete a segment
    await client.segments.delete(segment_id=1)

Securely verify requests doc

from async_customerio import validate_signature


def main():
    webhook_signing_key = (
        "755781b5e03a973f3405a85474d5a032a60fd56fabaad66039b12eadd83955fa"
    )
    x_cio_timestamp = 1692633432  # header  value
    x_cio_signature = "d7c655389bb364d3e8bdbb6ec18a7f1b6cf91f39bba647554ada78aa61de37b9"  # header value
    body = b'{"key": "value"}'
    if validate_signature(
        signing_key=webhook_signing_key,
        timestamp=x_cio_timestamp,
        request_body=body,
        signature=x_cio_signature,
    ):
        print("Request is sent from CustomerIO")
    else:
        print("Malicious request received")


if __name__ == "__main__":
    main()

License

async-customerio is offered under the MIT license.

Source code

The latest developer version is available in a GitHub repository: https://github.com/healthjoy/async-customerio