Skip to content

analytics_routes

analytics_routes

Small router exposing the anonymous analytics identity to the frontend.

The Tauri desktop app and web frontend need the same anon_id that the backend and install.sh use so that all events tie to one person. This endpoint returns that ID (generating it on first call) plus the public PostHog project key and host so the frontend can initialise posthog-js against the same project.

Uses the server's normal API-key authentication when configured. The returned project key is a public PostHog token, not a secret.

Classes

AnalyticsPreference

Bases: BaseModel

Only the analytics choice belongs here; contest details go elsewhere.

Functions

get_identity

get_identity(request: Request) -> AnalyticsIdentity

Return the analytics identity for the current install.

If analytics is disabled in config, still return a structurally valid response with enabled=False so the frontend can decide what to do; never throw.

Source code in src/openjarvis/server/analytics_routes.py
@router.get("/identity", response_model=AnalyticsIdentity)
def get_identity(request: Request) -> AnalyticsIdentity:
    """Return the analytics identity for the current install.

    If analytics is disabled in config, still return a
    structurally valid response with ``enabled=False`` so the frontend
    can decide what to do; never throw.
    """
    cfg = getattr(request.app.state, "config", None) or load_config()
    enabled = is_analytics_enabled(cfg.analytics)
    anon_id = get_or_create_anon_id(cfg.analytics.anon_id_path) if enabled else ""
    return AnalyticsIdentity(
        enabled=enabled,
        anon_id=anon_id,
        host=cfg.analytics.host,
        key=cfg.analytics.key if enabled else "",
        choice_recorded=cfg.analytics.choice_recorded,
    )

stop_analytics

stop_analytics(app: FastAPI, *, wait: bool = True) -> None

Detach the event bridge and close the current analytics client.

Source code in src/openjarvis/server/analytics_routes.py
def stop_analytics(app: FastAPI, *, wait: bool = True) -> None:
    """Detach the event bridge and close the current analytics client."""
    bridge = getattr(app.state, "analytics_bridge", None)
    client = getattr(app.state, "analytics_client", None)
    app.state.analytics_bridge = None
    app.state.analytics_client = None
    if client is not None and not wait:
        client.disable()
    if bridge is not None:
        try:
            bridge.stop()
        except Exception:
            logger.debug("Analytics bridge shutdown failed", exc_info=True)
    if client is not None:
        client.shutdown(wait=wait)

configure_analytics

configure_analytics(app: FastAPI, cfg: JarvisConfig) -> None

Apply a sharing choice immediately to this running server.

Source code in src/openjarvis/server/analytics_routes.py
def configure_analytics(app: FastAPI, cfg: JarvisConfig) -> None:
    """Apply a sharing choice immediately to this running server."""
    stop_analytics(app, wait=False)
    app.state.config = cfg
    if not is_analytics_enabled(cfg.analytics):
        return
    client = AnalyticsClient(cfg.analytics)
    app.state.analytics_client = client
    bus = getattr(app.state, "bus", None)
    if bus is not None and client.enabled:
        bridge = EventBridge(bus, client)
        app.state.analytics_bridge = bridge
        bridge.start()

set_preference

set_preference(choice: AnalyticsPreference, request: Request) -> AnalyticsIdentity

Persist an explicit UI choice and update analytics without restarting.

Source code in src/openjarvis/server/analytics_routes.py
@router.post("/preference", response_model=AnalyticsIdentity)
def set_preference(choice: AnalyticsPreference, request: Request) -> AnalyticsIdentity:
    """Persist an explicit UI choice and update analytics without restarting."""
    with _preference_lock:
        cfg = getattr(request.app.state, "config", None) or load_config()
        path = cfg._config_path
        temporary: Path | None = None
        try:
            path.parent.mkdir(parents=True, exist_ok=True)
            doc = (
                tomlkit.parse(path.read_text(encoding="utf-8"))
                if path.exists()
                else tomlkit.document()
            )
            if "analytics" not in doc:
                doc.add("analytics", tomlkit.table())
            doc["analytics"]["enabled"] = choice.enabled
            doc["analytics"]["choice_recorded"] = True
            with tempfile.NamedTemporaryFile(
                mode="w", encoding="utf-8", dir=path.parent, delete=False
            ) as handle:
                temporary = Path(handle.name)
                handle.write(tomlkit.dumps(doc))
            temporary.replace(path)
        except (OSError, ValueError, TypeError) as exc:
            logger.warning("Could not save analytics preference: %s", exc)
            raise HTTPException(
                status_code=500, detail="Could not save your analytics choice."
            ) from exc
        finally:
            if temporary is not None:
                temporary.unlink(missing_ok=True)
        cfg.analytics.enabled = choice.enabled
        cfg.analytics.choice_recorded = True
        load_config.cache_clear()
        client = getattr(request.app.state, "analytics_client", None)
        if (choice.enabled and (client is None or not client.enabled)) or (
            not choice.enabled and client is not None
        ):
            configure_analytics(request.app, cfg)
        else:
            request.app.state.config = cfg
        return get_identity(request)