flip_api.project_services.services.image_service

Functions

has_pending_imaging_tasks(→ bool)	Check if any CREATE_IMAGING tasks for this project are still pending or in progress.
to_utc_aware(→ datetime.datetime)	Convert a datetime to a timezone-aware UTC datetime. If the input is None, returns the minimum datetime.
base64_url_encode(→ str)	Encode a string using Base64 URL encoding without padding.
get_imaging_projects(...)	Retrieve imaging projects associated with a given project ID.
delete_imaging_project(→ bool)	Queue a task to delete an imaging project and update its status in the database.
get_xnat_project_status_info(...)	Retrieve the XNAT project status information for a given XNAT project ID.
_get_latest_imaging_status(...)	Look up the most recent completed GET_IMAGING_STATUS task result for a trust and XNAT project.
get_imaging_project_statuses(...)	Retrieve the imaging project statuses from local DB and queue status refresh tasks for trusts.
update_xnat_user_profile(→ None)	Queue user profile update tasks for all trusts' XNAT instances.
reimport_failed_studies(→ bool)	Queue reimport tasks for failed studies, ensuring that reimports are only attempted if the

Module Contents

flip_api.project_services.services.image_service.has_pending_imaging_tasks(project_id: uuid.UUID, db: sqlmodel.Session) → bool

Check if any CREATE_IMAGING tasks for this project are still pending or in progress.

Used to distinguish the transient “tasks not yet executed” state from genuine failure when no XNATProjectStatus records exist after project approval.

Parameters:

• project_id – The project to check.

• db – Database session.

Returns:

True if at least one CREATE_IMAGING task for this project is PENDING or IN_PROGRESS.

flip_api.project_services.services.image_service.to_utc_aware(dt: datetime.datetime | None) → datetime.datetime

Convert a datetime to a timezone-aware UTC datetime. If the input is None, returns the minimum datetime.

Parameters:

dt (datetime | None) – The datetime to normalise. Naive datetimes are assumed to be in UTC.

Returns:

A timezone-aware UTC datetime. Returns datetime.min with UTC tzinfo when dt is None.

Return type:

datetime

flip_api.project_services.services.image_service.base64_url_encode(data: str) → str

Encode a string using Base64 URL encoding without padding.

Parameters:

data (str) – The string to encode.

Returns:

The Base64 URL-encoded string with trailing = padding stripped.

Return type:

str

flip_api.project_services.services.image_service.get_imaging_projects(project_id: uuid.UUID, db: sqlmodel.Session) → list[flip_api.domain.schemas.projects.ImagingProject]

Retrieve imaging projects associated with a given project ID.

Parameters:

• project_id (UUID) – The ID of the project to retrieve imaging projects for.

• db (Session) – The database session for executing queries.

Returns:

A list of ImagingProject objects associated with the given project ID.

Return type:

list[ImagingProject]

Raises:

• SQLAlchemyError – If there is an error executing the database query.

• Exception – If there is an unexpected error during the retrieval process.

flip_api.project_services.services.image_service.delete_imaging_project(imaging_project: flip_api.domain.schemas.projects.ImagingProject, db: sqlmodel.Session) → bool

Queue a task to delete an imaging project and update its status in the database.

Parameters:

• imaging_project (ImagingProject) – The imaging project to delete.

• db (Session) – The database session for executing queries.

Returns:

True if the task was queued and status updated successfully, False otherwise.

Return type:

bool

flip_api.project_services.services.image_service.get_xnat_project_status_info(xnat_project_id: uuid.UUID, db: sqlmodel.Session) → flip_api.domain.schemas.projects.XnatProjectStatusInfo | None

Retrieve the XNAT project status information for a given XNAT project ID.

Parameters:

• xnat_project_id (UUID) – The ID of the XNAT project to retrieve status information for.

• db (Session) – The database session for executing queries.

Returns:

An object containing the XNAT project status information, or None if the

project status could not be found.

Return type:

XnatProjectStatusInfo | None

Raises:

• SQLAlchemyError – If there is an error executing the database query.

• Exception – If there is an unexpected error during the retrieval process.

flip_api.project_services.services.image_service._get_latest_imaging_status(trust_id: uuid.UUID, xnat_project_id: uuid.UUID, db: sqlmodel.Session) → flip_api.domain.interfaces.project.IImagingImportStatus | None

Look up the most recent completed GET_IMAGING_STATUS task result for a trust and XNAT project.

Parameters:

• trust_id (UUID) – The trust to look up.

• xnat_project_id (UUID) – The XNAT project ID to filter by (stored in task payload).

• db (Session) – Database session.

Returns:

Parsed import status counts, or None if no completed result exists.

Return type:

IImagingImportStatus | None

flip_api.project_services.services.image_service.get_imaging_project_statuses(imaging_projects: list[flip_api.domain.schemas.projects.ImagingProject], encoded_query: str, db: sqlmodel.Session) → list[flip_api.domain.interfaces.project.IImagingStatus]

Retrieve the imaging project statuses from local DB and queue status refresh tasks for trusts.

Returns the latest known import status from completed GET_IMAGING_STATUS tasks, and queues new refresh tasks so the next poll will have updated data.

Parameters:

• imaging_projects (list[ImagingProject]) – The list of imaging projects to retrieve statuses for.

• encoded_query (str) – The Base64 URL encoded query to send to the imaging project endpoints.

• db (Session) – The database session for executing queries.

Returns:

A list of IImagingStatus containing the status information for each imaging project.

Return type:

list[IImagingStatus]

flip_api.project_services.services.image_service.update_xnat_user_profile(request_data: flip_api.domain.interfaces.project.IUpdateXnatProfile, db: sqlmodel.Session) → None

Queue user profile update tasks for all trusts’ XNAT instances.

Parameters:

• request_data (IUpdateXnatProfile) – The user profile data to update.

• db (Session) – The database session for retrieving trust information.

Returns:

None

flip_api.project_services.services.image_service.reimport_failed_studies(reimport_queries: list[flip_api.domain.interfaces.project.IReimportQuery], db: sqlmodel.Session, project_reimport_rate_minutes: int) → bool

Queue reimport tasks for failed studies, ensuring that reimports are only attempted if the specified time interval has passed since the last reimport.

Parameters:

• reimport_queries (list[IReimportQuery]) – A list of queries containing information about which projects and trusts to reimport studies for, along with the last reimport time.

• db (Session) – The database session for updating reimport status.

• project_reimport_rate_minutes (int) – The minimum number of minutes that must have passed since the last reimport before attempting another reimport for the same project and trust.

Returns:

True if all eligible reimport tasks were queued successfully.

Return type:

bool