A Developer’s Guide to Handling Sensitive Data With DuckDB

import duckdb
import pandas as pd

def demonstrate_data_redaction():
    # Create a connection
    conn = duckdb.connect(':memory:')
    
    # Create sample data with various sensitive information
    conn.execute("""
        CREATE TABLE sensitive_info AS SELECT * FROM (
            VALUES
                ('John Doe', '[email protected]', 'CC: 4532-1234-5678-9012', 'Normal text'),
                ('Jane Smith', '[email protected]', 'SSN: 123-45-6789', 'Some notes'),
                ('Bob Wilson', '[email protected]', 'Password: SecretPass123!', 'Regular info'),
                ('Alice Brown', '[email protected]', 'API_KEY=abc123xyz', 'Basic text')
            ) AS t(name, email, sensitive_field, normal_text);
    """)
    
    # Define redaction rules
    redaction_rules = {
        'email': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',  # Email pattern
        'sensitive_field': r'(CC:\s*\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}|SSN:\s*\d{3}-\d{2}-\d{4}|Password:\s*\S+|API_KEY=\S+)',  # Various sensitive patterns
        'name': r'[A-Z][a-z]+ [A-Z][a-z]+'  # Full name pattern
    }
    
    # Show original data
    print("Original Data:")
    print(conn.execute("SELECT * FROM sensitive_info").fetchdf())
    
    # Apply redaction
    redact_sensitive_data(conn, 'sensitive_info', redaction_rules)
    
    # Show redacted data
    print("\nRedacted Data:")
    print(conn.execute("SELECT * FROM redacted_data").fetchdf())
    
    return conn

def redact_sensitive_data(conn, table_name, rules):
    """
    Redact sensitive data based on specified patterns.
    
    Parameters:
    - conn: DuckDB connection
    - table_name: Name of the table containing sensitive data
    - rules: Dictionary of column names and their corresponding regex patterns to match sensitive data
    """
    redaction_cases = []
    # This creates a CASE statement for each column
	# If the pattern matches, the value is redacted
	# If not, the original value is kept
    for column, pattern in rules.items():
        redaction_cases.append(f"""
            CASE 
                WHEN regexp_matches({column}, '{pattern}') 
                THEN '(REDACTED)' 
                ELSE {column} 
            END as {column}
        """)
    
    query = f"""
    CREATE TABLE redacted_data AS
    SELECT 
        {', '.join(redaction_cases)}
    FROM {table_name};
    """
    conn.execute(query)

# Example with custom redaction patterns
def demonstrate_custom_redaction():
    conn = duckdb.connect(':memory:')
    
    # Create sample data
    conn.execute("""
        CREATE TABLE customer_data AS SELECT * FROM (
            VALUES
                ('John Doe', '123-45-6789', 'ACC#12345', '$5000'),
                ('Jane Smith', '987-65-4321', 'ACC#67890', '$3000'),
                ('Bob Wilson', '456-78-9012', 'ACC#11111', '$7500')
            ) AS t(name, ssn, account, balance);
    """)
    
    # Define custom redaction rules with different patterns
    custom_rules = {
        'name': {
            'pattern': r'[A-Z][a-z]+ [A-Z][a-z]+',
            'replacement': lambda match: f"{match[0][0]}*** {match[0].split()[1][0]}***"
        },
        'ssn': {
            'pattern': r'\d{3}-\d{2}-\d{4}',
            'replacement': 'XXX-XX-XXXX'
        },
        'account': {
            'pattern': r'ACC#\d{5}',
            'replacement': 'ACC#*****'
        }
    }
    
    def apply_custom_redaction(conn, table_name, rules):
        redaction_cases = []
        for column, rule in rules.items():
            redaction_cases.append(f"""
                CASE 
                    WHEN regexp_matches({column}, '{rule['pattern']}') 
                    THEN '{rule['replacement']}'
                    ELSE {column} 
                END as {column}
            """)
        
        query = f"""
        CREATE TABLE custom_redacted AS
        SELECT 
            {', '.join(redaction_cases)},
            balance  -- Keep this column unchanged
        FROM {table_name};
        """
        conn.execute(query)
    
    # Show original data
    print("\nOriginal Customer Data:")
    print(conn.execute("SELECT * FROM customer_data").fetchdf())
    
    # Apply custom redaction
    apply_custom_redaction(conn, 'customer_data', custom_rules)
    
    # Show results
    print("\nCustom Redacted Data:")
    print(conn.execute("SELECT * FROM custom_redacted").fetchdf())

# Run demonstrations
print("=== Basic Redaction Demo ===")
demonstrate_data_redaction()
print("\n=== Custom Redaction Demo ===")
demonstrate_custom_redaction()