Stats

stats

Local usage-event collection and periodic flush to a remote endpoint.

Events are appended as JSON lines to ~/.local/share/oops/stats.jsonl. A flush is attempted at CLI startup when the last successful flush was more than 7 days ago. All I/O errors are silently swallowed so the feature never blocks normal CLI operation.

Functions:

Name	Description
append_event	Append one usage event as a JSON line to the stats file.
flush_stats	Read all pending events, POST them to the configured endpoint, then truncate.
maybe_flush	Flush pending events if the last flush was more than 7 days ago.

append_event

append_event(cmd: str, ms: float, error: str | None) -> None

Append one usage event as a JSON line to the stats file.

Silently does nothing if stats are disabled or any I/O error occurs.

Parameters:

Name	Type	Description	Default
cmd	str	Command identifier, e.g. "addons download".	required
ms	float	Elapsed time in milliseconds.	required
error	str | None	Exception class name if the command raised, otherwise None.	required

Source code in src/oops/services/stats.py

def append_event(cmd: str, ms: float, error: "str | None") -> None:
    """Append one usage event as a JSON line to the stats file.

    Silently does nothing if stats are disabled or any I/O error occurs.

    Args:
        cmd: Command identifier, e.g. ``"addons download"``.
        ms: Elapsed time in milliseconds.
        error: Exception class name if the command raised, otherwise ``None``.
    """
    try:
        from oops.core.config import config  # local import to avoid circular deps

        if not config.stats.enabled:
            return

        ts = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        event = {
            "ts": ts,
            "cmd": cmd,
            "ms": ms,
            "user": _get_user(),
            "error": error,
        }
        path = stats_file()
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(event) + "\n")
    except Exception:  # noqa: BLE001
        pass

flush_stats

flush_stats() -> None

Read all pending events, POST them to the configured endpoint, then truncate.

Args are read from config.stats. Raises on HTTP or I/O errors so that :func:maybe_flush can catch and swallow them silently.

Source code in src/oops/services/stats.py

def flush_stats() -> None:
    """Read all pending events, POST them to the configured endpoint, then truncate.

    Args are read from ``config.stats``.  Raises on HTTP or I/O errors so that
    :func:`maybe_flush` can catch and swallow them silently.
    """
    from oops.core.config import config  # local import to avoid circular deps

    path = stats_file()
    if not path.exists():
        _touch_marker()
        return

    raw = path.read_text(encoding="utf-8").strip()
    if not raw:
        _touch_marker()
        return

    endpoint = config.stats.endpoint
    if not endpoint:
        return

    events = [json.loads(line) for line in raw.splitlines() if line.strip()]

    requests.post(
        endpoint,
        json={"events": events},
        headers={"X-Oops-Token": secrets.token_hex(16)},
        timeout=5,
    )

    path.write_text("", encoding="utf-8")
    _touch_marker()

maybe_flush

maybe_flush() -> None

Flush pending events if the last flush was more than 7 days ago.

Silently does nothing when stats are disabled, the flush is not due yet, or any error occurs.

Source code in src/oops/services/stats.py

def maybe_flush() -> None:
    """Flush pending events if the last flush was more than 7 days ago.

    Silently does nothing when stats are disabled, the flush is not due yet,
    or any error occurs.
    """
    try:
        from oops.core.config import config  # local import to avoid circular deps

        if not config.stats.enabled:
            return
        if not config.stats.endpoint:
            return
        if not _flush_due():
            return
        flush_stats()
    except Exception:  # noqa: BLE001
        pass