InvergeDocs
Python

Python SDK

inverge-nexus — sync and asyncio clients for every product, with Django, Flask, FastAPI / Starlette, Celery, WSGI and ASGI integrations and zero required dependencies.

  • PyPI inverge-nexus 1.3 · import nexus · Python 3.8+ · source
  • No required dependencies (stdlib urllib); picks up requests / httpx if installed.
  • API reference
pip install inverge-nexus
pip install "inverge-nexus[requests]"   # pooled HTTP — recommended for servers
pip install "inverge-nexus[httpx]"      # incl. the async transport
pip install "inverge-nexus[realtime]"   # listening to rooms

Three ways to use it

import nexus

nexus.init("nxs_live_xxx")          # or NEXUS_API_KEY; installs excepthooks + atexit flush
nexus.identify("user_123", email="ada@example.com")
nexus.track("order_placed", {"total": 42})
nexus.get_client().realtime.emit("orders:42", "status", {"state": "shipped"})
nexus.flush()

Best for scripts, workers and management commands.

All three share the same ambient context, so they mix freely.

Configuration

OptionDefault
api_keyNEXUS_API_KEYScoped to one environment
timeout10.0 sPer request
max_retries, retry_backoff2, 0.25 sSafe requests only
default_headers, default_properties{}
app_version, release, os_type, os_version—Telemetry context
batchTrueBuffer events / logs on a background thread
flush_interval, max_batch, max_queue5.0 s, 50, 10000
silentFalseNever raise — log and return empty
debugFalseVerbose nexus logger
services, enabledall onSwitching services off

Each has an environment variable (NEXUS_TIMEOUT, NEXUS_RELEASE, NEXUS_BATCH, NEXUS_SILENT, NEXUS_DEBUG, …). Precedence, lowest first: environment → keyword options → api_key.

Frameworks

settings.py
INSTALLED_APPS = [..., "nexus.integrations.django"]
MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    "nexus.integrations.django.NexusMiddleware",   # below auth if you want request.user
    ...,
]
NEXUS = {
    "API_KEY": os.environ["NEXUS_API_KEY"],
    "RELEASE": os.environ.get("GIT_SHA"),
    "TRACK_REQUESTS": True,     # one event per request
    "TRACK_SESSIONS": False,
    "CAPTURE_ERRORS": True,
    "LOGGING": True,            # forward Django logging
    "LOG_LEVEL": "INFO",
    "IGNORE_PATHS": ["/health", "/static/"],
    "SAMPLE_RATE": 1.0,
}

python manage.py nexus_test verifies the wiring end to end. The middleware puts request.user.pk into the ambient context.

For request-journey tracing, add Telescope for Python.

Identity & context

from nexus import context

with context.scope(distinct_id="user_123", url="/checkout", properties={"tier": "pro"}):
    nexus.track("order_placed", {"total": 42})

context.current().distinct_id
context.reset()    # sign-out: new session, same device

Backed by contextvars — correct per thread and per asyncio task. Precedence: client config → ambient context → explicit argument.

Async client

from nexus.aio import AsyncNexusClient

async with AsyncNexusClient("nxs_live_xxx") as nexus:
    await nexus.identify("user_123")
    nexus.events.track("order_placed")          # buffered: no await
    flags = await nexus.flags.evaluate("user_123")

Errors

NexusError
├── ConfigurationError
├── TransportError            DNS, connection, TLS, timeout
└── ApiError                  status, code, message, details
    ├── AuthenticationError   401
    ├── PermissionDeniedError 403
    ├── NotFoundError         404
    ├── ValidationError       400 / 422
    ├── RateLimitError        429 (retry_after)
    └── ServerError           5xx

Transports, dispatchers, testing

client = NexusClient("nxs_…", transport=RequestsTransport(my_session))
bg = client.background()                          # every call fire-and-forget
queued = client.with_dispatcher(my_dispatcher)    # Celery, RQ, SQS…
class FakeTransport(Transport):
    def __init__(self): self.requests = []
    def send(self, method, url, headers, body, timeout):
        self.requests.append((method, url, body))
        return RawResponse(200, b"{}")

client = NexusClient("nxs_test", transport=FakeTransport(), batch=False)

Use batch=False in tests and nexus.context.clear() between cases; builders have .to_dict() to assert payloads without sending.

Exceptions

Exceptions in the PHP SDK.

Python API reference

Every public class, method and option in inverge-nexus 1.3.0.

On this page

Three ways to use itConfigurationFrameworksIdentity & contextAsync clientErrorsTransports, dispatchers, testing