Examples

This section provides complete, working examples for VeilData streaming functionality.

Streaming Examples

The following examples demonstrate various streaming redaction use cases.

Example 1: Basic Streaming

This example shows basic chunk-based streaming with cross-chunk entity detection.

 1def example_1_basic_streaming():
 2    """Basic example of streaming redaction."""
 3    print("=== Example 1: Basic Streaming ===\n")
 4
 5    # Setup detector
 6    patterns = {
 7        "EMAIL": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
 8        "PHONE": r"\d{3}-\d{4}",
 9    }
10    detector = RegexDetector(patterns)
11    pipeline = DetectionPipeline(detector)
12
13    # Create buffer
14    buffer = StreamingRedactionBuffer(pipeline, overlap_size=20)
15
16    # Simulate chunks (in real use, these would come from a file or stream)
17    chunks = [
18        "My email is john.do",
19        "[email protected] and my phone is 555-",
20        "1234. Contact me anytime!",
21    ]
22
23    print("Processing chunks...")
24    for i, chunk in enumerate(chunks, 1):
25        output = buffer.add_chunk(chunk)
26        print(f"Chunk {i}: {chunk!r}")
27        print(f"  Output: {output!r}")
28
29    final = buffer.finalize()
30    print(f"Final: {final!r}\n")
31
32    # Get stats
33    stats = buffer.get_stats()
34    print(f"Stats: {stats}\n")

Example 2: File Streaming

Process a file in chunks and write the redacted output.

 1def example_2_file_streaming():
 2    """Example of streaming a file."""
 3    print("=== Example 2: File Streaming ===\n")
 4
 5    # Create a test file
 6    with open("test_input.txt", "w") as f:
 7        for i in range(10):
 8            f.write(f"Line {i}: Contact user{i}@example.com or call 555-{i:04d}\n")
 9
10    # Setup
11    patterns = {"EMAIL": r"\S+@\S+", "PHONE": r"\d{3}-\d{4}"}
12    detector = RegexDetector(patterns)
13    pipeline = DetectionPipeline(detector)
14    store = TokenStore()
15
16    def read_file_chunks(filepath, chunk_size=100):
17        """Generator that yields file chunks."""
18        with open(filepath, "r") as f:
19            while True:
20                chunk = f.read(chunk_size)
21                if not chunk:
22                    break
23                yield chunk
24
25    # Process using stream_redact
26    print("Redacting test_input.txt...")
27    output_chunks = list(
28        stream_redact(
29            read_file_chunks("test_input.txt"), pipeline, overlap_size=30, store=store
30        )
31    )
32
33    # Write output
34    with open("test_output.txt", "w") as f:
35        f.write("".join(output_chunks))
36
37    print("✓ Redacted to test_output.txt")
38    print(f"✓ Found {len(store.mappings)} entities\n")
39
40    # Show first few mappings
41    for i, (token, original) in enumerate(list(store.mappings.items())[:5], 1):
42        print(f"  {token} → {original}")
43
44    print()

Example 3: API Usage

Instructions for using the streaming API endpoint.

 1def example_3_api_usage():
 2    """Example of using the streaming API."""
 3    print("=== Example 3: API Usage ===\n")
 4
 5    print("To use the streaming API:")
 6    print()
 7    print("1. Start the API server:")
 8    print("   $ uvicorn veildata.api.app:app --reload")
 9    print()
10    print("2. Stream redaction with curl:")
11    print('   $ curl -X POST "http://localhost:8000/v1/redact/stream" \\')
12    print('     -H "Content-Type: text/plain" \\')
13    print("     --data-binary @input.txt > output.txt")
14    print()
15    print("3. With custom parameters:")
16    print(
17        '   $ curl -X POST "http://localhost:8000/v1/redact/stream?overlap_size=256&return_store=true" \\'
18    )
19    print('     -H "Content-Type: text/plain" \\')
20    print("     --data-binary @input.txt -v 2>&1 | grep -i x-token-store")
21    print()

Example 4: CLI Usage

Command-line examples for streaming mode.

 1def example_4_cli_usage():
 2    """Example of using the CLI streaming mode."""
 3    print("=== Example 4: CLI Usage ===\n")
 4
 5    print("Stream redaction from command line:")
 6    print()
 7    print("1. Basic streaming:")
 8    print("   $ veildata redact input.txt --stream --output output.txt")
 9    print()
10    print("2. With custom chunk and overlap sizes:")
11    print("   $ veildata redact large_file.txt --stream \\")
12    print("       --chunk-size 8192 --overlap 1024 \\")
13    print("       --output redacted.txt --store store.json")
14    print()
15    print("3. With timing and verbose output:")
16    print("   $ veildata redact input.txt --stream --time --verbose \\")
17    print("       --output output.txt")
18    print()
19    print("4. Pipe stdin:")
20    print("   $ cat large_file.txt | veildata redact --stream > output.txt")
21    print()

Complete Example File

View the complete example file with all functions:

examples/streaming_example.py
  1"""
  2Example demonstrating streaming redaction with VeilData.
  3
  4This example shows how to:
  51. Process large files in chunks using the streaming buffer
  62. Use the streaming API endpoint
  73. Handle cross-chunk entity detection
  8"""
  9
 10from veildata.detectors import RegexDetector
 11from veildata.pipeline import DetectionPipeline
 12from veildata.revealers import TokenStore
 13from veildata.streaming_buffer import StreamingRedactionBuffer, stream_redact
 14
 15
 16def example_1_basic_streaming():
 17    """Basic example of streaming redaction."""
 18    print("=== Example 1: Basic Streaming ===\n")
 19
 20    # Setup detector
 21    patterns = {
 22        "EMAIL": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
 23        "PHONE": r"\d{3}-\d{4}",
 24    }
 25    detector = RegexDetector(patterns)
 26    pipeline = DetectionPipeline(detector)
 27
 28    # Create buffer
 29    buffer = StreamingRedactionBuffer(pipeline, overlap_size=20)
 30
 31    # Simulate chunks (in real use, these would come from a file or stream)
 32    chunks = [
 33        "My email is john.do",
 34        "[email protected] and my phone is 555-",
 35        "1234. Contact me anytime!",
 36    ]
 37
 38    print("Processing chunks...")
 39    for i, chunk in enumerate(chunks, 1):
 40        output = buffer.add_chunk(chunk)
 41        print(f"Chunk {i}: {chunk!r}")
 42        print(f"  Output: {output!r}")
 43
 44    final = buffer.finalize()
 45    print(f"Final: {final!r}\n")
 46
 47    # Get stats
 48    stats = buffer.get_stats()
 49    print(f"Stats: {stats}\n")
 50
 51
 52def example_2_file_streaming():
 53    """Example of streaming a file."""
 54    print("=== Example 2: File Streaming ===\n")
 55
 56    # Create a test file
 57    with open("test_input.txt", "w") as f:
 58        for i in range(10):
 59            f.write(f"Line {i}: Contact user{i}@example.com or call 555-{i:04d}\n")
 60
 61    # Setup
 62    patterns = {"EMAIL": r"\S+@\S+", "PHONE": r"\d{3}-\d{4}"}
 63    detector = RegexDetector(patterns)
 64    pipeline = DetectionPipeline(detector)
 65    store = TokenStore()
 66
 67    def read_file_chunks(filepath, chunk_size=100):
 68        """Generator that yields file chunks."""
 69        with open(filepath, "r") as f:
 70            while True:
 71                chunk = f.read(chunk_size)
 72                if not chunk:
 73                    break
 74                yield chunk
 75
 76    # Process using stream_redact
 77    print("Redacting test_input.txt...")
 78    output_chunks = list(
 79        stream_redact(
 80            read_file_chunks("test_input.txt"), pipeline, overlap_size=30, store=store
 81        )
 82    )
 83
 84    # Write output
 85    with open("test_output.txt", "w") as f:
 86        f.write("".join(output_chunks))
 87
 88    print("✓ Redacted to test_output.txt")
 89    print(f"✓ Found {len(store.mappings)} entities\n")
 90
 91    # Show first few mappings
 92    for i, (token, original) in enumerate(list(store.mappings.items())[:5], 1):
 93        print(f"  {token} → {original}")
 94
 95    print()
 96
 97
 98def example_3_api_usage():
 99    """Example of using the streaming API."""
100    print("=== Example 3: API Usage ===\n")
101
102    print("To use the streaming API:")
103    print()
104    print("1. Start the API server:")
105    print("   $ uvicorn veildata.api.app:app --reload")
106    print()
107    print("2. Stream redaction with curl:")
108    print('   $ curl -X POST "http://localhost:8000/v1/redact/stream" \\')
109    print('     -H "Content-Type: text/plain" \\')
110    print("     --data-binary @input.txt > output.txt")
111    print()
112    print("3. With custom parameters:")
113    print(
114        '   $ curl -X POST "http://localhost:8000/v1/redact/stream?overlap_size=256&return_store=true" \\'
115    )
116    print('     -H "Content-Type: text/plain" \\')
117    print("     --data-binary @input.txt -v 2>&1 | grep -i x-token-store")
118    print()
119
120
121def example_4_cli_usage():
122    """Example of using the CLI streaming mode."""
123    print("=== Example 4: CLI Usage ===\n")
124
125    print("Stream redaction from command line:")
126    print()
127    print("1. Basic streaming:")
128    print("   $ veildata redact input.txt --stream --output output.txt")
129    print()
130    print("2. With custom chunk and overlap sizes:")
131    print("   $ veildata redact large_file.txt --stream \\")
132    print("       --chunk-size 8192 --overlap 1024 \\")
133    print("       --output redacted.txt --store store.json")
134    print()
135    print("3. With timing and verbose output:")
136    print("   $ veildata redact input.txt --stream --time --verbose \\")
137    print("       --output output.txt")
138    print()
139    print("4. Pipe stdin:")
140    print("   $ cat large_file.txt | veildata redact --stream > output.txt")
141    print()
142
143
144def main():
145    """Run all examples."""
146    example_1_basic_streaming()
147    example_2_file_streaming()
148    example_3_api_usage()
149    example_4_cli_usage()
150
151    print("=" * 50)
152    print("Examples complete! Check test_output.txt for results.")
153    print("=" * 50)
154
155
156if __name__ == "__main__":
157    main()

Running the Examples

To run the complete example:

cd examples
python streaming_example.py

This will:

  1. Demonstrate basic buffer usage with console output

  2. Create test files (test_input.txt, test_output.txt)

  3. Show token store mappings

  4. Display API and CLI usage instructions

Additional Examples

Integration with Custom Detectors

from veildata.detectors import RegexDetector, SpacyDetector
from veildata.pipeline import DetectionPipeline
from veildata.streaming_buffer import stream_redact

# Combine multiple detectors
patterns = {
    "EMAIL": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
    "PHONE": r"\d{3}-\d{3}-\d{4}",
}
regex_detector = RegexDetector(patterns)
pipeline = DetectionPipeline(regex_detector)

# Stream from generator
def chunk_generator(filepath):
    with open(filepath, 'r') as f:
        while True:
            chunk = f.read(4096)
            if not chunk:
                break
            yield chunk

# Process stream
for output in stream_redact(chunk_generator("input.txt"), pipeline):
    print(output, end="")

Progress Tracking

Track progress while streaming large files:

from veildata.streaming_buffer import StreamingRedactionBuffer
from veildata.engine import build_redactor

redactor, store = build_redactor(method="regex")
buffer = StreamingRedactionBuffer(redactor, overlap_size=512)

file_size = os.path.getsize("large_file.txt")
processed = 0

with open("large_file.txt") as f:
    while chunk := f.read(4096):
        processed += len(chunk)
        output = buffer.add_chunk(chunk)

        # Show progress
        percent = (processed / file_size) * 100
        print(f"\rProgress: {percent:.1f}%", end="", flush=True)

        # Write output
        # ... handle output ...

final = buffer.finalize()
print(f"\n✓ Complete! {buffer.get_stats()['total_entities_redacted']} entities redacted")

Memory-Efficient Pipeline

For extremely large files, write directly to output without accumulating:

from veildata.streaming_buffer import StreamingRedactionBuffer
from veildata.engine import build_redactor

redactor, store = build_redactor(method="regex")
buffer = StreamingRedactionBuffer(redactor, overlap_size=512)

with open("input.txt") as infile, open("output.txt", "w") as outfile:
    while chunk := infile.read(8192):  # 8KB chunks
        redacted = buffer.add_chunk(chunk)
        if redacted:
            outfile.write(redacted)

    # Don't forget the final chunk
    final = buffer.finalize()
    if final:
        outfile.write(final)

# Save store for reversibility
store.save("store.json")

See Also