nat.observability.processor.redaction.redaction_processor

Attributes

RedactionInputT
RedactionDataT
logger

Classes

RedactionProcessor	Abstract base class for redaction processors.
RedactionContextState	Generic context state for redaction results.
RedactionManager	Generic manager for atomic redaction operations.
RedactionContext	Generic context provider for redaction operations.

Module Contents

RedactionInputT

RedactionDataT

logger

class RedactionProcessor

Bases: nat.observability.processor.processor.Processor[RedactionInputT, RedactionInputT], Generic[RedactionInputT, RedactionDataT]

Abstract base class for redaction processors.

abstractmethod should_redact(item: RedactionInputT) → bool

Async:

Determine if this item should be redacted.

Args:

item (RedactionInputT): The item to check.

Returns:

bool: True if the item should be redacted, False otherwise.

abstractmethod redact_item(item: RedactionInputT) → RedactionInputT

Async:

Redact the item.

Args:

item (RedactionInputT): The item to redact.

Returns:

RedactionInputT: The redacted item.

async process(item: RedactionInputT) → RedactionInputT

Perform redaction on the item if it should be redacted.

Args:

item (RedactionInputT): The item to process.

Returns:

RedactionInputT: The processed item.

class RedactionContextState

Generic context state for redaction results.

Stores the redaction result in a context variable to avoid redundant callback executions within the same request context.

redaction_result: contextvars.ContextVar[bool | None]

class RedactionManager(context_state: RedactionContextState)

Bases: Generic[RedactionDataT]

Generic manager for atomic redaction operations.

Handles state mutations and ensures atomic callback execution with proper result caching within a request context.

Args:

RedactionDataT: The type of data being processed for redaction decisions.

_context_state

set_redaction_result(result: bool) → None

Set the redaction result in the context.

Args:

result (bool): The redaction result to cache.

clear_redaction_result() → None

Clear the cached redaction result from the context.

async redaction_check(

callback: collections.abc.Callable[Ellipsis, Any],

data: RedactionDataT,

) → bool

Execute redaction callback with atomic result caching.

Checks for existing cached results first, then executes the callback and caches the result atomically. Since data is static per request, subsequent calls within the same context return the cached result.

Supports sync/async functions, generators, and async generators.

Args:

callback (Callable[…, Any]): The callback to execute (sync/async function, generator, etc.). data (RedactionDataT): The data to pass to the callback for redaction decision.

Returns:

bool: True if the item should be redacted, False otherwise.

class RedactionContext(context: RedactionContextState)

Bases: Generic[RedactionDataT]

Generic context provider for redaction operations.

Provides read-only access to redaction state and manages the RedactionManager lifecycle through async context managers.

Args:

RedactionDataT: The type of data being processed for redaction decisions.

_context_state: RedactionContextState

property redaction_result: bool | None

Get the current redaction result from context.

Returns:

bool | None: The cached redaction result, or None if not set.

async redaction_manager() → collections.abc.AsyncGenerator[RedactionManager[RedactionDataT], None]

Provide a redaction manager within an async context.

Creates and yields a RedactionManager instance for atomic redaction operations within the current context.

Yields:

RedactionManager[RedactionDataT]: Manager instance for redaction operations.