"""
VSO API Client: Client for the Volcano Space Observatory API
Copyright (C) CNES, CNRS, Université de Lille, Région Hauts-de-France
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as
published by the Free Software Foundation, either version 3 of the
License, or (at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.
You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
import datetime as dt
import logging
import urllib.parse
from os import PathLike
from typing import Any, Literal
from uuid import UUID
import orjson
import requests
from pydantic import validate_call
from .auth import DataTerraAuthClient, get_bearer_auth
from .utils import get_json_payload, raise_for_status
[docs]
class Client:
@validate_call(config=dict(arbitrary_types_allowed=True))
def __init__(
self,
api_url: str = "https://ap.icare.univ-lille.fr/vso/v1",
auth_client: DataTerraAuthClient = DataTerraAuthClient(
sso_url="https://sso.earth-data.fr/",
client_id="aeris-icare-vso",
realm="gaia-data",
),
):
"""Initialize the VSO API client.
:param api_url: Base URL for the VSO API endpoint.
:param sso_url: Base URL for the SSO authentication server.
:param sso_client: Client identifier for SSO authentication.
:param sso_realm: Realm name for SSO authentication.
:param logger: Logger instance for debugging and error reporting.
"""
self.logger = logging.getLogger(__name__)
self.api_url = api_url
self._auth_client = auth_client
@property
def api_url(self):
return self._api_url
@api_url.setter
def api_url(self, value: str):
self._api_url = value.rstrip("/")
[docs]
def get_workflows(self) -> dict:
"""
Retrieve a list of all supported workflows.
:return: Dictionary containing list of workflows with their names and descriptions.
"""
path = self._get_workflows_path()
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_workflow(self, workflow: str) -> dict:
"""
Retrieve detailed information about a specific workflow including parameter schema and examples.
:param workflow: The workflow type to get information for (e.g., 'sentinel1-insar').
:return: Dictionary containing detailed workflow information including schema and examples.
"""
path = self._get_workflow_info_path(workflow)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_workflow_parameters_schema(self, workflow: str) -> dict:
"""
Retrieve JSON schema for a specific workflow's parameters.
:param workflow: The workflow type to get schema for (e.g., 'sentinel1-insar').
:return: Dictionary containing the JSON schema for workflow parameters.
"""
path = self._get_workflow_schema_path(workflow)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_workflow_examples(self, workflow: str) -> dict[str, dict]:
"""
Retrieve example parameters for a specific workflow type from the API.
:param workflow: The workflow type to get examples for (e.g., 'sentinel1-insar').
:return: Dictionary containing example parameter sets for the workflow.
"""
path = self._get_workflow_examples_path(workflow)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_jobs(
self,
workflow: str | None = None,
completed: bool | None = None,
limit: int = 100,
offset: int = 0,
) -> list[dict]:
"""
Retrieve a list of jobs with optional filtering by workflow type and completion status.
:param workflow: Optional workflow type to filter jobs (e.g., 'sentinel1-insar').
:param completed: If True, return only completed jobs; if False, return only pending jobs. All jobs are returned if None.
:param limit: Maximum number of jobs to retrieve in paginated result.
:param offset: Offset to determine the number of jobs to skip in paginated result.
:return: List of job dictionaries containing job details.
"""
path = self._get_jobs_path()
params = dict(
workflow=workflow, completed=completed, limit=limit, offset=offset
)
url = self._get_url(path)
return self._get_request_response(url, params=params)
[docs]
def get_job(self, job_id: UUID) -> dict:
"""
Retrieve detailed information for a specific job.
:param job_id: Unique identifier of the job to retrieve.
:return: Dictionary containing comprehensive job details including status, parameters, and metadata.
"""
path = self._get_job_path(job_id)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_job_status(self, job_id: UUID) -> dict:
"""
Get the current status of a specific job.
:param job_id: Unique identifier of the job to check status for.
:return: Dictionary containing job status information (e.g., 'pending', 'running', 'completed', 'failed').
"""
path = self._get_job_status_path(job_id)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_has_job_ended(self, job_id: UUID) -> dict:
"""
Check if a job has completed (successfully or failed).
:param job_id: Unique identifier of the job to check completion status for.
:return: Dictionary indicating whether the job has ended and its completion status.
"""
path = self._get_job_ended_path(job_id)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_job_progress(self, job_id: UUID) -> dict:
"""
Get the current progress of a running job.
:param job_id: Unique identifier of the job to check progress for.
:return: Dictionary containing progress information (e.g., percentage completed, current step).
"""
path = self._get_job_progress_path(job_id)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def get_job_logs(
self,
job_id: UUID,
level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
after: dt.datetime | None = None,
before: dt.datetime | None = None,
) -> dict:
"""
Retrieve log entries for a specific job with optional filtering.
:param job_id: Unique identifier of the job to retrieve logs for.
:param level: Minimum log level to retrieve (e.g., 'DEBUG', 'INFO', 'WARNING', 'ERROR').
:param after: Optional datetime filter to get logs after this timestamp.
:param before: Optional datetime filter to get logs before this timestamp.
:return: List of log entry dictionaries with timestamp, level, and message.
"""
path = self._get_job_logs_path(job_id)
url = self._get_url(path)
params = dict(level=level, after=after, before=before)
return self._get_request_response(url, params=params)
[docs]
def get_job_result(self, job_id: UUID) -> dict:
"""
Download the result files for a successfully completed job.
:param job_id: Unique identifier of the completed job to download results from.
:return: Dictionary containing download information or file content for the job results.
"""
path = self._get_job_result_path(job_id)
url = self._get_url(path)
return self._get_request_response(url)
[docs]
def submit_job(
self,
workflow: str,
job_parameters: dict | str | PathLike,
access_token: str | None = None,
) -> dict:
"""
Submit a new job for processing with the specified workflow and parameters.
:param workflow: The type of workflow to execute (e.g., 'sentinel1-insar').
:param job_parameters: Dictionary or json file containing all required parameters for the selected workflow.
:param access_token: Optional authentication token. If not provided, client will
try to obtain one or prompt the user to login.
:return: Dictionary containing submitted job details including job ID and initial status.
"""
if access_token is None:
access_token = self._auth_client.get_access_token()
job_parameters = self._get_job_parameters(job_parameters)
path = self._get_submitjob_path(workflow)
url = self._get_url(path)
return self._get_request_response(
url, body=job_parameters, method="POST", access_token=access_token
)
@staticmethod
def _get_job_parameters(job_parameters: dict | str | PathLike) -> Any:
if isinstance(job_parameters, dict):
return job_parameters
with open(job_parameters, "r") as f:
try:
return orjson.loads(f.read())
except orjson.JSONDecodeError as e:
raise ValueError(f"Invalid JSON format in file: {e}")
[docs]
def cancel_job(self, job_id: UUID, access_token: str | None = None) -> dict:
"""
Cancel a running or pending job.
:param job_id: Unique identifier of the job to cancel.
:param access_token: Optional authentication token. If not provided, client will
try to obtain one or prompt the user to login.
:return: Dictionary confirming the cancellation status and job details.
"""
if access_token is None:
access_token = self._auth_client.get_access_token()
path = self._get_cancel_job_path(job_id)
url = self._get_url(path)
return self._get_request_response(
url, method="DELETE", access_token=access_token
)
def _get_url(self, path):
protocol, netloc, base_path = urllib.parse.urlparse(self.api_url)[:3]
url = urllib.parse.urlunparse((protocol, netloc, base_path + path, "", "", ""))
return url
@staticmethod
def _get_openapi_path() -> str:
path = "/openapi.json"
return path
@staticmethod
def _get_jobs_path():
return "/jobs"
@staticmethod
def _get_job_path(job_id):
return f"/jobs/{job_id}"
@staticmethod
def _get_job_status_path(job_id):
return f"/jobs/{job_id}/status"
@staticmethod
def _get_job_ended_path(job_id):
return f"/jobs/{job_id}/ended"
@staticmethod
def _get_job_progress_path(job_id):
return f"/jobs/{job_id}/progress"
@staticmethod
def _get_job_logs_path(job_id):
return f"/jobs/{job_id}/logs"
@staticmethod
def _get_job_result_path(job_id):
return f"/jobs/{job_id}/result"
@staticmethod
def _get_cancel_job_path(job_id):
return f"/jobs/{job_id}"
@staticmethod
def _get_workflows_path() -> str:
return "/workflows"
@staticmethod
def _get_workflow_info_path(workflow: str) -> str:
return f"/workflows/{workflow}"
@staticmethod
def _get_workflow_schema_path(workflow: str) -> str:
return f"/workflows/{workflow}/schema"
@staticmethod
def _get_workflow_examples_path(workflow: str) -> str:
return f"/workflows/{workflow}/examples"
@staticmethod
def _get_submitjob_path(workflow):
return f"/jobs/{workflow}"
def _get_request_response(
self, url, params=None, body=None, method="GET", access_token=None
):
self.logger.debug(f"{method} request to {url}")
auth = get_bearer_auth(access_token) if access_token is not None else None
match method:
case "GET":
response = requests.get(url, params=params, auth=auth)
case "POST":
response = requests.post(url, json=body, auth=auth)
case "DELETE":
response = requests.delete(url, auth=auth)
case _:
raise ValueError(f"Unsupported method: {method}")
raise_for_status(response)
return get_json_payload(response)