Structlog processor

logtide_sdk/__init__.py

@@ -1,18 +1,10 @@
 """LogTide SDK - Official Python SDK for LogTide."""
 
-from .client import LogTideClient, serialize_exception
-
-_has_async = False
-try:
-    from .async_client import AsyncLogTideClient
-
-    _has_async = True
-except ImportError:
-    pass  # type: ignore[assignment]
-from .enums import CircuitState, LogLevel
-from .exceptions import BufferFullError, CircuitBreakerOpenError, LogTideError
-from .handler import LogTideHandler
-from .models import (
+from logtide_sdk.client import LogTideClient, serialize_exception
+from logtide_sdk.enums import CircuitState, LogLevel
+from logtide_sdk.exceptions import BufferFullError, CircuitBreakerOpenError, LogTideError
+from logtide_sdk.handler import LogTideHandler
+from logtide_sdk.models import (
     AggregatedStatsOptions,
     AggregatedStatsResponse,
     ClientMetrics,
@@ -23,16 +15,20 @@
     QueryOptions,
 )
 
+_has_async = False
+try:
+    from logtide_sdk.async_client import AsyncLogTideClient
+
+    _has_async = True  # type: ignore[assignment]
+except ImportError:
+    pass
+
 __version__ = "0.8.5"
 
 __all__ = [
-    # Clients
     "LogTideClient",
-    # Logging integration
     "LogTideHandler",
-    # Error serialization utility
     "serialize_exception",
-    # Models
     "LogEntry",
     "ClientOptions",
     "QueryOptions",
@@ -41,10 +37,8 @@
     "LogsResponse",
     "AggregatedStatsResponse",
     "PayloadLimitsOptions",
-    # Enums
     "LogLevel",
     "CircuitState",
-    # Exceptions
     "LogTideError",
     "CircuitBreakerOpenError",
     "BufferFullError",

logtide_sdk/async_client.py

@@ -16,11 +16,11 @@
         "Install it with: pip install logtide-sdk[async]"
     )
 
-from .circuit_breaker import CircuitBreaker
-from .client import _process_value, serialize_exception
-from .enums import CircuitState, LogLevel
-from .exceptions import CircuitBreakerOpenError
-from .models import (
+from logtide_sdk.circuit_breaker import CircuitBreaker
+from logtide_sdk.client import _process_value, serialize_exception
+from logtide_sdk.enums import CircuitState, LogLevel
+from logtide_sdk.exceptions import CircuitBreakerOpenError
+from logtide_sdk.models import (
     AggregatedStatsOptions,
     AggregatedStatsResponse,
     ClientMetrics,
@@ -182,7 +182,7 @@ async def log(self, entry: LogEntry) -> None:
             await self.flush()
 
     async def debug(
-            self, service: str, message: str, metadata: Optional[Dict[str, Any]] = None
+        self, service: str, message: str, metadata: Optional[Dict[str, Any]] = None
     ) -> None:
         """Log a DEBUG-level message."""
         await self.log(
@@ -195,7 +195,7 @@ async def debug(
         )
 
     async def info(
-            self, service: str, message: str, metadata: Optional[Dict[str, Any]] = None
+        self, service: str, message: str, metadata: Optional[Dict[str, Any]] = None
     ) -> None:
         """Log an INFO-level message."""
         await self.log(
@@ -208,7 +208,7 @@ async def info(
         )
 
     async def warn(
-            self, service: str, message: str, metadata: Optional[Dict[str, Any]] = None
+        self, service: str, message: str, metadata: Optional[Dict[str, Any]] = None
     ) -> None:
         """Log a WARN-level message."""
         await self.log(
@@ -221,10 +221,10 @@ async def warn(
         )
 
     async def error(
-            self,
-            service: str,
-            message: str,
-            metadata_or_error: Union[Dict[str, Any], Exception, None] = None,
+        self,
+        service: str,
+        message: str,
+        metadata_or_error: Union[Dict[str, Any], Exception, None] = None,
     ) -> None:
         """Log an ERROR-level message. Accepts an Exception for automatic serialization."""
         metadata = self._process_metadata_or_error(metadata_or_error)
@@ -238,10 +238,10 @@ async def error(
         )
 
     async def critical(
-            self,
-            service: str,
-            message: str,
-            metadata_or_error: Union[Dict[str, Any], Exception, None] = None,
+        self,
+        service: str,
+        message: str,
+        metadata_or_error: Union[Dict[str, Any], Exception, None] = None,
     ) -> None:
         """Log a CRITICAL-level message. Accepts an Exception for automatic serialization."""
         metadata = self._process_metadata_or_error(metadata_or_error)
@@ -295,9 +295,9 @@ async def query(self, options: QueryOptions) -> LogsResponse:
             params["to"] = options.to_time.isoformat()
 
         async with self._get_session().get(
-                f"{self.options.api_url}/api/v1/logs",
-                headers=self._get_headers(),
-                params=params,
+            f"{self.options.api_url}/api/v1/logs",
+            headers=self._get_headers(),
+            params=params,
         ) as response:
             response.raise_for_status()
             data = await response.json()
@@ -306,14 +306,14 @@ async def query(self, options: QueryOptions) -> LogsResponse:
     async def get_by_trace_id(self, trace_id: str) -> List[Dict[str, Any]]:
         """Return all log entries for a given trace ID."""
         async with self._get_session().get(
-                f"{self.options.api_url}/api/v1/logs/trace/{trace_id}",
-                headers=self._get_headers(),
+            f"{self.options.api_url}/api/v1/logs/trace/{trace_id}",
+            headers=self._get_headers(),
         ) as response:
             response.raise_for_status()
             return await response.json()
 
     async def get_aggregated_stats(
-            self, options: AggregatedStatsOptions
+        self, options: AggregatedStatsOptions
     ) -> AggregatedStatsResponse:
         """Return aggregated statistics over a time range."""
         params: Dict[str, Any] = {
@@ -325,9 +325,9 @@ async def get_aggregated_stats(
             params["service"] = options.service
 
         async with self._get_session().get(
-                f"{self.options.api_url}/api/v1/logs/aggregated",
-                headers=self._get_headers(),
-                params=params,
+            f"{self.options.api_url}/api/v1/logs/aggregated",
+            headers=self._get_headers(),
+            params=params,
         ) as response:
             response.raise_for_status()
             data = await response.json()
@@ -338,10 +338,10 @@ async def get_aggregated_stats(
             )
 
     async def stream(
-            self,
-            on_log: Callable[[Dict[str, Any]], None],
-            on_error: Optional[Callable[[Exception], None]] = None,
-            filters: Optional[Dict[str, str]] = None,
+        self,
+        on_log: Callable[[Dict[str, Any]], None],
+        on_error: Optional[Callable[[Exception], None]] = None,
+        filters: Optional[Dict[str, str]] = None,
     ) -> None:
         """
         Stream logs in real-time via SSE. This coroutine runs until cancelled.
@@ -357,9 +357,9 @@ async def stream(
         params["token"] = self.options.api_key
 
         async with self._get_session().get(
-                f"{self.options.api_url}/api/v1/logs/stream",
-                headers=self._get_headers(),
-                params=params,
+            f"{self.options.api_url}/api/v1/logs/stream",
+            headers=self._get_headers(),
+            params=params,
         ) as response:
             response.raise_for_status()
             async for line_bytes in response.content:
@@ -458,58 +458,52 @@ async def _send_logs_with_retry(self, logs: List[LogEntry]) -> None:
 
                 if attempt > self.options.max_retries:
                     if self.options.debug:
-                        print(
-                            f"[LogTide] Failed to send logs after {attempt} attempts: {e}"
-                        )
+                        print(f"[LogTide] Failed to send logs after {attempt} attempts: {e}")
                     with self._metrics_lock:
                         self._metrics.logs_dropped += len(logs)
                     break
 
                 if self.options.debug:
-                    print(
-                        f"[LogTide] Retry {attempt}/{self.options.max_retries} in {delay}s"
-                    )
+                    print(f"[LogTide] Retry {attempt}/{self.options.max_retries} in {delay}s")
 
                 await asyncio.sleep(delay)
                 delay *= 2
 
-        if (self._circuit_breaker.state == CircuitState.OPEN
-                and state_before != CircuitState.OPEN):
+        if self._circuit_breaker.state == CircuitState.OPEN and state_before != CircuitState.OPEN:
             with self._metrics_lock:
                 self._metrics.circuit_breaker_trips += 1
 
     async def _send_logs(self, logs: List[LogEntry]) -> None:
         """POST a serialized batch to /api/v1/ingest."""
         payload = {"logs": [log.to_dict() for log in logs]}
         async with self._get_session().post(
-                f"{self.options.api_url}/api/v1/ingest",
-                headers=self._get_headers(),
-                json=payload,
+            f"{self.options.api_url}/api/v1/ingest",
+            headers=self._get_headers(),
+            json=payload,
         ) as response:
             response.raise_for_status()
 
     def _process_metadata_or_error(
-            self, metadata_or_error: Union[Dict[str, Any], Exception, None]
+        self, metadata_or_error: Union[Dict[str, Any], Exception, None]
     ) -> Dict[str, Any]:
         if metadata_or_error is None:
             return {}
         if isinstance(metadata_or_error, dict):
             return metadata_or_error
         return {"exception": serialize_exception(metadata_or_error)}
 
+    # NOTE: this is twice. (both in async and regular clients)
     def _apply_payload_limits(self, entry: LogEntry) -> None:
         """Enforce payload limits on entry.metadata in-place."""
         if not entry.metadata:
             return
         lim = self._payload_limits
         entry.metadata = _process_value(entry.metadata, "root", lim)
 
-        raw = json.dumps(entry.to_dict())
+        raw = json.dumps(entry.to_dict(), default=lambda o: repr(o))
         if len(raw.encode()) > lim.max_log_size:
             if self.options.debug:
-                print(
-                    f"[LogTide] Log entry too large ({len(raw)} bytes), truncating metadata"
-                )
+                print(f"[LogTide] Log entry too large ({len(raw)} bytes), truncating metadata")
             entry.metadata = {"_truncated": True, "_original_size": len(raw.encode())}
 
     def _update_latency(self, latency: float) -> None:
@@ -518,6 +512,4 @@ def _update_latency(self, latency: float) -> None:
             if len(self._latency_window) > 100:
                 self._latency_window.pop(0)
             if self._latency_window:
-                self._metrics.avg_latency_ms = sum(self._latency_window) / len(
-                    self._latency_window
-                )
+                self._metrics.avg_latency_ms = sum(self._latency_window) / len(self._latency_window)

logtide_sdk/circuit_breaker.py

@@ -4,8 +4,8 @@
 from threading import Lock
 from typing import Callable, TypeVar
 
-from .enums import CircuitState
-from .exceptions import CircuitBreakerOpenError
+from logtide_sdk.enums import CircuitState
+from logtide_sdk.exceptions import CircuitBreakerOpenError
 
 T = TypeVar("T")