Task
The Task module provides functionality for working with tasks.
- class Task(api, uuid, data={})[source]
Bases:
Base- Parameters:
api (GeoboxClient)
uuid (str)
data (Dict | None)
- BASE_ENDPOINT: str = 'tasks/'
- __init__(api, uuid, data={})[source]
Constructs all the necessary attributes for the Task object.
- Parameters:
api (GeoboxClient) – The API instance.
uuid (str) – The UUID of the task.
data (Dict, optional) – The task data.
- property output_asset: VectorLayer | Raster | Model | File | Tile3d | Table | None
output asset property
- Returns:
if task type is publish, it returns the published layer
- Return type:
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") >>> task.output_asset
- property data: Dict
Returns the task data.
- Returns:
the task data as a dictionary
- Return type:
Dict
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") >>> task.data
- property status: TaskStatus
Returns the status of the task. (auto refresh)
- Returns:
the status of the task(SUCCESS, FAILURE, ABORTED, PENDING, PROGRESS)
- Return type:
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") >>> task.status
- property errors: Dict | None
Get the task errors.
- Returns:
if there are any errors
- Return type:
Dict | None
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") >>> task.errors
- property progress: int | None
Returns the progress of the task.
- Returns:
the progress of the task in percentage or None if the task is not running
- Return type:
int | None
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") >>> task.progress
- wait(timeout=None, interval=1, progress_bar=True, retry=3)[source]
Wait for the task to finish.
- Parameters:
timeout (int, optional) – Maximum time to wait in seconds.
interval (int, optional) – Time between status checks in seconds.
progress_bar (bool, optional) – Whether to show a progress bar. default: True
retry (int, optional) – Number of times to retry if waiting for the task fails. default is 3
- Returns:
the status of the task(SUCCESS, FAILURE, ABORTED, PENDING, PROGRESS)
- Return type:
- Raises:
TimeoutError – If the task doesn’t complete within timeout seconds.
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") >>> task.wait() # return the status of the task
- _check_timeout(start_time, timeout)[source]
Checks if the task has exceeded the timeout period.
- Parameters:
start_time (float)
timeout (int | None)
- Return type:
None
- _is_final_state(status)[source]
Checks if the task has reached a final state.
- Parameters:
status (TaskStatus)
- Return type:
bool
- _update_progress_bar(pbar, last_progress, status=None)[source]
Updates the progress bar with current progress and returns the new last_progress.
- Parameters:
pbar (tqdm | None) – The progress bar to update
last_progress (int) – The last progress value
status (TaskStatus, optional) – The task status. If provided and SUCCESS, updates to 100%
- Returns:
The new last_progress value
- Return type:
int
- abort()[source]
Aborts the task.
- Returns:
None
- Return type:
None
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") >>> task.abort()
- classmethod get_tasks(api, **kwargs)[source]
Get a list of tasks
- Parameters:
api (GeoboxClient) – The GeoboxClient instance for making requests.
- Keyword Arguments:
state (TaskStatus) – Available values : TaskStatus.PENDING, TaskStatus.PROGRESS, TaskStatus.SUCCESS, TaskStatus.FAILURE, TaskStatus.ABORTED
q (str) – query filter based on OGC CQL standard. e.g. “field1 LIKE ‘%GIS%’ AND created_at > ‘2021-01-01’”
search (str) – search term for keyword-based searching among search_fields or all textual fields if search_fields does not have value. NOTE: if q param is defined this param will be ignored.
search_fields (str) – comma separated list of fields for searching.
order_by (str) – comma separated list of fields for sorting results [field1 A|D, field2 A|D, …]. e.g. name A, type D. NOTE: “A” denotes ascending order and “D” denotes descending order.
return_count (bool) – The count of the tasks. default is False.
skip (int) – The skip of the task. default is 0.
limit (int) – The limit of the task. default is 10.
user_id (int) – Specific user. privileges required.
shared (bool) – Whether to return shared tasks. default is False.
- Returns:
The list of task objects or the count of the tasks if return_count is True.
- Return type:
List[Task] | int
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> tasks = Task.get_tasks(client) or >>> tasks = client.get_tasks()
- classmethod get_task(api, uuid)[source]
Gets a task.
- Parameters:
api (GeoboxClient) – The GeoboxClient instance for making requests.
uuid (str) – The UUID of the task.
- Returns:
The task object.
- Return type:
Example
>>> from geobox import GeoboxClient >>> from geobox.task import Task >>> client = GeoboxClient() >>> task = Task.get_task(client, uuid="12345678-1234-5678-1234-567812345678") or >>> task = client.get_task(uuid="12345678-1234-5678-1234-567812345678")