Skip to content

client

client

PostHog client wrapper.

A thin adapter over the official posthog SDK that: - Initialises lazily and only if analytics is enabled. - Pipes every capture through :mod:redaction and :mod:events for PII stripping and structural validation. - Fails silently — analytics must never break the host application.

The underlying SDK handles batching, async send on a background thread, retries, and silent failure on network errors. We layer "never crash the app" on top by wrapping every call in try/except.

Classes

AnalyticsClient 

AnalyticsClient(config: AnalyticsConfig, anon_id: str | None = None)

Send anonymized usage events to PostHog.

Construct once at server / CLI startup, share for the process lifetime, call :meth:shutdown on exit to flush pending events.

Source code in src/openjarvis/analytics/client.py 
38
39
40
41
42
43
44
45
def __init__(self, config: AnalyticsConfig, anon_id: str | None = None) -> None:
    self.config = config
    self.anon_id = anon_id or get_or_create_anon_id(config.anon_id_path)
    self._lock = threading.Lock()
    self._posthog: Any = None
    self._enabled = is_analytics_enabled(config)
    if self._enabled:
        self._init_sdk()
Functions
capture 
capture(event_name: str, properties: dict[str, Any] | None = None) -> None

Send one event. Unknown events are silently dropped.

Runs through redaction → event-spec validation → SDK capture. Failures at any stage are swallowed; analytics is best-effort.

Source code in src/openjarvis/analytics/client.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def capture(
    self,
    event_name: str,
    properties: dict[str, Any] | None = None,
) -> None:
    """Send one event. Unknown events are silently dropped.

    Runs through redaction → event-spec validation → SDK capture.
    Failures at any stage are swallowed; analytics is best-effort.
    """
    if not self.enabled:
        return
    try:
        raw = properties or {}
        cleaned = redact(raw)
        validated = validate_event(event_name, cleaned)
        if validated is None:
            logger.debug("Dropped unknown analytics event: %s", event_name)
            return
        self._posthog.capture(
            distinct_id=self.anon_id,
            event=event_name,
            properties=validated,
        )
    except Exception as exc:
        logger.debug("Analytics capture failed for %s: %s", event_name, exc)
flush 
flush() -> None

Force-flush queued events. Safe to call repeatedly.

Source code in src/openjarvis/analytics/client.py 
104
105
106
107
108
109
110
111
def flush(self) -> None:
    """Force-flush queued events. Safe to call repeatedly."""
    if self._posthog is None:
        return
    try:
        self._posthog.flush()
    except Exception:
        pass
shutdown 
shutdown() -> None

Flush and close the SDK. Call once on process exit.

Source code in src/openjarvis/analytics/client.py 
113
114
115
116
117
118
119
120
121
122
123
124
def shutdown(self) -> None:
    """Flush and close the SDK. Call once on process exit."""
    with self._lock:
        if self._posthog is None:
            return
        try:
            self._posthog.flush()
            self._posthog.shutdown()
        except Exception:
            pass
        self._posthog = None
        self._enabled = False

Functions