Server SDKv1.xMIT License

Python SDK

acai-analytics is the Python server SDK. It processes events on a background thread, batches automatically, and supports a plugin pipeline for enrichment.

Installation

Shell

pip install acai-analytics

Initialise

Python

from acai.client import Acai
from acai.config import Config

config = Config(
    api_key='YOUR_API_KEY',
    server_url='https://events.yourdomain.com/2/httpapi',
)

client = Acai('YOUR_API_KEY', configuration=config)

Track events

Python

from acai.event import BaseEvent

client.track(
    BaseEvent(
        event_type='Purchase Completed',
        user_id='user_abc123',
        event_properties={
            'product_id': 'prod_123',
            'price': 49.99,
            'currency': 'USD',
        },
    )
)

# Flush buffered events
client.flush()

# Graceful shutdown (flushes then stops background thread)
client.shutdown()

The SDK registers an atexit handler automatically, but calling client.shutdown() explicitly is recommended in long-running services.

Identify users

Python

from acai.event import Identify, EventOptions

identify = Identify()
identify.set('plan', 'enterprise')
identify.set_once('first_seen', '2024-01-15')
identify.add('api_call_count', 1)
identify.append('roles', 'admin')
identify.unset('trial_active')

client.identify(
    identify_obj=identify,
    event_options=EventOptions(user_id='user_abc123'),
)

# Group assignment
client.set_group(
    group_type='company',
    group_name='Acme Corp',
    event_options=EventOptions(user_id='user_abc123'),
)

Plugins

Python

from acai.plugin import PluginType

class ServerContextPlugin:
    plugin_type = PluginType.ENRICHMENT

    def setup(self, client):
        pass

    def execute(self, event):
        event.event_properties = event.event_properties or {}
        event.event_properties['server_region'] = 'us-east-1'
        return event

client.add(ServerContextPlugin())

Config reference

Option	Type	Default	Description
`api_key`	str	required	Your project API key.
`server_url`	str	`None`	Custom collector endpoint.
`flush_queue_size`	int	`200`	Events per batch before flush.
`flush_interval_millis`	int	`10000`	Max wait before flush (ms).
`flush_max_retries`	int	`12`	Max retry attempts per batch.
`use_batch`	bool	`False`	Use the `/batch` endpoint.
`opt_out`	bool	`False`	Disable all event sending.
`min_id_length`	int	`None`	Minimum length for user/device IDs.
`callback`	callable	`None`	Called after each event: `(event, code, message)`.
`logger`	Logger	default	Standard Python logging instance.