in src/clients/comprehend_client.py [0:0]
def _update_doc_with_pii_entities(self, document: Document, language) -> Document:
start_time = time.time()
response = None
try:
response = self.comprehend.detect_pii_entities(Text=document.text, LanguageCode=language)
finally:
if response is not None:
self.detection_metrics.add_fault_count(response['ResponseMetadata']['RetryAttempts'])
self.detection_metrics.add_latency(start_time, time.time())
# updating the document itself instead of creating a new copy to save space
document.pii_entities = response['Entities']
document.pii_classification = {entity['Type']: max(entity['Score'], document.pii_classification[entity['Type']])
if entity['Type'] in document.pii_classification else entity['Score']
for entity in response['Entities']}
return document