getfluxly (Python)

Python SDK

Server-side ingest from any Python runtime. Mirrors the Node SDK's shape and batching defaults. Sync Client and AsyncClient exposed from the same package.

Install

pip install getfluxly

Supports Python 3.10+. The only runtime dependency is httpx.

Quick start (sync)

from getfluxly import Client

client = Client(token="gflux_secret_yourtoken")

client.track("subscription_started",
             external_id="user_42",
             properties={"plan": "pro"})

client.identify(external_id="user_42",
                traits={"email": "x@y.com", "plan": "pro"})

client.alias(user_id="user_42",
             anonymous_id="anon_a8f3c2")

client.flush()
client.shutdown()  # also runs from atexit

Quick start (async)

from getfluxly import AsyncClient

async with AsyncClient(token="gflux_secret_yourtoken") as ac:
    await ac.track("subscription_started",
                   external_id="user_42",
                   properties={"plan": "pro"})
    await ac.identify(external_id="user_42",
                      traits={"email": "x@y.com"})
    await ac.alias(user_id="user_42",
                   anonymous_id="anon_a8f3c2")

The async with block calls await ac.shutdown() on exit, flushing any buffered events.

track()

track() is the primary ingest call and works end-to-end.

client.track(
    "invoice_paid",
    external_id="user_42",
    properties={
        "amount": 49.00,
        "currency": "usd",
        "plan": "pro",
    },
)

Argument	Type	Required	Notes
`event`	`str`	yes	Event name
`external_id`	`str`	yes	Your stable user identifier
`anonymous_id`	`str`	no	Pre-identification device ID
`properties`	`dict`	no	Arbitrary key/value pairs
`timestamp`	`datetime`	no	Defaults to `datetime.utcnow()`

identify()

Experimental: identify() currently sends the event name $identify but the server expects identify, so end-to-end identity stitching does not yet work. Use track() with your own identify event as a workaround until the fix ships.

The method signature is available for forward compatibility:

client.identify(
    external_id="user_42",
    traits={
        "email": "x@y.com",
        "plan": "pro",
        "created_at": "2026-01-15T00:00:00Z",
    },
)

Argument	Type	Required	Notes
`external_id`	`str`	yes	Stable user identifier
`anonymous_id`	`str`	no	Link an anonymous session
`traits`	`dict`	no	User-level attributes

Workaround until the fix ships:

client.track("user_identified",
             external_id="user_42",
             properties={"email": "x@y.com", "plan": "pro"})

alias()

Links two identifiers in GetFluxly so that events previously attributed to anonymous_id are merged into the user_id profile.

client.alias(
    user_id="user_42",
    anonymous_id="anon_a8f3c2",
)

Argument	Type	Required	Notes
`user_id`	`str`	yes	Canonical, post-auth identifier
`anonymous_id`	`str`	yes	Pre-auth device/session ID to merge

AsyncClient

AsyncClient has the same surface as Client but every method is a coroutine. Use it as a context manager so shutdown is guaranteed:

import asyncio
from getfluxly import AsyncClient

async def run():
    async with AsyncClient(token="gflux_secret_yourtoken") as ac:
        await ac.track("page_viewed",
                       external_id="user_42",
                       properties={"path": "/pricing"})
        await ac.alias(user_id="user_42",
                       anonymous_id="anon_a8f3c2")
    # shutdown + flush called automatically on exit

asyncio.run(run())

You can also manage lifecycle manually:

ac = AsyncClient(token="gflux_secret_yourtoken")
await ac.track("page_viewed", external_id="user_42")
await ac.flush()
await ac.shutdown()

Configuration

All options mirror the Node SDK so that observability across SDKs uses one mental model.

Option	Default	Notes
`flush_at`	20	Events queued before a forced flush
`flush_interval`	5.0	Periodic flush cadence in seconds
`max_retries`	2	Per failed batch
`timeout`	5.0	Per HTTP request in seconds
`max_queue_size`	1000	Hard cap; exceeding it raises `queue_overflow`

flush_interval is the cadence at which buffered events are sent even when flush_at has not been reached. There is no background flusher thread; the flush is triggered on the next call after the interval has elapsed.

Retry-After is honored. Each batch carries a unique X-Idempotency-Key that survives retries. Retries apply to 408, 425, 429, and 5xx responses with exponential backoff and ±25% jitter.

Server-key safety

Client(token="gflux_secret_...") refuses to construct if it detects a Pyodide or Emscripten runtime. A server-side script that accidentally ships to the browser fails loudly rather than leaking the key into client traffic.

Errors

from getfluxly import GFluxError

try:
    client.track("invoice_paid", external_id="user_42")
except GFluxError as e:
    if e.code == "queue_overflow":
        # back-pressure; the batch is full
        ...
    elif e.retryable:
        # SDK already retried max_retries times
        ...

Full error code reference: error codes.