trust_api.services.task_handlers

Task handlers for processing tasks received from the central hub via polling.

Each handler processes a specific task type and returns a result dict.

Attributes

`DATA_ACCESS_API_URL`
`CENTRAL_HUB_API_URL`
`IMAGING_API_URL`
`TRUST_API_KEY`
`TRUST_API_KEY_HEADER`
`TRUST_INTERNAL_SERVICE_KEY`
`TRUST_INTERNAL_SERVICE_KEY_HEADER`
`TASK_COHORT_QUERY`
`TASK_CREATE_IMAGING`
`TASK_DELETE_IMAGING`
`TASK_GET_IMAGING_STATUS`
`TASK_REIMPORT_STUDIES`
`TASK_UPDATE_USER_PROFILE`
`TASK_HANDLERS`

Functions

`_trust_internal_headers`(→ dict[str, str])	Return the auth header sent on every trust-internal call.
`handle_cohort_query`(→ dict[str, Any])	Process a cohort query task.
`_report_cohort_error_to_hub`(→ None)	Tell the hub this trust's cohort query failed.
`handle_create_imaging`(→ dict[str, Any])	Process an imaging project creation task.
`handle_delete_imaging`(→ dict[str, Any])	Process an imaging project deletion task.
`handle_get_imaging_status`(→ dict[str, Any])	Process an imaging status retrieval task.
`handle_reimport_studies`(→ dict[str, Any])	Process a reimport studies task.
`handle_update_user_profile`(→ dict[str, Any])	Process a user profile update task.

Module Contents

trust_api.services.task_handlers.DATA_ACCESS_API_URL = 'http://data-access-api:8000'

trust_api.services.task_handlers.CENTRAL_HUB_API_URL

trust_api.services.task_handlers.IMAGING_API_URL = 'http://imaging-api:8000'

trust_api.services.task_handlers.TRUST_API_KEY

trust_api.services.task_handlers.TRUST_API_KEY_HEADER

trust_api.services.task_handlers.TRUST_INTERNAL_SERVICE_KEY = ''

trust_api.services.task_handlers.TRUST_INTERNAL_SERVICE_KEY_HEADER = 'X-Trust-Internal-Service-Key'

trust_api.services.task_handlers._trust_internal_headers() → dict[str, str]

Return the auth header sent on every trust-internal call.

Used for imaging-api and data-access-api requests; both validate the same per-trust TRUST_INTERNAL_SERVICE_KEY against their stored hash.

Returns:: Single-entry dict mapping the configured header name to the trust-internal service key.
Return type:: dict[str, str]

trust_api.services.task_handlers.TASK_COHORT_QUERY = 'cohort_query'

trust_api.services.task_handlers.TASK_CREATE_IMAGING = 'create_imaging'

trust_api.services.task_handlers.TASK_DELETE_IMAGING = 'delete_imaging'

trust_api.services.task_handlers.TASK_GET_IMAGING_STATUS = 'get_imaging_status'

trust_api.services.task_handlers.TASK_REIMPORT_STUDIES = 'reimport_studies'

trust_api.services.task_handlers.TASK_UPDATE_USER_PROFILE = 'update_user_profile'

async trust_api.services.task_handlers.handle_cohort_query(payload: dict[str, Any]) → dict[str, Any]

Process a cohort query task.

Calls the local data-access-api, then pushes results back to the central hub. On data-access-api failure, posts an error report to the hub so the per-trust UI status switches from “running” to “error” instead of staying stuck.

Parameters:: payload – Task payload containing query details (query, query_name, encrypted_project_id, etc.)
Returns:: dict with success status and any error details.

async trust_api.services.task_handlers._report_cohort_error_to_hub(payload: dict[str, Any], error: str) → None

Tell the hub this trust’s cohort query failed.

Without this, the hub never sees the trust at all for this query and the UI leaves the per-trust chip on “running” forever. We swallow any failure here because the original cohort error is what callers care about — surfacing a hub-post failure would mask the real cause in the task result.

async trust_api.services.task_handlers.handle_create_imaging(payload: dict[str, Any]) → dict[str, Any]

Process an imaging project creation task.

Calls the local imaging-api to create the project.

Parameters:: payload – Task payload containing project details.
Returns:: dict with success status and imaging project result.

async trust_api.services.task_handlers.handle_delete_imaging(payload: dict[str, Any]) → dict[str, Any]

Process an imaging project deletion task.

Parameters:: payload – Task payload containing imaging_project_id.
Returns:: dict with success status.

async trust_api.services.task_handlers.handle_get_imaging_status(payload: dict[str, Any]) → dict[str, Any]

Process an imaging status retrieval task.

Gets the import status from the local imaging-api and reports back to the hub.

Parameters:: payload – Task payload containing imaging_project_id and encoded_query.
Returns:: dict with success status and imaging status result.

async trust_api.services.task_handlers.handle_reimport_studies(payload: dict[str, Any]) → dict[str, Any]

Process a reimport studies task.

Parameters:: payload – Task payload containing imaging_project_id and encoded_query.
Returns:: dict with success status.

async trust_api.services.task_handlers.handle_update_user_profile(payload: dict[str, Any]) → dict[str, Any]

Process a user profile update task.

Parameters:: payload – Task payload containing email and enabled fields.
Returns:: dict with success status.

trust_api.services.task_handlers.TASK_HANDLERS