in src/clients/comprehend_client.py [0:0]
def __init__(self, s3ol_access_point: str, pii_classification_thread_count: int = CONTAINS_PII_ENTITIES_THREAD_COUNT,
pii_redaction_thread_count: int = DETECT_PII_ENTITIES_THREAD_COUNT,
session_id: str = ''.join(choices(string.ascii_uppercase + string.digits, k=10)),
user_agent=DEFAULT_USER_AGENT, endpoint_url=None):
self.session_id = session_id
session_config = botocore.config.Config(
user_agent_extra=user_agent,
retries={
'max_attempts': COMPREHEND_MAX_RETRIES,
'mode': 'standard'
})
if endpoint_url is None:
self.comprehend = boto3.client('comprehend', config=session_config)
else:
self.comprehend = boto3.client('comprehend', config=session_config, endpoint_url=endpoint_url, verify=False)
self.comprehend.meta.events.register('before-sign.comprehend.*', self._add_session_header)
self.classification_executor_service = ThreadPoolExecutor(max_workers=pii_classification_thread_count)
self.redaction_executor_service = ThreadPoolExecutor(max_workers=pii_redaction_thread_count)
self.classify_metrics = Metrics(service_name=COMPREHEND, api=CONTAINS_PII_ENTITIES, s3ol_access_point=s3ol_access_point)
self.detection_metrics = Metrics(service_name=COMPREHEND, api=DETECT_PII_ENTITIES, s3ol_access_point=s3ol_access_point)