AWS clients

All AWS access goes through these wrappers. Table shapes are in DynamoDB tables; S3 key conventions in S3 datalake.

DynamoDB

Every client uses boto3.resource('dynamodb') and a table name from settings.

File Table setting Role

components/task_execution_table.py

SUPPLIER_DATA_VALIDATION_TASK_TABLE_NAME

Task lifecycle. save, get, update, get_by_supplier_key_and_process_name (GSI, paginated), get_latest_processed_task, get_latest_task. Status constants PENDING_STATUS, PROCESSED_STATUS, FAILED_STATUS, PURGED_STATUS; update rejects anything else.

components/aws/dynamodb/final_status_table.py

FINAL_STATUS_TABLE_NAME

Verdicts per correlation_id. save, update, get_by_correlation_id, get_latest_by_correlation_id; writes a changelog row on every mutation; paginates queries correctly.

components/aws/dynamodb/tool/tool_table.py

TOOL_TABLE_NAME

Per-supplier/person tool state. save, update (upsert), get_by_supplier_and_type, get_all_by_supplier, get_tools_and_create_missing, create_missing_tools. Lookups use paginated scan with filters.

components/aws/dynamodb/changelog_table.py

CHANGELOG_TABLE_NAME

Append-only audit. save(supplier_key, changelog_type, data, changelog_action) with enums ChangelogType (FINAL_STATUS, TOOL) and ChangelogAction (INIT, UPDATE); stamps app_version from version.py.

FinalStatusTable and ToolTable both take a ChangelogTable in their constructor — that is how auditing is guaranteed rather than remembered.

Enums and exceptions

Module Contents

aws/dynamodb/tool/tool_status.py

ToolStatus: NOT_RUN, PENDING, ERROR, NOT_VERIFIED, VERIFIED, VERIFIED_ACCESS, VERIFIED_ADVANCE, VERIFIED_ACCESS_AND_ADVANCE.

aws/dynamodb/tool/tool_type.py

ToolType (storage key → display name) plus get_by_process_name().

aws/dynamodb/exceptions.py

NotFoundFinalStatus, NotFoundTool, NotSupportedToolType.

 
class ToolType(Enum):
    THOMSON_REUTERS_CLEAR_ID = 'Thomson Reuters Clear ID (KYC Check)'
    THOMSON_REUTERS_PERSONAL_BACKGROUND_REPORT = 'Thomson Reuters Personal Background Report'
    AU10TIX = 'Au10tix'
    ENTRUST = 'Entrust Identity Verification'

    @staticmethod
    def get_by_process_name(process_name: str) -> 'ToolType':
        supported_processes = {
            Process.VALIDATE_IDENTITY_DOCUMENT: ToolType.AU10TIX,
            Process.VALIDATE_PERSONAL_INFORMATION: ToolType.THOMSON_REUTERS_CLEAR_ID,
            Process.DOWNLOAD_PERSONAL_BACKGROUND_REPORT: ToolType.THOMSON_REUTERS_PERSONAL_BACKGROUND_REPORT
        }

        try:
            return supported_processes[process_name]
        except KeyError:
            raise NotSupportedToolType(process_name=process_name)

Au10tix / Entrust coexistence. Entrust Identity Verification (Onfido) is the eventual replacement for Au10tix identity-document validation. During the rollout both vendors run under the same validate_identity_document process, so the GSI, dedup logic and admin reporting keep working unchanged. ToolType.get_by_process_name therefore still maps validate_identity_document to AU10TIX; the Entrust code paths reference ToolType.ENTRUST directly. Do not switch that mapping until Au10tix is fully retired.

ToolTable.SUPPORTED_TOOLS does include ENTRUST, so GET /tools auto-creates an Entrust row (status NOT_RUN) for every supplier/person it is asked about. Only the process → tool mapping still points at Au10tix.

S3

File Contents

components/s3_client.py

upload_json_data(data, bucket, key) — JSON upload with ACL: bucket-owner-full-control; get_json_data(bucket, path) — download and parse.

components/s3_purgable/purgable_s3_object_producer.py

PurgableS3ObjectProducer interface: purge(supplier_key, person_id, s3_object_purger) → List[str].

components/s3_purgable/s3_object_purger.py

S3ObjectPurger.delete_s3_object(bucket_name, object_key).

Implemented by the three Thomson Reuters handle tasks and driven by DELETE /tasks/result.

SWF (worker side)

components/swf_client.py — a boto3.client('swf') activity worker.

Method Behaviour

poll_activity_task(domain, identity, task_list)

Loops on poll_for_activity_task until a taskToken appears; swallows ReadTimeoutError and ConnectionClosedError (long-poll timeouts are normal).

notify_about_task_success(task_token, result)

respond_activity_task_completed with the JSON-encoded output.

notify_about_task_failure(task_token, reason)

respond_activity_task_failed. An Exception becomes {error_type, error_message}; a short human-readable reason (≤ 256 chars) is derived and the full payload goes into details (≤ 32768 chars). Exceeding SWF’s caps would make the call itself fail and leave the activity hanging until its timeout.

EventBridge

File Contents

event_bridge/event_bridge_event.py

Base EventBridgeEvent holding the put_events entry; get_body().

event_bridge/event_bridge_client.py

put_event(event); raises ConnectionError on a non-zero FailedEntryCount.

event_bridge/events/

ProcessFinishedEvent, ThomsonReutersDataMatchEvent, ThomsonReutersDataNotMatchEvent, ThomsonReutersNotWorkingEvent.

Catalog and payloads: EventBridge events.

CloudWatch

Not a components/ client: utils/metric_utils.py wraps pyb-sdk’s `CriticalMetricNamespace. See Observability.

IAM surface

The two ECS task roles need, at minimum: DynamoDB read/write on the four tables and the supplier_key-process_name-index GSI; s3:PutObject/GetObject/DeleteObject on the datalake bucket; SWF poll and respond on the domain and task list (worker); events:PutEvents on the bus; and CloudWatch PutMetricData. Secrets Manager access is only needed at build time by config_setup.py.