trodo-python 1.0.1

Trodo Analytics SDK for Python — server-side event tracking

trodo-python

Server-side Python SDK for Trodo Analytics. Track backend events, identify users, and manage people/groups — all merging seamlessly with your frontend Trodo data under the same site_id.

Installation

pip install trodo-python

For async support (optional):

pip install trodo-python[async]

Requires Python 3.8+.

Quick Start

import trodo

# Initialize once at app startup
trodo.init(
    site_id='your-site-id',
    debug=False,        # optional: log API calls
    auto_events=True,   # optional: hook sys.excepthook
)

# Get a user context
user = trodo.for_user('user-123')

# Track a custom event
user.track('purchase_completed', {'amount': 99.99, 'plan': 'pro'})

# Identify the user (merges with frontend events under the same identity)
user.identify('[e-mail]')   # distinct_id becomes [e-mail]

# Update people profile
user.people.set({'plan': 'pro', 'company': 'Acme'})

# Track a server-side error
user.capture_error(Exception('payment failed'))

# Flush queued events before process exit
trodo.shutdown()

Flask / FastAPI Example

# Flask
from flask import Flask, request
import trodo

app = Flask(__name__)
trodo.init(site_id='your-site-id')

@app.route('/purchase', methods=['POST'])
def purchase():
    user_id = request.json['user_id']
    user = trodo.for_user(user_id)
    user.track('purchase_completed', {'amount': request.json['amount']})
    return {'ok': True}

Cross-SDK Identity Merging

Frontend and backend events merge when both sides call identify() with the same value:

# Python SDK
user.identify('[e-mail]')    # → [e-mail]

# Browser SDK (same value)
# Trodo.identify('[e-mail]') → [e-mail]

# Both event streams now appear together in the Trodo dashboard

API Reference

trodo.init(config)

Parameter	Type	Default	Description
site_id	str	required	Your Trodo site ID
api_base	str	https://sdkapi.trodo.ai	API base URL
debug	bool	False	Log API requests/responses
auto_events	bool	False	Hook sys.excepthook + threading.excepthook
retries	int	3	HTTP retry attempts on 5xx errors
timeout	int	10	HTTP timeout in seconds
batch_enabled	bool	False	Queue events and flush in bulk
batch_size	int	50	Max events per batch flush
batch_flush_interval	float	5.0	Flush interval in seconds
on_error	callable	None	Callback for SDK errors

trodo.for_user(distinct_id, session_id=None)

Returns a user-bound context. All subsequent calls use this user's session.

user = trodo.for_user(
    'user-123',
    session_id=request.cookies.get('trodo_session'),  # optional: correlate with browser session
)

User Context Methods

user.track(event_name, properties=None)          # Track custom event
user.track_event(event_name, properties=None)    # Alias for track()
user.identify(identify_id)                       # Merge identity
user.wallet_address(address)                     # Set crypto wallet address
user.reset()                                     # Clear session context
user.capture_error(exception)                    # Track server_error event

# People profile
user.people.set(properties)
user.people.set_once(properties)
user.people.unset(keys)
user.people.increment(properties)
user.people.append(properties)
user.people.union(properties)
user.people.remove(properties)
user.people.track_charge(amount, properties=None)
user.people.clear_charges()
user.people.delete_user()

# Groups
user.set_group(group_key, group_id)
user.add_group(group_key, group_id)
user.remove_group(group_key, group_id)
group = user.get_group(group_key, group_id)
group.set(properties)
group.set_once(properties)
group.union(properties)
group.remove(properties)
group.unset(keys)
group.increment(properties)
group.append(properties)
group.delete()

Direct Call Pattern (for pipelines/scripts)

trodo.track('user-123', 'event_name', {'key': 'value'})
trodo.identify('user-123', 'identify_id')
trodo.people_set('user-123', {'plan': 'pro'})
trodo.set_group('user-123', 'company', 'acme')

Global Methods

trodo.enable_auto_events()    # Enable sys.excepthook hooks
trodo.disable_auto_events()   # Disable hooks
trodo.flush()                 # Flush pending batch queue
trodo.shutdown()              # Flush + stop background timers

Auto Events

When auto_events=True, the SDK wraps Python's exception hooks and sends server_error events to Trodo:

• sys.excepthook — unhandled exceptions in the main thread
• threading.excepthook — unhandled exceptions in threads

These events use distinct_id: 'server_global' in the dashboard.

Per-user error capture: user.capture_error(e) uses the user's own distinct_id.

Batching

trodo.init(
    site_id='your-site-id',
    batch_enabled=True,
    batch_size=100,
    batch_flush_interval=3.0,
)

# Events are queued and flushed every 3s or when 100 events accumulate
user.track('page_view')

# Always flush before process exit
import atexit
atexit.register(trodo.shutdown)

Thread Safety

The SDK is thread-safe. SessionManager, EventQueue, and BatchFlusher all use threading.Lock internally. Safe to use in multi-threaded Flask/Django/FastAPI applications.

License

ISC