Utilities

components/retry/

class Retry:
    def __init__(self, total: int = 10, delay: int = 10): ...

    def execute(self, method, method_args: dict, total=None, delay=None,
                retry_condition=None, exception_force_list=()): ...

Generic poll-and-retry. execute calls method(**method_args) up to total times, sleeping delay seconds between attempts, and returns the first result that satisfies the stopping condition:

  • retry_condition(result) truthy → retry;

  • no retry_condition and a falsy result → retry;

  • an exception in exception_force_list → retry.

Exhausting the attempts raises RetryException (a ConnectionError, components/retry/exceptions.py).

Instances: the worker injects Retry(total=20); the Web API’s factories use the default Retry(). Call sites override per loop — the Au10tix readiness and result polls use total=10, delay=5.

Retry sleeps on the calling thread, which is fine for a worker thread but would block a web request. Keep it out of the request path.

components/logger.py

def get_logger(name: str, correlation_id: str) -> logging.Logger:
    log = logging.getLogger(name)
    log.setLevel(level=logging.DEBUG)
    formatter = logging.Formatter(f'{correlation_id} | %(name)s | %(levelname)s | %(asctime)s - %(message)s')

    if not log.handlers:
        ch = logging.StreamHandler()
        ch.setLevel(level=logging.INFO)
        log.addHandler(ch)

    for handler in log.handlers:
        handler.setFormatter(formatter)

    return log

Every line is prefixed with the correlation id, so one execution can be grepped out of the worker’s interleaved thread output. Safe to call repeatedly for the same name: the handler is attached once and only the formatter is refreshed, so handlers never accumulate and lines are never duplicated.

TaskFactory creates one logger per task (name = task_name, correlation_id = task_correlation_id); the Entrust endpoint handlers use entrust/create_session and entrust/webhook, the latter re-created with the workflow_run_id once the body is parsed.

interface/web_api/web_request_logger.py

Logs one INFO line per handled request: [{login_id}:{source}] {path} ({payload}) under the logger name WEB REQUEST LOGGER. Called at the top of every handler, so request bodies — including PII in validate_personal_information inputs — appear in the Web API log stream.

utils/

Module Contents

task_utils.py

task_failure_handling(task_instance) — the failure context manager (domain-model.adoc#failure-handling). find_key_in_tr_response(result, key_name, prefix='ns') — resolves Thomson Reuters namespace prefixes, raising MalformedResponse when the node is absent.

string_utils.py

remove_accents(text) (NFKD fold; leaves ł, ø, æ, œ, ß alone), is_matching_fuzzy(a, b, accuracy=100) (thefuzz.token_set_ratio), is_full_name_match(given, expected) — the accent-folded, order-insensitive, exact-token name comparison used for IDNameMatch.

dict_utils.py

remove_empty_values(d) — drops falsy values; every TaskInput.from_dict runs it first, so "", null, 0 and false are treated as absent.

time_utils.py

round_time_diff(start_time, end_time) — elapsed seconds, 2 dp.

metric_utils.py

send_metric(metrics_namespace, name, value, unit=SECONDS, process_name='automatic-uw-tool') — publishes through pyb-sdk’s `CriticalMetricNamespace with PybUserProcess.UNDERWRITING / PybService.INTG. Swallows and logs any error, so a metrics outage never fails a request or a task.

remove_empty_values means a legitimately falsy input value cannot reach a task — for example test=false is simply dropped (which is harmless, since absence and False mean the same thing there). Keep that in mind when adding boolean or numeric task inputs.

Test helpers

development/unit_tests/utils_/test_dict_utils.py and test_string_utils.py cover the helpers above; mockito is used for all mocking. See Testing.