From c124f04ca03f7d256c5133a10e77cc7d2cb8325d Mon Sep 17 00:00:00 2001 From: James Murdza Date: Mon, 11 Aug 2025 09:32:00 -0400 Subject: [PATCH] Consolidate telemetry logic into one file --- libs/python/core/core/telemetry/__init__.py | 2 +- .../{posthog_client.py => posthog.py} | 35 ++++++- libs/python/core/core/telemetry/telemetry.py | 96 ------------------- 3 files changed, 35 insertions(+), 98 deletions(-) rename libs/python/core/core/telemetry/{posthog_client.py => posthog.py} (89%) delete mode 100644 libs/python/core/core/telemetry/telemetry.py diff --git a/libs/python/core/core/telemetry/__init__.py b/libs/python/core/core/telemetry/__init__.py index ee07ecb4..b5846715 100644 --- a/libs/python/core/core/telemetry/__init__.py +++ b/libs/python/core/core/telemetry/__init__.py @@ -3,7 +3,7 @@ It provides a low-overhead way to collect anonymous usage data. """ -from core.telemetry.telemetry import ( +from core.telemetry.posthog import ( record_event, is_telemetry_enabled, destroy_telemetry_client, diff --git a/libs/python/core/core/telemetry/posthog_client.py b/libs/python/core/core/telemetry/posthog.py similarity index 89% rename from libs/python/core/core/telemetry/posthog_client.py rename to libs/python/core/core/telemetry/posthog.py index 10e53277..652a899a 100644 --- a/libs/python/core/core/telemetry/posthog_client.py +++ b/libs/python/core/core/telemetry/posthog.py @@ -275,4 +275,37 @@ def get_posthog_telemetry_client() -> PostHogTelemetryClient: if _client is None: _client = PostHogTelemetryClient() - return _client \ No newline at end of file + return _client + +# --------------------------------------------------------------------------- +# Lightweight wrapper functions (migrated from telemetry.py) +# --------------------------------------------------------------------------- + +def _telemetry_disabled() -> bool: + return ( + os.environ.get("CUA_TELEMETRY", "").lower() == "off" + or os.environ.get("CUA_TELEMETRY_ENABLED", "true").lower() not in {"1", "true", "yes", "on"} + ) + + +def destroy_telemetry_client() -> None: + """Destroy the global telemetry client instance.""" + global _client + _client = None + + +def is_telemetry_enabled() -> bool: + """Return True if telemetry is currently active.""" + if _telemetry_disabled(): + return False + client = get_posthog_telemetry_client() + return client.config.enabled if client else False + + +def record_event(event_name: str, properties: Optional[Dict[str, Any]] | None = None) -> None: + """Record an arbitrary PostHog event.""" + if _telemetry_disabled(): + return + client = get_posthog_telemetry_client() + if client and client.config.enabled: + client.record_event(event_name, properties or {}) \ No newline at end of file diff --git a/libs/python/core/core/telemetry/telemetry.py b/libs/python/core/core/telemetry/telemetry.py deleted file mode 100644 index 300e036d..00000000 --- a/libs/python/core/core/telemetry/telemetry.py +++ /dev/null @@ -1,96 +0,0 @@ -"""Lightweight telemetry wrapper for PostHog. All helpers are thin wrappers around -a single, lazily-initialised PostHog client. - -Usage: - from core.telemetry import record_event, increment - - record_event("my_event", {"foo": "bar"}) - -Configuration: - • Disable telemetry globally by setting the environment variable - CUA_TELEMETRY=off OR CUA_TELEMETRY_ENABLED=false/0. - • Control log verbosity with CUA_TELEMETRY_LOG_LEVEL (DEBUG|INFO|WARNING|ERROR). - -If the `posthog` package (and the accompanying `PostHogTelemetryClient` helper) -are not available, every public function becomes a no-op. -""" - -from __future__ import annotations - -import logging -import os -from typing import Any, Dict, Optional - -# --------------------------------------------------------------------------- -# Logging configuration -# --------------------------------------------------------------------------- - -_DEFAULT_LOG_LEVEL = os.environ.get("CUA_TELEMETRY_LOG_LEVEL", "WARNING").upper() -logging.basicConfig(level=getattr(logging, _DEFAULT_LOG_LEVEL, logging.WARNING)) -_LOGGER = logging.getLogger("core.telemetry") - -# --------------------------------------------------------------------------- -# Attempt to import the PostHog client helper. If unavailable, telemetry is -# silently disabled. -# --------------------------------------------------------------------------- - -try: - from core.telemetry.posthog_client import get_posthog_telemetry_client # type: ignore -except ImportError: # pragma: no cover - get_posthog_telemetry_client = None # type: ignore[misc, assignment] - -# --------------------------------------------------------------------------- -# Internal helpers & primitives -# --------------------------------------------------------------------------- - -def _telemetry_disabled() -> bool: - """Return True if the user has disabled telemetry via environment vars.""" - return ( - os.environ.get("CUA_TELEMETRY", "").lower() == "off" # legacy flag - or os.environ.get("CUA_TELEMETRY_ENABLED", "true").lower() # new flag - not in {"1", "true", "yes", "on"} - ) - - -_CLIENT = None # Lazily instantiated PostHog client instance -_ENABLED = False # Guard to avoid making calls when telemetry disabled - -def _ensure_client() -> None: - """Initialise the PostHog client once and cache it globally.""" - global _CLIENT, _ENABLED - - # Bail early if telemetry is disabled or already initialised - if _CLIENT is not None or _telemetry_disabled(): - return - - if get_posthog_telemetry_client is None: - _LOGGER.debug("posthog package not found – telemetry disabled") - return - - try: - _CLIENT = get_posthog_telemetry_client() - _ENABLED = True - except Exception as exc: # pragma: no cover - _LOGGER.debug("Failed to initialise PostHog client: %s", exc) - _CLIENT = None - _ENABLED = False - -# --------------------------------------------------------------------------- -# Public API -# --------------------------------------------------------------------------- - -def destroy_telemetry_client() -> None: - """Destroy the global telemetry client.""" - global _CLIENT - _CLIENT = None - -def is_telemetry_enabled() -> bool: - """Return True if telemetry is currently active.""" - _ensure_client() - return _ENABLED and not _telemetry_disabled() - -def record_event(event_name: str, properties: Optional[Dict[str, Any]] | None = None) -> None: - """Send an arbitrary PostHog event.""" - _ensure_client() - if _CLIENT and _ENABLED: - _CLIENT.record_event(event_name, properties or {}) \ No newline at end of file