Module pii_codex.services.adapters.detection_adapters.aws_detection_adapter
Expand source code
from typing import List
from pii_codex.config import PII_MAPPER
from pii_codex.models.analysis import DetectionResultItem, DetectionResult
from pii_codex.services.adapters.detection_adapters.detection_adapter_base import (
BasePIIDetectionAdapter,
)
class AWSComprehendPIIDetectionAdapter(BasePIIDetectionAdapter):
def convert_analyzed_item(self, pii_detection: dict) -> List[DetectionResultItem]:
"""
Converts an AWS Comprehend detect_pii() result into a collection of DetectionResultItem
@param pii_detection: dict from AWS Comprehend detect_pii {
'Entities': [
{
'Score': ...,
'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'|
'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'|
'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'|
'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'|
'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'|
'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'|
'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'|
'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER',
'BeginOffset': 123,
'EndOffset': 123
},
]
}
@return: List[DetectionResultItem]
"""
return [
DetectionResultItem(
entity_type=PII_MAPPER.convert_aws_comprehend_pii_to_common_pii_type(
result["Type"]
).name,
score=result["Score"],
start=result["BeginOffset"],
end=result["EndOffset"],
)
for result in pii_detection["Entities"]
]
def convert_analyzed_collection(
self, pii_detections: List[dict]
) -> List[DetectionResult]:
"""
Converts a collection of AWS Comprehend detect_pii() results to a collection of DetectionResult.
@param pii_detections: List[dict] of response from AWS Comprehend detect_pii - [{
'Entities': [
{
'Score': ...,
'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'|
'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'|
'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'|
'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'|
'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'|
'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'|
'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'|
'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER',
'BeginOffset': 123,
'EndOffset': 123
},
]
}]
"""
detection_results: List[DetectionResult] = []
for i, detection in enumerate(pii_detections):
# Return results in formatted Analysis Result List object
detection_results.append(
DetectionResult(
index=i,
detections=self.convert_analyzed_item(pii_detection=detection),
)
)
return detection_results
Classes
class AWSComprehendPIIDetectionAdapter-
Expand source code
class AWSComprehendPIIDetectionAdapter(BasePIIDetectionAdapter): def convert_analyzed_item(self, pii_detection: dict) -> List[DetectionResultItem]: """ Converts an AWS Comprehend detect_pii() result into a collection of DetectionResultItem @param pii_detection: dict from AWS Comprehend detect_pii { 'Entities': [ { 'Score': ..., 'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'| 'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'| 'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'| 'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'| 'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'| 'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'| 'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'| 'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER', 'BeginOffset': 123, 'EndOffset': 123 }, ] } @return: List[DetectionResultItem] """ return [ DetectionResultItem( entity_type=PII_MAPPER.convert_aws_comprehend_pii_to_common_pii_type( result["Type"] ).name, score=result["Score"], start=result["BeginOffset"], end=result["EndOffset"], ) for result in pii_detection["Entities"] ] def convert_analyzed_collection( self, pii_detections: List[dict] ) -> List[DetectionResult]: """ Converts a collection of AWS Comprehend detect_pii() results to a collection of DetectionResult. @param pii_detections: List[dict] of response from AWS Comprehend detect_pii - [{ 'Entities': [ { 'Score': ..., 'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'| 'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'| 'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'| 'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'| 'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'| 'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'| 'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'| 'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER', 'BeginOffset': 123, 'EndOffset': 123 }, ] }] """ detection_results: List[DetectionResult] = [] for i, detection in enumerate(pii_detections): # Return results in formatted Analysis Result List object detection_results.append( DetectionResult( index=i, detections=self.convert_analyzed_item(pii_detection=detection), ) ) return detection_resultsAncestors
Methods
def convert_analyzed_collection(self, pii_detections: List[dict]) ‑> List[DetectionResult]-
Converts a collection of AWS Comprehend detect_pii() results to a collection of DetectionResult.
@param pii_detections: List[dict] of response from AWS Comprehend detect_pii - [{ 'Entities': [ { 'Score': …, 'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'| 'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'| 'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'| 'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'| 'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'| 'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'| 'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'| 'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER', 'BeginOffset': 123, 'EndOffset': 123 }, ] }]
Expand source code
def convert_analyzed_collection( self, pii_detections: List[dict] ) -> List[DetectionResult]: """ Converts a collection of AWS Comprehend detect_pii() results to a collection of DetectionResult. @param pii_detections: List[dict] of response from AWS Comprehend detect_pii - [{ 'Entities': [ { 'Score': ..., 'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'| 'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'| 'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'| 'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'| 'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'| 'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'| 'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'| 'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER', 'BeginOffset': 123, 'EndOffset': 123 }, ] }] """ detection_results: List[DetectionResult] = [] for i, detection in enumerate(pii_detections): # Return results in formatted Analysis Result List object detection_results.append( DetectionResult( index=i, detections=self.convert_analyzed_item(pii_detection=detection), ) ) return detection_results def convert_analyzed_item(self, pii_detection: dict) ‑> List[DetectionResultItem]-
Converts an AWS Comprehend detect_pii() result into a collection of DetectionResultItem
@param pii_detection: dict from AWS Comprehend detect_pii { 'Entities': [ { 'Score': …, 'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'| 'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'| 'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'| 'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'| 'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'| 'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'| 'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'| 'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER', 'BeginOffset': 123, 'EndOffset': 123 }, ] } @return: List[DetectionResultItem]
Expand source code
def convert_analyzed_item(self, pii_detection: dict) -> List[DetectionResultItem]: """ Converts an AWS Comprehend detect_pii() result into a collection of DetectionResultItem @param pii_detection: dict from AWS Comprehend detect_pii { 'Entities': [ { 'Score': ..., 'Type': 'BANK_ACCOUNT_NUMBER'|'BANK_ROUTING'|'CREDIT_DEBIT_NUMBER'|'CREDIT_DEBIT_CVV'| 'CREDIT_DEBIT_EXPIRY'|'PIN'|'EMAIL'|'ADDRESS'|'NAME'|'PHONE'|'SSN'|'DATE_TIME'|'PASSPORT_NUMBER'| 'DRIVER_ID'|'URL'|'AGE'|'USERNAME'|'PASSWORD'|'AWS_ACCESS_KEY'|'AWS_SECRET_KEY'|'IP_ADDRESS'| 'MAC_ADDRESS'|'ALL'|'LICENSE_PLATE'|'VEHICLE_IDENTIFICATION_NUMBER'|'UK_NATIONAL_INSURANCE_NUMBER'| 'CA_SOCIAL_INSURANCE_NUMBER'|'US_INDIVIDUAL_TAX_IDENTIFICATION_NUMBER'| 'UK_UNIQUE_TAXPAYER_REFERENCE_NUMBER'|'IN_PERMANENT_ACCOUNT_NUMBER'|'IN_NREGA'| 'INTERNATIONAL_BANK_ACCOUNT_NUMBER'|'SWIFT_CODE'|'UK_NATIONAL_HEALTH_SERVICE_NUMBER'| 'CA_HEALTH_NUMBER'|'IN_AADHAAR'|'IN_VOTER_NUMBER', 'BeginOffset': 123, 'EndOffset': 123 }, ] } @return: List[DetectionResultItem] """ return [ DetectionResultItem( entity_type=PII_MAPPER.convert_aws_comprehend_pii_to_common_pii_type( result["Type"] ).name, score=result["Score"], start=result["BeginOffset"], end=result["EndOffset"], ) for result in pii_detection["Entities"] ]