(backend) introduce a configurable analytics system

Add an analytics abstraction that allows configuring which analytics
solution the app uses. PostHog is implemented as one backend, but by
default no analytics backend is activated.

The goal is to track events in a sufficiently organized way and to
let any developer implement their own backend as long as it follows
the same protocol.
This commit is contained in:
lebaudantoine
2026-07-01 18:33:12 +02:00
committed by aleb_the_flash
parent be0d0927d4
commit 9ba97fd14f
7 changed files with 464 additions and 0 deletions
+59
View File
@@ -0,0 +1,59 @@
"""
Pluggable analytics.
Usage anywhere in the codebase:
from core import analytics
analytics.capture(request.user, "room_created", {"room_id": str(room.pk)})
The concrete backend is resolved lazily from Django settings, so swapping
PostHog for anything else is a configuration change, not a code change.
"""
from functools import lru_cache
from typing import Any
from django.conf import settings
from django.utils.module_loading import import_string
from .base import AnalyticsBackend, NoOpAnalytics
from .events import AnalyticsEvent
__all__ = [
"get_analytics",
"identify",
"capture",
"AnalyticsBackend",
"AnalyticsEvent",
]
@lru_cache(maxsize=1)
def get_analytics() -> AnalyticsBackend:
"""Instantiate the configured backend once per process."""
dotted_path = getattr(settings, "ANALYTICS_BACKEND", None)
options = getattr(settings, "ANALYTICS_BACKEND_SETTINGS", {}) or {}
if not dotted_path:
return NoOpAnalytics()
backend_class = import_string(dotted_path)
return backend_class(**options)
# Convenience module-level shortcuts
analytics_instance = get_analytics()
def identify(user, properties: dict[str, Any] | None = None) -> None:
"""Associate traits with an identified user."""
analytics_instance.identify(user, properties)
def capture(
user, event: AnalyticsEvent, properties: dict[str, Any] | None = None
) -> None:
"""Record an event performed by an identified user."""
analytics_instance.capture(user, event, properties)
+50
View File
@@ -0,0 +1,50 @@
"""Analytics backend protocol and default no-op implementation."""
from typing import Any, Protocol
from ..models import User
from .events import AnalyticsEvent
class AnalyticsBackend(Protocol):
"""
Interface every analytics backend must implement.
Backends are instantiated once (singleton) with the kwargs declared in
settings.ANALYTICS_BACKEND_SETTINGS, e.g.:
ANALYTICS_BACKEND = "core.analytics.posthog.PostHogAnalytics"
ANALYTICS_BACKEND_SETTINGS = {"api_key": "...", "host": "..."}
"""
def __init__(self, **kwargs: Any) -> None: ...
def identify(self, user: User, properties: dict[str, Any] | None = None) -> None:
"""Associate traits (email, name, ...) with an identified user."""
def capture(
self,
user: User,
event: AnalyticsEvent,
properties: dict[str, Any] | None = None,
) -> None:
"""Record an event performed by an identified user."""
def shutdown(self) -> None:
"""Flush pending events. Called on process exit."""
class NoOpAnalytics:
"""Default backend: silently discards everything."""
def __init__(self, **kwargs: Any) -> None:
"""No-op: accepts and ignores any backend settings kwargs."""
def identify(self, user: User, properties=None) -> None:
"""No-op: discards identify calls."""
def capture(self, user, event, properties=None) -> None:
"""No-op: discards captured events."""
def shutdown(self) -> None:
"""No-op: nothing to flush."""
+10
View File
@@ -0,0 +1,10 @@
"""Catalog of all analytics events emitted by the backend."""
from enum import StrEnum
class AnalyticsEvent(StrEnum):
"""All trackable events. Values are the wire names sent to the provider."""
# Rooms
ROOM_CREATED = "room_created"
+74
View File
@@ -0,0 +1,74 @@
"""PostHog implementation of the analytics backend protocol."""
import logging
from typing import Any
from posthog import Posthog
from ..models import User
from .events import AnalyticsEvent
logger = logging.getLogger(__name__)
class PostHogAnalytics:
"""Send events to PostHog, keyed on the user's primary key (UUID)."""
def __init__(
self,
*,
api_key: str,
host: str = "https://eu.i.posthog.com",
**kwargs: Any,
) -> None:
# The SDK batches and sends in a background thread by default,
# so calls below never block the request/response cycle.
self._client = Posthog(
project_api_key=api_key,
host=host,
**kwargs,
)
@staticmethod
def _distinct_id(user: User) -> str | None:
"""Return the PostHog distinct_id for a user, or None if anonymous."""
if user is None or not getattr(user, "is_authenticated", False):
return None
return str(user.pk)
def identify(self, user: User, properties: dict[str, Any] | None = None) -> None:
"""Associate traits (email, name, ...) with an identified user."""
distinct_id = self._distinct_id(user)
if distinct_id is None:
return
try:
self._client.set(
distinct_id=distinct_id,
properties=properties or {},
)
except Exception: # pylint: disable=broad-exception-caught
logger.exception("PostHog identify failed")
def capture(
self,
user: User,
event: AnalyticsEvent,
properties: dict[str, Any] | None = None,
) -> None:
"""Record an event performed by an identified user."""
distinct_id = self._distinct_id(user)
if distinct_id is None:
return
try:
self._client.capture(
distinct_id=distinct_id,
event=str(event),
properties=properties or {},
)
except Exception: # pylint: disable=broad-exception-caught
logger.exception("PostHog capture failed for event %s", event)
def shutdown(self) -> None:
"""Flush pending events. Called on process exit."""
self._client.shutdown()