Domain model: processes & tasks

Everything under task/ is the domain core. This page describes the framework; what each process does is in Process catalog.

Two concepts

Concept Definition

Process

A named business workflow with exactly two ordered steps: a request task and a handle-response task. Enumerated in task/process.py.

Task

A single executable unit with a unique task_name, e.g. thomson_reuters/request_personal_information_validation.

Initial (request) task

Step 1. Runs synchronously inside the Web API. Persists the input, sets the tool PENDING, validates, and starts the SWF workflow for step 2. Returns the correlation_id.

Handle task

Step 2. Runs asynchronously in the SWF worker. Reloads the input, calls the vendor, archives raw output to S3, writes verdicts, emits events.

The four processes (task/process.py):

class Process:

    VALIDATE_PERSONAL_INFORMATION = 'validate_personal_information'
    DOWNLOAD_PERSONAL_BACKGROUND_REPORT = 'download_personal_background_report'
    DOWNLOAD_BUSINESS_BACKGROUND_REPORT = 'download_business_background_report'
    VALIDATE_IDENTITY_DOCUMENT = 'validate_identity_document'

Task package layout

Every task lives in its own package with a fixed file layout. Match it exactly when adding one.

task/<vendor>/<task_name>/
├── __init__.py           # re-exports TaskImplementation / TaskInput / TaskOutput (+ validator)
├── implementation.py     # class TaskImplementation  — the executable logic
├── input.py              # class TaskInput           — typed input + from_dict/to_dict
├── output.py             # class TaskOutput          — typed output + to_dict
└── input_validator.py    # request tasks only: TaskInputValidator, raises ValidationError

The GenericTask contract

class GenericTask:

    class Input:
        def to_dict(self) -> dict: ...
        @staticmethod
        def from_dict(d: dict): ...

    class Output:
        def to_dict(self): ...

    def __init__(self, correlation_id: str, logger: Optional[Logger] = None): ...

    def execute(self, input_: Input) -> Output: ...

    @staticmethod
    def build_task_input(d: dict): ...
    @staticmethod
    def get_task_name(): ...
    @staticmethod
    def get_process_name(): ...

    def _update_task_failure_in_task_execution_table(self, reason): ...
Method Purpose

build_task_input(d)

Deserialize a dict (HTTP body or SWF payload) into a typed Input.

execute(input_)

Run the business logic and return a typed Output.

get_task_name()

Unique id used for TaskFactory dispatch and process mapping.

get_process_name()

Parent process constant — drives metrics, DynamoDB keys and events.

_update_task_failure_in_task_execution_table(reason)

Mark the task failed (and usually the tool ERROR); invoked by the failure handler.

InitialTask adds

class InitialTask(GenericTask):
    def __init__(self, correlation_id, process_source, requested_by=None, logger=None): ...

    def __save_task_in_task_execution_table(self, input_): ...

    @staticmethod
    def get_available_process_sources() -> Tuple[ProcessSource, ...]: ...

  • process_source — who triggered it: INTERNAL, SUPPLIER_DASHBOARD or ONBOARDING (components/process_source.py).

  • requested_by — the caller’s login id, persisted on the task row.

  • get_available_process_sources() — the allow-list enforced by POST /tasks/execute.

TaskExecutionError

A structured validation error serialized straight into API responses as {"error_class": …​, "error_body": …​}. ValidationError (input validators) and CannotRunWorkflow (B2B Task Manager) both subclass it, which is why a rejected input and an "already running" workflow both surface as clean 400 responses.

Factories

TaskFactory (task/task_factory.py)

Built once at worker startup with every shared client injected, then build(task_definition: dict) dispatches on task_name:

{
  "platform": "automatic-uw",
  "task_name": "thomson_reuters/handle_personal_information_validation_response",
  "task_correlation_id": "3f6c1b64-8f0a-4d2b-9c2e-9f1a5f7c1d22",
  "process_source": "internal",
  "requested_by": "underwriter@payability.com",
  "task_input": {
    "correlation_id": "3f6c1b64-8f0a-4d2b-9c2e-9f1a5f7c1d22"
  },
  "gci": "1"
}

Unknown task_name raises TaskNotSupported. Injected components (see swf_app.py): task_execution_table, final_status_table, tool_table, thomson_reuters_client, au10tix_client, entrust_client, hubspot_client, supplier_event_logger, b2b_task_manager, b2b_documents, s3_client, event_bridge_client, retry_client, payability_internal_api.

The Web API builds its own TaskFactory per request inside the handlers that need one (execute_task.py, get_task_result.py, delete_task_results.py).

ProcessFactory (task/process_factory.py)

  • identify_process_name(s) — accepts a process name or a task name and resolves the process (case-insensitive); raises ProcessNotSupported otherwise.

  • build(process_name, correlation_id=None, process_source=None, requested_by=None) — returns both task instances for the process, sharing one correlation_id. POST /tasks/execute executes only the first; DELETE /tasks/result walks both looking for PurgableS3ObjectProducer implementations.

PREDEFINED_PROCESS_STEPS maps each process to its two step names:

Process Step 1 (request) Step 2 (handle)

validate_personal_information

thomson_reuters/request_personal_information_validation

thomson_reuters/handle_personal_information_validation_response

download_personal_background_report

thomson_reuters/request_personal_background_report

thomson_reuters/handle_personal_background_report_response

download_business_background_report

thomson_reuters/request_business_background_report

thomson_reuters/handle_business_background_report_response

validate_identity_document

au10tix/request_identity_document_validation

au10tix/handle_identity_document_validation_response

entrust/handle_identity_document_validation_response is registered in TaskFactory but is deliberately absent from PREDEFINED_PROCESS_STEPS: it has no synchronous request step, and is fired by the Onfido webhook instead. POST /tasks/execute special-cases that task name and treats it as an admin "reprocess the latest Entrust run" action — see api/tasks.adoc#entrust-reprocess.

Task inputs and outputs

Task Input fields Output

au10tix/request_identity_document_validation

supplier_key, person_id, document_id

{correlation_id}

au10tix/handle_identity_document_validation_response

supplier_key, person_id, document_id (from the SWF payload)

{raw_result_s3_path}

entrust/handle_identity_document_validation_response

supplier_key, person_id, workflow_run_id, optional applicant_id

{raw_result_s3_path, status, verdict?, document_actions?}

thomson_reuters/request_personal_information_validation

supplier_key, person_id, first_name, last_name, birth_day, birth_month, birth_year, ssn, street, city, state, zip_code

{correlation_id}

thomson_reuters/handle_personal_information_validation_response

correlation_id

{match_result}

thomson_reuters/request_personal_background_report

supplier_key, person_id, first_name, last_name, ssn, optional test, test_timeout

{correlation_id}

thomson_reuters/handle_personal_background_report_response

correlation_id

{raw_result_s3_path}

thomson_reuters/request_business_background_report

supplier_key, person_id, business_name, state

{correlation_id}

thomson_reuters/handle_business_background_report_response

correlation_id

{raw_result_s3_path}

TaskInput.from_dict runs remove_empty_values first, so empty strings and nulls are treated as absent — a missing required field raises KeyError, which the API maps to a 400 naming the field.

Failure handling

Every side-effecting execute() body runs inside this context manager (utils/task_utils.py):

@contextmanager
def task_failure_handling(task_instance: GenericTask):
    try:
        yield
    except CannotRunWorkflow:
        raise
    except Exception as e:
        if task_instance.logger:
            task_instance.logger.error(msg=f'Task has been failed with error - {str(e)}')
        task_instance._update_task_failure_in_task_execution_table({'error': str(e)})
        raise

On any error except CannotRunWorkflow (which propagates cleanly so the API can answer 400/409): log it, mark the task failed in TaskExecutionTable, flip the tool to ERROR with the error message, then re-raise so the worker reports activity failure and Sentry captures it.

Domain exceptions (task/exceptions/request.py)

Exception Meaning

ResultNotFound

The vendor returned no matching records (e.g. a Thomson Reuters search GroupCount of 0).

MalformedResponse

An expected XML/JSON node was missing from the vendor response.

MissingResponseSection

A personal background report is missing a required section.

components/retry/exceptions.py adds RetryException (a ConnectionError) for an exhausted poll loop, and components/pyb_documents.py raises DocumentNotFound for a 404 from the document service.

Adding a process

  1. Add the constant to task/process.py.

  2. Create task/<vendor>/request_<name>/ and task/<vendor>/handle_<name>_response/, each with implementation.py, input.py, output.py (plus input_validator.py on the request task if it needs validation).

  3. Register both task names in PREDEFINED_PROCESS_STEPS (task/process_factory.py).

  4. Add dispatch branches in TaskFactory.build injecting the clients the tasks need.

  5. If a new vendor client is required, add it under components/ and wire it into the TaskFactory construction in swf_app.py (and any Web API handler that builds a factory).

  6. If the process maps to a tool, extend ToolType and ToolType.get_by_process_name (components/aws/dynamodb/tool/tool_type.py), and ToolTable.SUPPORTED_TOOLS if the row should be auto-created for every supplier.

  7. If underwriters may override its verdict, add the process to VALIDATION_FIELDS in interface/web_api/endpoint_handling/update_final_status.py.

  8. Add unit tests under development/unit_tests/task_/… and update these docs (Process catalog, DynamoDB tables, EventBridge events).