Telemetry Implementation: Pydantic Logfire

Using Pydantic Logfire as the telemetry solution.

Implementation

# garth/telemetry.py
import os
import re
import logfire
from dataclasses import dataclass, field

SENSITIVE_PATTERNS = [
    r'"password"\s*:\s*"[^"]*"',
    r'"refresh_token"\s*:\s*"[^"]*"',
    r'"access_token"\s*:\s*"[^"]*"',
    r'"oauth_token"\s*:\s*"[^"]*"',
    r'"oauth_token_secret"\s*:\s*"[^"]*"',
    r'password=[^&\s]*',
    r'embed=[^&\s]*',
]

def _sanitize(text: str) -> str:
    """Redact sensitive data from request/response bodies."""
    for pattern in SENSITIVE_PATTERNS:
        text = re.sub(pattern, '[REDACTED]', text, flags=re.IGNORECASE)
    return text

def _response_hook(span, request, response):
    """Log sanitized response body on failures for debugging."""
    if response.status_code >= 400:
        try:
            body = response.text[:1000]
            span.set_attribute("http.response.body.sanitized", _sanitize(body))
        except Exception:
            pass

@dataclass
class Telemetry:
    service_name: str = "garth"
    enabled: bool = False  # Disabled by default
    send_to_logfire: bool = True  # Logfire Cloud when enabled
    token: str | None = None
    _configured: bool = field(default=False, repr=False)
    
    def configure(
        self,
        service_name: str | None = None,
        enabled: bool | None = None,
        send_to_logfire: bool | None = None,
        token: str | None = None,
    ):
        """
        Configure telemetry. Disabled by default.
        
        Args:
            service_name: Service name for traces (default: "garth")
            enabled: Enable/disable telemetry (default: False)
            send_to_logfire: Send to Logfire Cloud (default: True)
            token: Logfire write token (or use LOGFIRE_TOKEN env var)
        """
        if service_name is not None:
            self.service_name = service_name
        if enabled is not None:
            self.enabled = enabled
        if send_to_logfire is not None:
            self.send_to_logfire = send_to_logfire
        if token is not None:
            self.token = token
        
        # Check env var overrides
        env_enabled = os.getenv("GARTH_TELEMETRY", "").lower()
        if env_enabled == "true":
            self.enabled = True
        elif env_enabled == "false":
            self.enabled = False
            
        if os.getenv("GARTH_TELEMETRY_SERVICE_NAME"):
            self.service_name = os.getenv("GARTH_TELEMETRY_SERVICE_NAME")
        if os.getenv("LOGFIRE_SEND_TO_LOGFIRE", "").lower() == "false":
            self.send_to_logfire = False
        
        if not self.enabled or self._configured:
            return
        
        # Set token if provided
        if self.token:
            os.environ["LOGFIRE_TOKEN"] = self.token
        
        logfire.configure(
            service_name=self.service_name,
            send_to_logfire=self.send_to_logfire,
        )
        logfire.instrument_requests(response_hook=_response_hook)
        self._configured = True

# garth/http.py (additions)
from .telemetry import Telemetry

class Client:
    telemetry: Telemetry
    
    def __init__(self, session: Session | None = None, **kwargs):
        self.sess = session if session else Session()
        self.sess.headers.update(USER_AGENT)
        self.telemetry = Telemetry()
        self.configure(...)
    
    def configure(
        self,
        # ... existing params ...
        telemetry_enabled: bool | None = None,
        telemetry_service_name: str | None = None,
        telemetry_send_to_logfire: bool | None = None,
        telemetry_token: str | None = None,
    ):
        # ... existing logic ...
        self.telemetry.configure(
            enabled=telemetry_enabled,
            service_name=telemetry_service_name,
            send_to_logfire=telemetry_send_to_logfire,
            token=telemetry_token,
        )

Configuration

Default behavior: Telemetry disabled. When enabled, sends to Logfire Cloud with service_name="garth".

Enable telemetry:

# Environment variable
export GARTH_TELEMETRY=true
export LOGFIRE_TOKEN=your_write_token

# Via configure
garth.client.configure(
    telemetry_enabled=True,
    telemetry_token="your_write_token"
)

Custom service name:

export GARTH_TELEMETRY_SERVICE_NAME=my-fitness-app

garth.client.configure(telemetry_service_name="my-fitness-app")

Use alternative OTLP backend:

export GARTH_TELEMETRY=true
export LOGFIRE_SEND_TO_LOGFIRE=false
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318

garth.client.configure(
    telemetry_enabled=True,
    telemetry_send_to_logfire=False
)

What Gets Logged (When Enabled)

Request Type	Success	Failure
API calls	Method, URL, status, latency	+ sanitized error body
Auth (SSO)	Method, URL, status, latency	+ sanitized error body

Always redacted: Authorization headers, passwords, tokens in request/response bodies

Dependencies

dependencies = [
    "requests>=2.0.0,<3.0.0",
    "pydantic>=1.10.12,<3.0.0",
    "requests-oauthlib>=1.3.1,<3.0.0",
    "logfire>=2.0",
]

Summary

Feature	Default	Override
Enabled	`False`	`GARTH_TELEMETRY=true` or `.configure(telemetry_enabled=True)`
Service name	`"garth"`	`GARTH_TELEMETRY_SERVICE_NAME` or `.configure()`
Logfire Cloud	`True` (when enabled)	`LOGFIRE_SEND_TO_LOGFIRE=false` or `.configure()`
Auth token	`LOGFIRE_TOKEN` env var	`.configure(telemetry_token=...)`
Auth failures	Logged with sanitized body	-
Credentials	Always redacted	-

Telemetry #161

Description

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions