stubgen

logfire-api/logfire_api/_internal/constants.pyi

@@ -9,8 +9,10 @@ LOGGING_TO_OTEL_LEVEL_NUMBERS: Incomplete
 ATTRIBUTES_LOG_LEVEL_NAME_KEY: Incomplete
 ATTRIBUTES_LOG_LEVEL_NUM_KEY: Incomplete
 
+
 def log_level_attributes(level: LevelName | int) -> dict[str, otel_types.AttributeValue]: ...
 
+
 SpanTypeType: Incomplete
 ATTRIBUTES_SPAN_TYPE_KEY: Incomplete
 ATTRIBUTES_PENDING_SPAN_REAL_PARENT_KEY: Incomplete
@@ -21,6 +23,7 @@ DISABLE_CONSOLE_KEY: Incomplete
 ATTRIBUTES_JSON_SCHEMA_KEY: Incomplete
 ATTRIBUTES_LOGGING_ARGS_KEY: Incomplete
 ATTRIBUTES_VALIDATION_ERROR_KEY: str
+ATTRIBUTES_SCRUBBED_KEY: str
 NULL_ARGS_KEY: str
 PENDING_SPAN_NAME_SUFFIX: str
 LOGFIRE_BASE_URL: str

logfire-api/logfire_api/_internal/formatter.pyi

@@ -1,8 +1,8 @@
 import ast
 import executing
 import types
-from .constants import MESSAGE_FORMATTED_VALUE_LENGTH_LIMIT as MESSAGE_FORMATTED_VALUE_LENGTH_LIMIT
-from .scrubbing import BaseScrubber as BaseScrubber
+from .constants import ATTRIBUTES_SCRUBBED_KEY as ATTRIBUTES_SCRUBBED_KEY, MESSAGE_FORMATTED_VALUE_LENGTH_LIMIT as MESSAGE_FORMATTED_VALUE_LENGTH_LIMIT
+from .scrubbing import BaseScrubber as BaseScrubber, ScrubbedNote as ScrubbedNote
 from .utils import truncate_string as truncate_string
 from _typeshed import Incomplete
 from logfire._internal.stack_info import get_user_frame_and_stacklevel as get_user_frame_and_stacklevel

logfire-api/logfire_api/_internal/scrubbing.pyi

@@ -1,46 +1,91 @@
 import re
-from .constants import ATTRIBUTES_JSON_SCHEMA_KEY as ATTRIBUTES_JSON_SCHEMA_KEY, ATTRIBUTES_LOG_LEVEL_NAME_KEY as ATTRIBUTES_LOG_LEVEL_NAME_KEY, ATTRIBUTES_LOG_LEVEL_NUM_KEY as ATTRIBUTES_LOG_LEVEL_NUM_KEY, ATTRIBUTES_MESSAGE_KEY as ATTRIBUTES_MESSAGE_KEY, ATTRIBUTES_MESSAGE_TEMPLATE_KEY as ATTRIBUTES_MESSAGE_TEMPLATE_KEY, ATTRIBUTES_PENDING_SPAN_REAL_PARENT_KEY as ATTRIBUTES_PENDING_SPAN_REAL_PARENT_KEY, ATTRIBUTES_SAMPLE_RATE_KEY as ATTRIBUTES_SAMPLE_RATE_KEY, ATTRIBUTES_SPAN_TYPE_KEY as ATTRIBUTES_SPAN_TYPE_KEY, ATTRIBUTES_TAGS_KEY as ATTRIBUTES_TAGS_KEY, NULL_ARGS_KEY as NULL_ARGS_KEY, RESOURCE_ATTRIBUTES_PACKAGE_VERSIONS as RESOURCE_ATTRIBUTES_PACKAGE_VERSIONS
+import typing_extensions
+from .constants import (
+    ATTRIBUTES_JSON_SCHEMA_KEY as ATTRIBUTES_JSON_SCHEMA_KEY,
+    ATTRIBUTES_LOG_LEVEL_NAME_KEY as ATTRIBUTES_LOG_LEVEL_NAME_KEY,
+    ATTRIBUTES_LOG_LEVEL_NUM_KEY as ATTRIBUTES_LOG_LEVEL_NUM_KEY,
+    ATTRIBUTES_MESSAGE_KEY as ATTRIBUTES_MESSAGE_KEY,
+    ATTRIBUTES_MESSAGE_TEMPLATE_KEY as ATTRIBUTES_MESSAGE_TEMPLATE_KEY,
+    ATTRIBUTES_PENDING_SPAN_REAL_PARENT_KEY as ATTRIBUTES_PENDING_SPAN_REAL_PARENT_KEY,
+    ATTRIBUTES_SAMPLE_RATE_KEY as ATTRIBUTES_SAMPLE_RATE_KEY,
+    ATTRIBUTES_SCRUBBED_KEY as ATTRIBUTES_SCRUBBED_KEY,
+    ATTRIBUTES_SPAN_TYPE_KEY as ATTRIBUTES_SPAN_TYPE_KEY,
+    ATTRIBUTES_TAGS_KEY as ATTRIBUTES_TAGS_KEY,
+    NULL_ARGS_KEY as NULL_ARGS_KEY,
+    RESOURCE_ATTRIBUTES_PACKAGE_VERSIONS as RESOURCE_ATTRIBUTES_PACKAGE_VERSIONS,
+)
 from .stack_info import STACK_INFO_KEYS as STACK_INFO_KEYS
 from .utils import ReadableSpanDict as ReadableSpanDict
 from _typeshed import Incomplete
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from opentelemetry.sdk.trace import Event
-from typing import Any, Callable, Sequence
+from typing import Any, Callable, Sequence, TypedDict
 
 DEFAULT_PATTERNS: Incomplete
+JsonPath: typing_extensions.TypeAlias = 'tuple[str | int, ...]'
+
 
 @dataclass
 class ScrubMatch:
     """An object passed to the [`scrubbing_callback`][logfire.configure(scrubbing_callback)] function."""
-    path: tuple[str | int, ...]
+
+    path: JsonPath
     value: Any
     pattern_match: re.Match[str]
+
+
 ScrubCallback = Callable[[ScrubMatch], Any]
 
+
+class ScrubbedNote(TypedDict):
+    path: JsonPath
+    matched_substring: str
+
+
 @dataclass
 class ScrubbingOptions:
     """Options for redacting sensitive data."""
+
     callback: ScrubCallback | None = ...
     extra_patterns: Sequence[str] | None = ...
 
+
 class BaseScrubber(ABC):
     SAFE_KEYS: Incomplete
+
     @abstractmethod
     def scrub_span(self, span: ReadableSpanDict): ...
     @abstractmethod
-    def scrub(self, path: tuple[str | int, ...], value: Any) -> Any: ...
+    def scrub_value(self, path: JsonPath, value: Any) -> tuple[Any, list[ScrubbedNote]]: ...
+
 
 class NoopScrubber(BaseScrubber):
     def scrub_span(self, span: ReadableSpanDict): ...
-    def scrub(self, path: tuple[str | int, ...], value: Any) -> Any: ...
+    def scrub_value(self, path: JsonPath, value: Any) -> tuple[Any, list[ScrubbedNote]]: ...
+
 
 class Scrubber(BaseScrubber):
     """Redacts potentially sensitive data."""
+
     def __init__(self, patterns: Sequence[str] | None, callback: ScrubCallback | None = None) -> None: ...
     def scrub_span(self, span: ReadableSpanDict): ...
+    def scrub_value(self, path: JsonPath, value: Any) -> tuple[Any, list[ScrubbedNote]]: ...
+
+
+class SpanScrubber:
+    """Does the actual scrubbing work.
+
+    This class is separate from Scrubber so that it can be instantiated more regularly
+    and hold and mutate state about the span being scrubbed, specifically the scrubbed notes.
+    """
+
+    scrubbed: list[ScrubbedNote]
+
+    def __init__(self, parent: Scrubber) -> None: ...
+    def scrub_span(self, span: ReadableSpanDict): ...
     def scrub_event_attributes(self, event: Event, index: int): ...
-    def scrub(self, path: tuple[str | int, ...], value: Any) -> Any:
+    def scrub(self, path: JsonPath, value: Any) -> Any:
         """Redacts sensitive data from `value`, recursing into nested sequences and mappings.
 
         `path` is a list of keys and indices leading to `value` in the span.