flip_api.project_services.services.image_service ================================================ .. py:module:: flip_api.project_services.services.image_service Functions --------- .. autoapisummary:: flip_api.project_services.services.image_service.has_pending_imaging_tasks flip_api.project_services.services.image_service.to_utc_aware flip_api.project_services.services.image_service.base64_url_encode flip_api.project_services.services.image_service.get_imaging_projects flip_api.project_services.services.image_service.delete_imaging_project flip_api.project_services.services.image_service.get_xnat_project_status_info flip_api.project_services.services.image_service._get_latest_imaging_status flip_api.project_services.services.image_service.get_imaging_project_statuses flip_api.project_services.services.image_service.update_xnat_user_profile flip_api.project_services.services.image_service.reimport_failed_studies Module Contents --------------- .. py:function:: 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. :param project_id: The project to check. :param db: Database session. :returns: True if at least one CREATE_IMAGING task for this project is PENDING or IN_PROGRESS. .. py:function:: 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. :param dt: The datetime to normalise. Naive datetimes are assumed to be in UTC. :type dt: datetime | None :returns: A timezone-aware UTC datetime. Returns ``datetime.min`` with UTC tzinfo when ``dt`` is None. :rtype: datetime .. py:function:: base64_url_encode(data: str) -> str Encode a string using Base64 URL encoding without padding. :param data: The string to encode. :type data: str :returns: The Base64 URL-encoded string with trailing ``=`` padding stripped. :rtype: str .. py:function:: 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. :param project_id: The ID of the project to retrieve imaging projects for. :type project_id: UUID :param db: The database session for executing queries. :type db: Session :returns: A list of ImagingProject objects associated with the given project ID. :rtype: list[ImagingProject] :raises SQLAlchemyError: If there is an error executing the database query. :raises Exception: If there is an unexpected error during the retrieval process. .. py:function:: 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. :param imaging_project: The imaging project to delete. :type imaging_project: ImagingProject :param db: The database session for executing queries. :type db: Session :returns: True if the task was queued and status updated successfully, False otherwise. :rtype: bool .. py:function:: 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. :param xnat_project_id: The ID of the XNAT project to retrieve status information for. :type xnat_project_id: UUID :param db: The database session for executing queries. :type db: Session :returns: An object containing the XNAT project status information, or None if the project status could not be found. :rtype: XnatProjectStatusInfo | None :raises SQLAlchemyError: If there is an error executing the database query. :raises Exception: If there is an unexpected error during the retrieval process. .. py:function:: _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. :param trust_id: The trust to look up. :type trust_id: UUID :param xnat_project_id: The XNAT project ID to filter by (stored in task payload). :type xnat_project_id: UUID :param db: Database session. :type db: Session :returns: Parsed import status counts, or None if no completed result exists. :rtype: IImagingImportStatus | None .. py:function:: 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. :param imaging_projects: The list of imaging projects to retrieve statuses for. :type imaging_projects: list[ImagingProject] :param encoded_query: The Base64 URL encoded query to send to the imaging project endpoints. :type encoded_query: str :param db: The database session for executing queries. :type db: Session :returns: A list of IImagingStatus containing the status information for each imaging project. :rtype: list[IImagingStatus] .. py:function:: 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. :param request_data: The user profile data to update. :type request_data: IUpdateXnatProfile :param db: The database session for retrieving trust information. :type db: Session :returns: None .. py:function:: 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. :param reimport_queries: A list of queries containing information about which projects and trusts to reimport studies for, along with the last reimport time. :type reimport_queries: list[IReimportQuery] :param db: The database session for updating reimport status. :type db: Session :param project_reimport_rate_minutes: The minimum number of minutes that must have passed since the last reimport before attempting another reimport for the same project and trust. :type project_reimport_rate_minutes: int :returns: True if all eligible reimport tasks were queued successfully. :rtype: bool