Best Practices

# ❌ Inefficient - 3 API calls
result1 = client.tokenize("User 1: john@example.com")
result2 = client.tokenize("User 2: jane@example.com")
result3 = client.tokenize("User 3: bob@example.com")

# ✅ Efficient - 1 API call
combined_text = """
User 1: john@example.com
User 2: jane@example.com
User 3: bob@example.com
"""
result = client.tokenize(combined_text, policy="basic")

# Parse results by line
lines = result.text.split('\n')

from functools import lru_cache
import hashlib

class CachedBlindfold:
    def __init__(self, api_key):
        self.client = Blindfold(api_key=api_key)
        self._cache = {}

    def tokenize(self, text, policy="basic"):
        # Create cache key
        cache_key = hashlib.sha256(
            f"{text}:{policy}".encode()
        ).hexdigest()

        # Check cache
        if cache_key in self._cache:
            return self._cache[cache_key]

        # Call API
        result = self.client.tokenize(text, policy=policy)

        # Store in cache
        self._cache[cache_key] = result
        return result

# Use cached client
client = CachedBlindfold(api_key=api_key)

# First call - hits API
result1 = client.tokenize("john@example.com", policy="basic")

# Second call - uses cache (instant)
result2 = client.tokenize("john@example.com", policy="basic")

# Using GDPR policy (threshold: 0.35)
result = client.tokenize(text, policy="gdpr_eu")

# Using custom entities and threshold for specific needs
result = client.tokenize(
    text,
    entities=["person", "email address", "phone number"],
    score_threshold=0.60  # Higher threshold for fewer false positives
)

# Lower threshold to catch more edge cases
result = client.tokenize(
    text,
    entities=["person", "email address", "phone number"],
    score_threshold=0.25
)

import time
from blindfold import Blindfold, APIError, RateLimitError

def tokenize_with_retry(text, policy="gdpr_eu", max_retries=3):
    """Tokenize with exponential backoff retry strategy"""

    for attempt in range(max_retries):
        try:
            result = client.tokenize(text, policy=policy)
            return result

        except RateLimitError as e:
            # Use retry-after header
            wait_time = e.retry_after
            logger.warning(f"Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)
            continue

        except APIError as e:
            if e.status_code >= 500:
                # Server error - retry with exponential backoff
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                logger.warning(f"Server error. Retry {attempt + 1}/{max_retries} after {wait_time}s")
                time.sleep(wait_time)
                continue
            else:
                # Client error - don't retry
                raise

        except Exception as e:
            # Unexpected error - don't retry
            raise

    raise Exception(f"Failed after {max_retries} retries")

import logging
from datetime import datetime

class MonitoredBlindfold:
    def __init__(self, api_key):
        self.client = Blindfold(api_key=api_key)
        self.request_count = 0
        self.error_count = 0
        self.logger = logging.getLogger(__name__)

    def tokenize(self, text, policy="basic"):
        start_time = datetime.now()

        try:
            result = self.client.tokenize(text, policy=policy)
            self.request_count += 1

            # Log successful request
            duration = (datetime.now() - start_time).total_seconds()
            self.logger.info(
                f"Tokenize success: {result.entities_count} entities, "
                f"{duration:.2f}s, policy={policy}"
            )

            return result

        except Exception as e:
            self.error_count += 1
            duration = (datetime.now() - start_time).total_seconds()

            # Log error
            self.logger.error(
                f"Tokenize error: {str(e)}, "
                f"{duration:.2f}s, policy={policy}"
            )
            raise

    def get_stats(self):
        return {
            "total_requests": self.request_count,
            "total_errors": self.error_count,
            "error_rate": self.error_count / max(self.request_count, 1)
        }

# Example: DataDog monitoring
from datadog import statsd

def tokenize_with_metrics(text, policy="gdpr_eu"):
    with statsd.timed('blindfold.tokenize.duration'):
        try:
            result = client.tokenize(text, policy=policy)

            # Track success
            statsd.increment('blindfold.tokenize.success')
            statsd.gauge('blindfold.entities_detected', result.entities_count)

            return result

        except RateLimitError:
            statsd.increment('blindfold.tokenize.rate_limited')
            raise

        except Exception as e:
            statsd.increment('blindfold.tokenize.error')
            raise

import unittest
from blindfold import Blindfold

class TestPrivacyProtection(unittest.TestCase):
    def setUp(self):
        self.client = Blindfold(api_key="test_key")

    def test_email_detection(self):
        """Test that emails are properly detected"""
        result = self.client.tokenize(
            "Contact: john@example.com",
            policy="basic"
        )

        self.assertEqual(result.entities_count, 1)
        self.assertIn("<email_address_1>", result.text)
        self.assertIn("<email_address_1>", result.mapping)
        self.assertEqual(result.mapping["<email_address_1>"], "john@example.com")

    def test_detokenization_restores_original(self):
        """Test that detokenization works correctly"""
        original = "My email is john@example.com"

        # Tokenize
        protected = self.client.tokenize(original, policy="basic")

        # Detokenize
        restored = self.client.detokenize(
            protected.text,
            protected.mapping
        )

        self.assertEqual(restored.text, original)

    def test_no_pii_returns_unchanged(self):
        """Test that text without PII is unchanged"""
        text = "The weather is nice today"
        result = self.client.tokenize(text, policy="basic")

        self.assertEqual(result.entities_count, 0)
        self.assertEqual(result.text, text)
        self.assertEqual(result.mapping, {})

import pytest
from blindfold import Blindfold
from openai import OpenAI

@pytest.fixture
def clients():
    return {
        'blindfold': Blindfold(api_key=os.getenv("BLINDFOLD_API_KEY")),
        'openai': OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    }

def test_ai_integration_flow(clients):
    """Test complete privacy-preserving AI flow"""
    user_input = "My name is John Doe and my email is john@example.com"

    # Step 1: Protect PII
    protected = clients['blindfold'].tokenize(
        user_input,
        policy="gdpr_eu"
    )

    assert protected.entities_count > 0
    assert "John Doe" not in protected.text
    assert "john@example.com" not in protected.text

    # Step 2: Send to AI
    completion = clients['openai'].chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": protected.text}
        ]
    )

    ai_response = completion.choices[0].message.content

    # Step 3: Restore original data
    final_response = clients['blindfold'].detokenize(
        ai_response,
        protected.mapping
    )

    # Verify restoration worked
    if "<person_1>" in ai_response:
        assert "John Doe" in final_response.text
    if "<email_address_1>" in ai_response:
        assert "john@example.com" in final_response.text