AI API Gateway Data Privacy

class PrivacyGateway:
    """Privacy-aware AI API gateway"""
    
    def __init__(self, pii_detector, anonymizer, encryptor):
        self.detector = pii_detector
        self.anonymizer = anonymizer
        self.encryptor = encryptor
        self.audit_log = AuditLogger()
        
    async def process_request(
        self,
        request: dict,
        user_context: dict
    ) -> dict:
        """Process request with privacy protections"""
        
        # Check consent
        if not await self.check_consent(user_context):
            raise ConsentRequiredError()
        
        # Detect PII
        pii_entities = await self.detector.scan(request)
        
        # Log PII detection
        await self.audit_log.log({
            'event': 'pii_detected',
            'entities': [e.type for e in pii_entities],
            'user_id': user_context['user_id']
        })
        
        # Anonymize PII
        anonymized, mapping = await self.anonymizer.anonymize(
            request,
            pii_entities
        )
        
        # Encrypt if required
        if user_context.get('require_encryption'):
            anonymized = await self.encryptor.encrypt(anonymized)
        
        # Store mapping for deanonymization
        await self.store_mapping(
            request_id=request['id'],
            mapping=mapping,
            ttl=3600
        )
        
        return anonymized
    
    async def process_response(
        self,
        response: dict,
        request_id: str
    ) -> dict:
        """Restore original values in response"""
        
        # Retrieve mapping
        mapping = await self.get_mapping(request_id)
        
        # Deanonymize response
        deanonymized = await self.anonymizer.deanonymize(
            response,
            mapping
        )
        
        # Audit successful processing
        await self.audit_log.log({
            'event': 'response_processed',
            'request_id': request_id
        })
        
        return deanonymized

class PIIDetector:
    """Detect PII in text using ML models"""
    
    def __init__(self):
        self.patterns = self.load_patterns()
        self.ml_model = load_pii_model()
        
    async def scan(self, data: dict) -> List[PIIEntity]:
        """Scan data for PII entities"""
        entities = []
        
        # Pattern-based detection
        for field, value in data.items():
            if isinstance(value, str):
                for pii_type, pattern in self.patterns.items():
                    matches = pattern.findall(value)
                    for match in matches:
                        entities.append(PIIEntity(
                            type=pii_type,
                            value=match,
                            field=field
                        ))
        
        # ML-based detection for complex cases
        text_fields = ' '.join(
            str(v) for v in data.values() if isinstance(v, str)
        )
        ml_entities = await self.ml_model.detect(text_fields)
        entities.extend(ml_entities)
        
        return entities