Future

def cancel(self):
    """Attempt to cancel the execution of the task.

    Returns:
        The backend cancellation response when cancellation is submitted;
        otherwise None if cancellation raises and a warning is emitted.
    """

    self.authenticate()
    with TasksClient(self.app_context) as client:
        try:
            # NOTE: typing issue because client is seen as BaseClient instead of TaskClient
            return client.cancel(id=self.task_id)  # type: ignore
        except Exception as e:
            warn(
                f"Exception encountered when trying to cancel task with ID {self.task_id}: {str(repr(e))}"
            )

def cancelled(self) -> bool:
    """Return whether the task was cancelled.

    Returns:
        bool: `True` if the task status is `CANCELLED`.
    """
    return self.status() == TaskStatus.CANCELLED

def done(self) -> bool:
    """Return whether the task has reached a terminal status.

    Returns:
        bool: `True` if the task status is one of `EXIT_STATUS`
            (completed, cancelled, failed, or payload processing error).
    """
    return self.status() in self.EXIT_STATUS

def export_to(
    self,
    storage: StorageBackend,
    chunk_size: int = 1000,
    shot_filter: ShotFilter | None = None,
):
    """Copy stored shots and task definitions to another storage backend.

    Args:
        storage (StorageBackend): Destination storage backend.
        chunk_size (int): Maximum number of shots to write per batch.
            Defaults to 1000.
        shot_filter (ShotFilter | None): Optional shot filter. When the
            filter includes `task_ids`, only those task definitions are
            copied. Otherwise, all task definitions from this future's
            storage are copied. Defaults to None.
    """
    chunk = []
    for shot in self.storage.get_shots(shot_filter=shot_filter):
        chunk.append(shot)
        if len(chunk) >= chunk_size:
            storage.add_shots(chunk)
            chunk = []
    if chunk:
        storage.add_shots(chunk)

    if shot_filter is not None and shot_filter.task_ids is not None:
        task_ids = shot_filter.task_ids
    else:
        task_ids = self.storage.task_ids()

    for task_id in task_ids:
        task_def = self.storage.get_task_definition(task_id)
        creation_time = self.storage.get_task_creation_time(task_id)
        storage.add_task_definition(task_id, task_def, creation_time)

def fetch(self) -> None:
    """Fetch currently available shot results into this future's storage.

    Results are requested from the first known incomplete subtask page and
    paginated by both subtask and shot page. Repeated calls are safe because
    storage backends de-duplicate rows by task ID, shot index, and frame
    type.
    """
    self.authenticate()
    subtask_page = self._first_incomplete_subtask_page
    done = False

    with ResultsClient(self.app_context) as client:
        while not done:
            done = self._fetch_subtask_page(
                client=client,  # type: ignore
                subtask_page=subtask_page,
            )

            subtask_page += 1

def fetch_and_export_to(self, storage: StorageBackend, chunk_size: int = 1000):
    """Fetch available results and export this future's storage.

    Args:
        storage (StorageBackend): Destination storage backend.
        chunk_size (int): Maximum number of shots to write per batch.
            Defaults to 1000.
    """
    self.fetch()
    self.export_to(storage, chunk_size=chunk_size)

@classmethod
def from_storage(
    cls,
    *,
    storage: StorageBackend,
    new_storage: StorageBackend | None = None,
    task_id: str | None = None,
    fetch_options: ApiFetchOptions = ApiFetchOptions(),
    context_name: str | None = None,
) -> Self:
    """Create a future from task metadata already present in storage.

    Args:
        storage (StorageBackend): Storage used to discover and validate the
            task ID.
        new_storage (StorageBackend | None): Storage backend to attach to
            the returned future. When None, `storage` is reused. Defaults to
            None.
        task_id (str | None): Task ID to attach to. Required when `storage`
            contains more than one task ID. Defaults to None.
        fetch_options (ApiFetchOptions): Pagination and polling options.
            Defaults to `ApiFetchOptions()`.
        context_name (str | None): Name of the qlam context to attach to
            the returned future. When None, the class-level default on
            `cls` is used. Defaults to None.

    Returns:
        Self: A future attached to the selected task ID.

    Raises:
        ValueError: If storage has no task IDs, the requested task ID is not
            present, multiple task IDs are present without an explicit
            `task_id`, or `context_name` is None and `cls` has no
            class-level default.
    """
    context_name = cls._resolve_context_name(context_name)

    if new_storage is None:
        new_storage = storage

    task_ids_at_storage = storage.task_ids()

    if not task_ids_at_storage:
        raise ValueError("Found no task IDs in storage.")

    if task_id is not None and task_id not in task_ids_at_storage:
        raise ValueError(f"Task with ID {task_id} not found at storage {storage}")

    if len(task_ids_at_storage) > 1 and task_id is None:
        msg = "More than one task ID found! Please specify a task_id. Candidates are:\n"
        for candidate_task_id in task_ids_at_storage:
            creation_time = storage.get_task_creation_time(candidate_task_id)
            msg += f"  * {candidate_task_id}, created at {creation_time}\n"
        raise ValueError(msg)

    if task_id is None:
        task_id = task_ids_at_storage.pop()

    return cls(
        task_id=task_id,
        storage=new_storage,
        fetch_options=fetch_options,
        result_cls=cls.result_cls,
        context_name=context_name,
    )

@classmethod
def from_task_id(
    cls,
    *,
    task_id: str,
    storage: StorageBackend | None = None,
    fetch_options: ApiFetchOptions = ApiFetchOptions(),
    context_name: str | None = None,
) -> Self:
    """Create a future from a backend task ID.

    The task record and task definition are fetched from the backend, and
    the task definition is stored in `storage` before the future is
    returned.

    Args:
        task_id (str): Backend task ID.
        storage (StorageBackend | None): Storage backend that will receive
            the task definition and later fetched shots. When None, a fresh
            `DictStorage` is used (in-memory; not persisted across
            processes). Defaults to None.
        fetch_options (ApiFetchOptions): Pagination and polling options.
            Defaults to `ApiFetchOptions()`.
        context_name (str | None): Name of the qlam context used to fetch
            the task and attached to the returned future. When None, the
            class-level default on `cls` is used. Defaults to None.

    Returns:
        Self: A future attached to `task_id`.

    Raises:
        ValueError: If `context_name` is None and `cls` has no
            class-level default.
    """
    if storage is None:
        storage = DictStorage()

    context_name = cls._resolve_context_name(context_name)
    auth = AuthMixin(context_name=context_name)
    auth.authenticate()
    with TasksClient(auth.app_context) as client:
        task = client.get(id=task_id)  # type: ignore

    # fetch subtasks for metadata
    with DefinitionsClient(auth.app_context) as client:
        task_def = client.get(id=task.definition_id)  # type: ignore

    storage.add_task_definition(
        task_id, task_definition=task_def, creation_time=task.created_date
    )

    return cls(
        task_id=task_id,
        storage=storage,
        fetch_options=fetch_options,
        result_cls=cls.result_cls,
        context_name=context_name,
    )

def get_compilation(self, compilation_id: str | None = None):
    """Fetch the compilation record associated with this task.

    Args:
        compilation_id (str | None): ID of the compilation to fetch. When
            `None`, the compilation ID is retrieved from the task record.
            Defaults to `None`.

    Returns:
        The compilation object returned by the backend.
    """
    self.authenticate()

    if compilation_id is None:
        compilation_id = self.get_task().compilation_id

    with CompilationsClient(self.app_context) as client:
        return client.get(id=compilation_id)  # type: ignore

def get_task(self) -> "Task":
    """Fetch the current task record from the backend.

    Returns:
        Task: The task object returned by the backend.
    """
    self.authenticate()
    with TasksClient(self.app_context) as client:
        # NOTE: typing issue in qlam-core
        # every client is BaseRestApi, which doesn't have get, but it actually does
        task = client.get(id=self.task_id)  # type: ignore
        logger.info(
            f"Fetched task with id {self.task_id}. Current status: {task.task_status}"
        )
        return task

def partial_result(self) -> ResultType:
    """Fetch currently available results and return a result view.

    Unlike `result`, this method does not wait for the task to finish.

    Returns:
        ResultType: A result view scoped to this future's task ID and the
            DETECTED frame type.
    """
    self.fetch()
    return self.results_from_storage()

def result(
    self,
    *,
    timeout: float | None = None,
) -> ResultType:
    """Wait for completion, fetch results, and return a result view.

    Keyword Args:
        timeout (float | None): Maximum number of seconds to wait for a
            terminal task status. If None, wait indefinitely. Defaults to
            None.

    Returns:
        ResultType: A result view scoped to this future's task ID and the
            DETECTED frame type.

    Raises:
        TimeoutError: If `timeout` elapses before a terminal status is
            reached.
        ValueError: If the task was cancelled or failed.
    """
    self._wait_for_completion(timeout=timeout)
    self.fetch()

    return self.results_from_storage()

def results_from_storage(self, shot_filter: ShotFilter | None = None) -> ResultType:
    """Build a result view over this future's storage.

    Args:
        shot_filter (ShotFilter | None): Filter to apply to the result view.
            When None, the view is scoped to this future's task ID and the
            DETECTED frame type. Defaults to None.

    Returns:
        ResultType: A result view over the storage backend.
    """
    if shot_filter is None:
        shot_filter = ShotFilter(task_ids=(self.task_id,), frame_type="DETECTED")
    return self.result_cls(
        storage=self.storage,
        shot_filter=shot_filter,
    )

def status(self) -> TaskStatus:
    """Return the current status of the task.

    Returns:
        TaskStatus: The task status as reported by the backend.
    """
    return self.get_task().task_status