chore: unify logging usage

Author: akx

langfuse/_client/observe.py

@@ -1,7 +1,6 @@
 import asyncio
 import contextvars
 import inspect
-import logging
 import os
 from functools import wraps
 from typing import (
@@ -42,6 +41,7 @@
     LangfuseSpan,
     LangfuseTool,
 )
+from langfuse.logger import langfuse_logger as logger
 from langfuse.types import TraceContext
 
 F = TypeVar("F", bound=Callable[..., Any])
@@ -69,8 +69,6 @@ class LangfuseDecorator:
     - Thread-safe client resolution when multiple Langfuse projects are used
     """
 
-    _log = logging.getLogger("langfuse")
-
     @overload
     def observe(self, func: F) -> F: ...
 
@@ -166,7 +164,7 @@ def sub_process():
         """
         valid_types = set(get_observation_types_list(ObservationTypeLiteralNoEvent))
         if as_type is not None and as_type not in valid_types:
-            self._log.warning(
+            logger.warning(
                 f"Invalid as_type '{as_type}'. Valid types are: {', '.join(sorted(valid_types))}. Defaulting to 'span'."
             )
             as_type = "span"

langfuse/_task_manager/media_manager.py

@@ -1,4 +1,3 @@
-import logging
 import os
 import time
 from queue import Empty, Full, Queue
@@ -12,6 +11,7 @@
 from langfuse._utils import _get_timestamp
 from langfuse.api import LangfuseAPI, MediaContentType
 from langfuse.api.core import ApiError
+from langfuse.logger import langfuse_logger as logger
 from langfuse.media import LangfuseMedia
 
 from .media_upload_queue import UploadMediaJob
@@ -21,8 +21,6 @@
 
 
 class MediaManager:
-    _log = logging.getLogger("langfuse")
-
     def __init__(
         self,
         *,
@@ -42,7 +40,7 @@ def __init__(
     def process_next_media_upload(self) -> None:
         try:
             upload_job = self._queue.get(block=True, timeout=1)
-            self._log.debug(
+            logger.debug(
                 f"Media: Processing upload for media_id={upload_job['media_id']} in trace_id={upload_job['trace_id']}"
             )
             self._process_upload_media_job(data=upload_job)
@@ -51,7 +49,7 @@ def process_next_media_upload(self) -> None:
         except Empty:
             pass
         except Exception as e:
-            self._log.error(
+            logger.error(
                 f"Media upload error: Failed to upload media due to unexpected error. Queue item marked as done. Error: {e}"
             )
             self._queue.task_done()
@@ -184,7 +182,7 @@ def _process_media(
             return
 
         if media._media_id is None:
-            self._log.error("Media ID is None. Skipping upload.")
+            logger.error("Media ID is None. Skipping upload.")
             return
 
         try:
@@ -203,17 +201,17 @@ def _process_media(
                 item=upload_media_job,
                 block=False,
             )
-            self._log.debug(
+            logger.debug(
                 f"Queue: Enqueued media ID {media._media_id} for upload processing | trace_id={trace_id} | field={field}"
             )
 
         except Full:
-            self._log.warning(
+            logger.warning(
                 f"Queue capacity: Media queue is full. Failed to process media_id={media._media_id} for trace_id={trace_id}. Consider increasing queue capacity."
             )
 
         except Exception as e:
-            self._log.error(
+            logger.error(
                 f"Media processing error: Failed to process media_id={media._media_id} for trace_id={trace_id}. Error: {str(e)}"
             )
 
@@ -235,14 +233,14 @@ def _process_upload_media_job(
         upload_url = upload_url_response.upload_url
 
         if not upload_url:
-            self._log.debug(
+            logger.debug(
                 f"Media status: Media with ID {data['media_id']} already uploaded. Skipping duplicate upload."
             )
 
             return
 
         if upload_url_response.media_id != data["media_id"]:
-            self._log.error(
+            logger.error(
                 f"Media integrity error: Media ID mismatch between SDK ({data['media_id']}) and Server ({upload_url_response.media_id}). Upload cancelled. Please check media ID generation logic."
             )
 
@@ -298,7 +296,7 @@ def _upload_with_status_check() -> httpx.Response:
             upload_time_ms=upload_time_ms,
         )
 
-        self._log.debug(
+        logger.debug(
             f"Media upload: Successfully uploaded media_id={data['media_id']} for trace_id={data['trace_id']} | status_code={upload_response.status_code} | duration={upload_time_ms}ms | size={data['content_length']} bytes"
         )
 

langfuse/_task_manager/media_upload_consumer.py

@@ -1,11 +1,11 @@
-import logging
 import threading
 
+from langfuse.logger import langfuse_logger as logger
+
 from .media_manager import MediaManager
 
 
 class MediaUploadConsumer(threading.Thread):
-    _log = logging.getLogger("langfuse")
     _identifier: int
     _max_retries: int
     _media_manager: MediaManager
@@ -30,15 +30,15 @@ def __init__(
 
     def run(self) -> None:
         """Run the media upload consumer."""
-        self._log.debug(
+        logger.debug(
             f"Thread: Media upload consumer thread #{self._identifier} started and actively processing queue items"
         )
         while self.running:
             self._media_manager.process_next_media_upload()
 
     def pause(self) -> None:
         """Pause the media upload consumer."""
-        self._log.debug(
+        logger.debug(
             f"Thread: Pausing media upload consumer thread #{self._identifier}"
         )
         self.running = False

langfuse/_task_manager/score_ingestion_consumer.py

@@ -1,5 +1,4 @@
 import json
-import logging
 import os
 import threading
 import time
@@ -12,6 +11,7 @@
 from langfuse._utils.parse_error import handle_exception
 from langfuse._utils.request import APIError, LangfuseClient
 from langfuse._utils.serializer import EventSerializer
+from langfuse.logger import langfuse_logger as logger
 
 from ..version import __version__ as langfuse_version
 
@@ -27,8 +27,6 @@ class ScoreIngestionMetadata(BaseModel):
 
 
 class ScoreIngestionConsumer(threading.Thread):
-    _log = logging.getLogger("langfuse")
-
     def __init__(
         self,
         *,
@@ -83,7 +81,7 @@ def _next(self) -> list:
                 try:
                     json.dumps(event, cls=EventSerializer)
                 except Exception as e:
-                    self._log.error(
+                    logger.error(
                         f"Data error: Failed to serialize score object for ingestion. Score will be dropped. Error: {e}"
                     )
                     self._ingestion_queue.task_done()
@@ -94,7 +92,7 @@ def _next(self) -> list:
 
                 total_size += item_size
                 if total_size >= MAX_BATCH_SIZE_BYTES:
-                    self._log.debug(
+                    logger.debug(
                         f"Batch management: Reached maximum batch size limit ({total_size} bytes). Processing {len(events)} events now."
                     )
                     break
@@ -103,7 +101,7 @@ def _next(self) -> list:
                 break
 
             except Exception as e:
-                self._log.warning(
+                logger.warning(
                     f"Data processing error: Failed to process score event in consumer thread #{self._identifier}. Event will be dropped. Error: {str(e)}",
                     exc_info=True,
                 )
@@ -117,7 +115,7 @@ def _get_item_size(self, item: Any) -> int:
 
     def run(self) -> None:
         """Run the consumer."""
-        self._log.debug(
+        logger.debug(
             f"Startup: Score ingestion consumer thread #{self._identifier} started with batch size {self._flush_at} and interval {self._flush_interval}s"
         )
         while self.running:
@@ -143,7 +141,7 @@ def pause(self) -> None:
         self.running = False
 
     def _upload_batch(self, batch: List[Any]) -> None:
-        self._log.debug(
+        logger.debug(
             f"API: Uploading batch of {len(batch)} score events to Langfuse API"
         )
 
@@ -171,6 +169,6 @@ def execute_task_with_backoff(batch: List[Any]) -> None:
                 raise e
 
         execute_task_with_backoff(batch)
-        self._log.debug(
+        logger.debug(
             f"API: Successfully sent {len(batch)} score events to Langfuse API in batch mode"
         )

langfuse/_utils/__init__.py

@@ -1,13 +1,10 @@
 """@private"""
 
-import logging
 import typing
 from datetime import datetime, timezone
 
 from langfuse.model import PromptClient
 
-log = logging.getLogger("langfuse")
-
 
 def _get_timestamp() -> datetime:
     return datetime.now(timezone.utc)

langfuse/_utils/error_logging.py

@@ -1,8 +1,7 @@
 import functools
-import logging
 from typing import Any, Callable, List, Optional
 
-logger = logging.getLogger("langfuse")
+from langfuse.logger import langfuse_logger as logger
 
 
 def catch_and_log_errors(func: Callable[..., Any]) -> Callable[..., Any]:

langfuse/_utils/parse_error.py

@@ -1,4 +1,3 @@
-import logging
 from typing import Union
 
 # our own api errors
@@ -14,6 +13,7 @@
     UnauthorizedError,
 )
 from langfuse.api.core import ApiError
+from langfuse.logger import langfuse_logger as logger
 
 SUPPORT_URL = "https://langfuse.com/support"
 API_DOCS_URL = "https://api.reference.langfuse.com"
@@ -67,10 +67,9 @@ def generate_error_message_fern(error: Error) -> str:
 
 
 def handle_fern_exception(exception: Error) -> None:
-    log = logging.getLogger("langfuse")
-    log.debug(exception)
+    logger.debug(exception)
     error_message = generate_error_message_fern(exception)
-    log.error(error_message)
+    logger.error(error_message)
 
 
 def generate_error_message(exception: Union[APIError, APIErrors, Exception]) -> str:
@@ -95,7 +94,6 @@ def generate_error_message(exception: Union[APIError, APIErrors, Exception]) ->
 
 
 def handle_exception(exception: Union[APIError, APIErrors, Exception]) -> None:
-    log = logging.getLogger("langfuse")
-    log.debug(exception)
+    logger.debug(exception)
     error_message = generate_error_message(exception)
-    log.error(error_message)
+    logger.error(error_message)