RdaktSession

Manages the bidirectional mapping between anonymization tokens and original values. Handles both full-text and streaming chunk-by-chunk deanonymization.

Import

from rdakt_ai import RdaktSession

Constructor

RdaktSession(
    *,
    session_id: str | None = None,
    entity_map: dict[str, str] | None = None,
)

Parameter	Type	Default	Description
`session_id`	`str \| None`	UUID4	Unique session identifier
`entity_map`	`dict[str, str] \| None`	`None`	Pre-existing token-to-value mappings

Properties

Property	Type	Description
`session_id`	`str`	Unique session identifier
`entity_map`	`dict[str, str]`	Read-only view of token → value mappings
`has_mappings`	`bool`	`True` if any mappings exist
`real_values`	`set[str]`	Set of original (non-anonymized) values
`token_names`	`set[str]`	All token names in the session

Methods

`add_mappings`

def add_mappings(self, new_mappings: dict[str, str]) -> None

Add new entity mappings to the session.

`get_anonymized`

def get_anonymized(self, real_value: str) -> str | None

Get the token for an original value, or None if not mapped.

`get_token_for_value`

def get_token_for_value(self, value: str) -> str | None

Alias for get_anonymized.

`deanonymize`

def deanonymize(self, text: str) -> str

Replace all tokens in text with their original values.

`deanonymize_structured`

def deanonymize_structured(self, data: Any) -> Any

Recursively deanonymize dicts, lists, and strings.

`deanonymize_chunk`

def deanonymize_chunk(self, chunk: str) -> list[str]

Deanonymize a streaming chunk. Returns a list of text fragments (may be empty if buffering).

`flush`

def flush(self) -> list[str]

Flush any remaining buffered content. Call at end of stream.

Constants

Constant	Value	Description
`MAX_BUFFER`	`256`	Maximum buffer size before forced flush

Usage

session = RdaktSession()
session.add_mappings({"<PERSON_1>": "John Smith", "<EMAIL_1>": "john@example.com"})

# Full-text deanonymization
result = session.deanonymize("Hello <PERSON_1>, your email is <EMAIL_1>")
# → "Hello John Smith, your email is john@example.com"

# Streaming deanonymization
fragments = session.deanonymize_chunk("Hello <PERS")
# → ["Hello "]  (buffering "<PERS")
fragments = session.deanonymize_chunk("ON_1>!")
# → ["John Smith", "!"]