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
46
47
48
49
def __init__(self, config: AnalyticsConfig, anon_id: str | None = None) -> None:
    self.config = config
    self._enabled = is_analytics_enabled(config)
    self.anon_id = (
        anon_id or get_or_create_anon_id(config.anon_id_path)
        if self._enabled
        else ""
    )
    self._lock = threading.Lock()
    self._posthog: Any = None
    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 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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 
108
109
110
111
112
113
114
115
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
disable 
disable() -> None

Stop accepting events without waiting for network delivery.

Source code in src/openjarvis/analytics/client.py 
117
118
119
120
def disable(self) -> None:
    """Stop accepting events without waiting for network delivery."""
    with self._lock:
        self._enabled = False
shutdown 
shutdown(*, wait: bool = True) -> None

Close the SDK, optionally moving network cleanup off the caller.

Source code in src/openjarvis/analytics/client.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def shutdown(self, *, wait: bool = True) -> None:
    """Close the SDK, optionally moving network cleanup off the caller."""
    with self._lock:
        sdk = self._posthog
        self._posthog = None
        self._enabled = False
    if sdk is None:
        return

    def close_sdk() -> None:
        try:
            sdk.flush()
            sdk.shutdown()
        except Exception:
            pass

    if wait:
        close_sdk()
    else:
        threading.Thread(
            target=close_sdk, name="analytics-shutdown", daemon=True
        ).start()

Functions