Cohorly Python SDK

The official server-side Python SDK for Cohorly, a hosted Mixpanel-style product analytics platform. The API mirrors mixpanel-python, so migrating existing code is mostly a matter of swapping the import and pointing at your Cohorly server.

Documentation: Python SDK reference · Quickstart · HTTP API

• Zero runtime dependencies (stdlib urllib only)
• Python 3.8+, fully typed (py.typed)
• Synchronous Consumer and batching BufferedConsumer with the Cohorly retry contract (exponential backoff, Retry-After, bounded queue)

Installation

pip install cohorly

Or from this repo:

pip install ./sdks/python

Quickstart

from cohorly import Cohorly

ch = Cohorly("YOUR_PROJECT_TOKEN", api_host="https://cohorly-service.velloalabs.com")

# Track an event
ch.track("user-1", "Signed Up", {"plan": "pro", "source": "landing"})

# Link an alias to an existing distinct_id
ch.alias("user-1", "anon-7f3a")

# Update a user profile
ch.people_set("user-1", {"$first_name": "Ada", "plan": "pro"})

The project token comes from your Cohorly dashboard (Settings -> Projects). api_host is the base URL of your Cohorly server (default https://cohorly-service.velloalabs.com).

Tracking events

track(distinct_id, event_name, properties=None, meta=None) stamps these default properties before sending:

Property	Value
distinct_id	the id you pass
time	current unix time in milliseconds
$insert_id	random uuid4 hex (server-side dedup)
$lib	"python"
$lib_version	SDK version
token	your project token (stripped by server)

Your properties merge over the defaults, so you may supply a custom time or $insert_id (e.g. for idempotent re-sends):

ch.track("user-1", "Order Completed", {
    "amount": 42.5,
    "$insert_id": f"order-{order.id}",  # dedup key
})

Historical imports

Use import_data to record events with an explicit timestamp (unix milliseconds - Cohorly's convention throughout):

ch.import_data("user-1", "Legacy Signup", 1600000000000, {"source": "csv"})

Unlike Mixpanel there is no separate import endpoint, API secret, or 5-day cutoff - it is the same /track pipeline.

User profiles (people)

ch.people_set("user-1", {"plan": "pro"})          # set/overwrite
ch.people_set_once("user-1", {"created": "..."})  # only if unset
ch.people_increment("user-1", {"logins": 1})      # numeric add
ch.people_unset("user-1", ["plan"])               # remove properties
ch.people_delete("user-1")                        # delete the profile
ch.people_update({"distinct_id": "user-1", "$set": {"x": 1}})  # raw op

These map to the Cohorly /engage operations $set, $set_once, $add, $unset, $delete.

Note: $unset/$delete (so people_unset/people_delete) are destructive and gated server-side: the server refuses them on the project token alone (they need an org-owner or superadmin Authorization credential this SDK does not send) and answers HTTP 200 with {"status": 0, ..., "refused"}. Use the dashboard or the admin privacy API for profile removal.

Feature flags

Flags are evaluated server-side on every call - there is no cache, so a change in the dashboard takes effect on the next call:

if ch.is_feature_enabled("new-checkout", "user-123"):
    ...

ch.get_feature_flag("pricing-test", "user-123")          # "variant-b" or a bool
ch.get_feature_flag_payload("pricing-test", "user-123")  # variant payload or None
ch.get_all_flags("user-123")                             # {key: {enabled, variant, payload, reason}}

Unknown flags read as False / None. A failed evaluation raises CohorlyException (like every other call path here) - wrap the call if you want it to fail open.

Consumers

By default every call sends immediately via a synchronous Consumer. For higher throughput use BufferedConsumer, which batches messages (default 50 per request, server max 500) and implements the Cohorly retry contract:

from cohorly import Cohorly, BufferedConsumer

consumer = BufferedConsumer(max_size=50, api_host="https://cohorly-service.velloalabs.com")
ch = Cohorly("YOUR_PROJECT_TOKEN", consumer=consumer)

for user in users:
    ch.track(user.id, "Backfill Event", {"batch": True})

consumer.flush()  # IMPORTANT: drain remaining messages before exit

Retry behavior (BufferedConsumer):

• 429 / 5xx / network error - the queue is kept and retried with exponential backoff: base 2s, doubling per consecutive failure, capped at 10 minutes, +/-20% jitter. A Retry-After header is honored when present.
• 413 - the flush batch size is halved (floor 1) and retried.
• 400 - the rejected batch is dropped and CohorlyException is raised.
• 401 (invalid token) - the queue is kept; backoff at the maximum delay.
• The in-memory queue is capped at 1000 messages per endpoint; the oldest message is dropped on overflow.

Cohorly batch rejections are atomic (nothing partially inserted), so retrying the same payload is always safe.

The synchronous Consumer(api_host, request_timeout=10, retry_limit=4) retries 429/5xx/network errors inline with the same backoff schedule up to retry_limit times, then raises CohorlyException.

Because the token travels with each message, several Cohorly instances with different project tokens can share one consumer.

Error handling

Delivery failures raise cohorly.CohorlyException:

from cohorly import Cohorly, CohorlyException

try:
    ch.track("user-1", "event")
except CohorlyException as exc:
    log.warning("cohorly delivery failed: %s", exc)

Serialization

Messages are JSON. datetime/date values are serialized to ISO-8601 by the default DatetimeSerializer; pass your own json.JSONEncoder subclass via Cohorly(..., serializer=MyEncoder) for custom types.

Development

cd sdks/python
python3 -m venv .venv
.venv/bin/pip install pytest
.venv/bin/pytest

Tests run against the source tree (no install needed) and use a mocked transport - no network required.