API Reference

This section documents the public API of veildata.

Main Interface

class veildata.Compose(transforms)

Bases: object

class veildata.Module

Bases: object

Base component class for all VeilData transformers.

eval() → Module

Shortcut for setting eval mode.

forward(x: Any) → Any

Override this method with the transformation logic.

train(mode: bool = True) → Module

Set the module to training or eval mode.

class veildata.RegexRedactor(pattern: str, redaction_token: str = '[REDACTED_{counter}]', store: TokenStore | None = None)

Bases: Module

Redact substrings in text using a regex pattern, optionally tracking reversibility.

forward(text: str) → str

Override this method with the transformation logic.

class veildata.TokenStore

Bases: object

A reversible token store that maps redacted tokens back to original values.

Example

store = TokenStore() store.record(“[REDACTED_1]”, “john.doe@example.com”) text = store.reveal(“Contact [REDACTED_1]”)

bulk_record(mappings: Dict[str, str]) → None

Add multiple mappings at once.

clear() → None

Reset the store.

classmethod load(path: str)

property mappings: Dict[str, str]

Return a copy of the internal mapping.

record(token: str, original: str) → None

Record a mapping between a redacted token and its original value.

reveal(text: str) → str

Replace all stored tokens with their original values in text.

save(path: str)

Redactors

The core of VeilData is the Redactor interface.

class veildata.redactors.regex.RegexRedactor(pattern: str, redaction_token: str = '[REDACTED_{counter}]', store: TokenStore | None = None)

Bases: Module

Redact substrings in text using a regex pattern, optionally tracking reversibility.

forward(text: str) → str

Override this method with the transformation logic.

class veildata.redactors.ner_spacy.SpacyNERRedactor(model: str = 'en_core_web_sm', entities: list[str] | None = None, redaction_token: str = '[REDACTED_{counter}]', store: TokenStore | None = None)

Bases: Module

Redact named entities in text using a spaCy model, with optional reversible tracking.

forward(text: str) → str

Override this method with the transformation logic.

class veildata.redactors.ner_bert.BERTNERRedactor(model_name: str = 'dslim/bert-base-NER', redaction_token: str = '[REDACTED_{counter}]', store: TokenStore | None = None, device: str | None = None, use_fp16: bool = False)

Bases: Module

Redact named entities in text using a fine-tuned BERT NER model.

forward(text: str) → str

Redact entities using model predictions.

train(mode: bool = True) → BERTNERRedactor

Toggle train/eval mode on BERT model.

Composition & Pipeline

class veildata.transforms.Compose(transforms)

Bases: object

Reversibility

class veildata.revealers.TokenStore

Bases: object

A reversible token store that maps redacted tokens back to original values.

Example

store = TokenStore() store.record(“[REDACTED_1]”, “john.doe@example.com”) text = store.reveal(“Contact [REDACTED_1]”)

bulk_record(mappings: Dict[str, str]) → None

Add multiple mappings at once.

clear() → None

Reset the store.

classmethod load(path: str)

property mappings: Dict[str, str]

Return a copy of the internal mapping.

record(token: str, original: str) → None

Record a mapping between a redacted token and its original value.

reveal(text: str) → str

Replace all stored tokens with their original values in text.

save(path: str)

Configuration

class veildata.core.config.BertConfig(*, enabled: bool = False, model_path: str = 'dslim/bert-base-NER', threshold: float = 0.5, label_mapping: Dict[str, str] | None = None)

Bases: BaseModel

enabled: bool

label_mapping: Dict[str, str] | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_path: str

threshold: float

class veildata.core.config.GlobalOptions(*, hybrid: ~veildata.core.config.HybridOptions = <factory>, fallback: str = 'plaintext')

Bases: BaseModel

fallback: str

hybrid: HybridOptions

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class veildata.core.config.HybridOptions(*, strategy: str = 'union', prefer: str = 'ml')

Bases: BaseModel

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

prefer: str

strategy: str

class veildata.core.config.MLConfig(*, spacy: ~veildata.core.config.SpacyConfig = <factory>, bert: ~veildata.core.config.BertConfig = <factory>)

Bases: BaseModel

bert: BertConfig

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

spacy: SpacyConfig

class veildata.core.config.RedactionMethod(*values)

Bases: str, Enum

HYBRID = 'hybrid'

NER_BERT = 'ner_bert'

NER_SPACY = 'ner_spacy'

REGEX = 'regex'

class veildata.core.config.SpacyConfig(*, enabled: bool = False, model: str = 'en_core_web_lg', pii_labels: List[str] | None = None)

Bases: BaseModel

enabled: bool

model: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

pii_labels: List[str] | None

class veildata.core.config.TraversalConfig(*, policy: ~veildata.core.config.TraversalPolicy = TraversalPolicy.ALLOW, keys_to_redact: ~typing.List[str] = <factory>, keys_to_ignore: ~typing.List[str] = <factory>, max_depth: int = 100)

Bases: BaseModel

Configuration for JSON traversal.

keys_to_ignore: List[str]

keys_to_redact: List[str]

max_depth: int

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

policy: TraversalPolicy

class veildata.core.config.TraversalPolicy(*values)

Bases: str, Enum

ALLOW = 'allow'

DENY = 'deny'

class veildata.core.config.VeilConfig(*, method: ~veildata.core.config.RedactionMethod = RedactionMethod.REGEX, patterns: ~typing.Dict[str, str] | None = None, pattern: ~typing.Dict[str, str] | None = None, ml: ~veildata.core.config.MLConfig = <factory>, traversal: ~veildata.core.config.TraversalConfig = <factory>, options: ~veildata.core.config.GlobalOptions = <factory>)

Bases: BaseModel

Root configuration for VeilData.

get_patterns() → Dict[str, str]

Helper to get patterns from either ‘patterns’ or ‘pattern’ field.

method: RedactionMethod

ml: MLConfig

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

options: GlobalOptions

pattern: Dict[str, str] | None

patterns: Dict[str, str] | None

traversal: TraversalConfig

veildata.core.config.load_config(config_path: str | None = None, verbose: bool = False) → VeilConfig

Load configuration from a file or environment variables.

Parameters:

• config_path – Path to the config file (YAML, JSON, TOML).

• verbose – Whether to print loading status.

Returns:

VeilConfig object.

Raises:

• ConfigMissingError – If specified config file is not found.

• ValidationError – If config is invalid.

Engine

veildata.engine.build_redactor(method: str = 'regex', detect_mode: str = 'rules', config_path: str | None = None, ml_config_path: str | None = None, verbose: bool = False, config: VeilConfig | None = None) → Tuple[Module, TokenStore]

Factory function to build a redactor based on configuration.

veildata.engine.build_revealer(store_path: str)

Build a callable revealer from a saved TokenStore mapping.

Returns:

str) -> str

Return type:

callable(text

veildata.engine.list_available_redactors() → List[str]

Return available redaction engines.

veildata.engine.list_engines()

Streaming

Streaming redaction buffer for chunk-based processing.

This module provides a buffer that can process text in chunks while correctly detecting and redacting entities that span across chunk boundaries.

class veildata.streaming_buffer.ChunkMetadata(chunk_index: int, input_size: int, output_size: int, buffer_size: int, entities_detected: int)

Bases: object

Metadata about a processed chunk for debugging and tracking.

buffer_size: int

chunk_index: int

entities_detected: int

input_size: int

output_size: int

class veildata.streaming_buffer.StreamingRedactionBuffer(pipeline: DetectionPipeline, overlap_size: int = 512, store: TokenStore | None = None, redaction_format: str = '[REDACTED_{counter}]')

Bases: object

Buffer for streaming redaction with cross-chunk entity detection.

This class maintains a sliding window to ensure entities that span chunk boundaries are correctly detected and redacted. It progressively yields sanitized output while maintaining minimal state.

Example

>>> from veildata.detectors import RegexDetector
>>> from veildata.pipeline import DetectionPipeline
>>> detector = RegexDetector({"EMAIL": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"})
>>> pipeline = DetectionPipeline(detector)
>>> buffer = StreamingRedactionBuffer(pipeline, overlap_size=20)
>>> for chunk in ["My email is john.do", "[email protected] and more"]:
...     output = buffer.add_chunk(chunk)
...     print(output, end="")
>>> print(buffer.finalize())

Parameters:

• pipeline – DetectionPipeline to use for entity detection

• overlap_size – Number of characters to retain between chunks for boundary detection

• store – Optional TokenStore to record redacted mappings

• redaction_format – Format string for redaction tokens (default: “[REDACTED_{counter}]”)

add_chunk(chunk: str) → str

Add a chunk of text and return the redacted output for the safe zone.

The safe zone is the portion of the buffer that is guaranteed not to contain any entities that might continue in the next chunk.

Parameters:

chunk – New text chunk to process

Returns:

Redacted text for the safe portion of the buffer

finalize() → str

Process any remaining text in the buffer.

This should be called after all chunks have been added to ensure the overlap region is processed.

Returns:

Redacted text for the remaining buffer contents

get_metadata() → List[ChunkMetadata]

Get metadata about all processed chunks.

Returns:

List of ChunkMetadata objects

get_stats() → dict

Get statistics about the streaming process.

Returns:

Dictionary with processing statistics

reset() → None

Reset the buffer to initial state.

veildata.streaming_buffer.stream_redact(chunks: Generator[str, None, None], pipeline: DetectionPipeline, overlap_size: int = 512, store: TokenStore | None = None) → Generator[str, None, None]

Convenience function for streaming redaction.

This is a simple wrapper around StreamingRedactionBuffer for common use cases.

Parameters:

• chunks – Generator yielding text chunks

• pipeline – DetectionPipeline to use for entity detection

• overlap_size – Number of characters to retain between chunks

• store – Optional TokenStore to record redacted mappings

Yields:

Redacted text chunks

Example

>>> def read_file_chunks(filepath, chunk_size=4096):
...     with open(filepath, 'r') as f:
...         while True:
...             chunk = f.read(chunk_size)
...             if not chunk:
...                 break
...             yield chunk
>>>
>>> from veildata.engine import build_redactor
>>> redactor, store = build_redactor(method="regex")
>>> for redacted_chunk in stream_redact(read_file_chunks("input.txt"), redactor):
...     print(redacted_chunk, end="")