Data Privacy and Security: A Developer's Guide to Handling Sensitive Data With DuckDB

pip install duckdb --upgrade

import duckdb
import pandas as pd

# Create a DuckDB connection - it's this simple!
conn = duckdb.connect(database=':memory:')

CREATE TABLE customer_data AS 
SELECT 
 'John Doe' as name,
 '123-45-6789' as ssn,
 '[email protected]' as email,
    '123-456-7890' as phone;

-- Implement PII masking patterns
CREATE TABLE masked_data AS
SELECT 
 regexp_replace(name, '[a-zA-Z]', 'X') as masked_name,
 regexp_replace(ssn, '[0-9]', '*') as masked_ssn,
 regexp_replace(email, '(^[^@]+)(@.*$)', '****$2') as masked_email,
 regexp_replace(phone, '[0-9]', '#') as masked_phone
FROM customer_data;

import duckdb
import pandas as pd

def mask_pii_data():
 # Create a DuckDB connection in memory
 conn = duckdb.connect(database=':memory:')
 
 try:
 # Create and populate sample data
 conn.execute("""
 CREATE TABLE customer_data AS 
 SELECT 
 'John Doe' as name,
 '123-45-6789' as ssn,
 '[email protected]' as email,
 '123-456-7890' as phone
 """)
 
 # Implement PII masking
 conn.execute("""
 CREATE TABLE masked_data AS
 SELECT 
 regexp_replace(name, '[a-zA-Z]', 'X') as masked_name,
 regexp_replace(ssn, '[0-9]', '*') as masked_ssn,
 regexp_replace(email, '(^[^@]+)(@.*$)', '****$2') as masked_email,
 regexp_replace(phone, '[0-9]', '#') as masked_phone
 FROM customer_data
 """)
 
 # Fetch and display original data
 print("Original Data:")
 original_data = conn.execute("SELECT * FROM customer_data").fetchdf()
 print(original_data)
 print("\n")
 
 # Fetch and display masked data
 print("Masked Data:")
 masked_data = conn.execute("SELECT * FROM masked_data").fetchdf()
 print(masked_data)
 
 return original_data, masked_data
 
 except Exception as e:
 print(f"An error occurred: {str(e)}")
 return None, None
 
 finally:
 # Close the connection
        conn.close()

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()