API

pyeodh.set_log_level(level: int)[source]

Set message level for logging.

Parameters:: level (int) – Message level, recommended to use constants provided by logging module.

class pyeodh.client.Client(base_url: str = 'https://test.eodatahub.org.uk', username: str | None = None, token: str | None = None)[source]

Bases: object

get_ades() → Ades[source]

Initializes the workflow execution service (ADES) client.

Calls: GET /ades/ogc-api

Returns:: Object representing the ADES.
Return type:: Ades

get_catalog_service() → CatalogService[source]

Initializes the resource catalog API client.

Calls: GET /api/catalogue/stac

Returns:: Object representing the Resource catalog service.
Return type:: CatalogService

get_wms() → WebMapService_1_1_1 | WebMapService_1_3_0[source]

Initialized the OWSLib WebMapService

Returns:: Initialized WMS
Return type:: WebMapService

get_wmts() → WebMapTileService[source]

Initializes the OWSLib WebMapTileService

Returns:: Initialized WMTS
Return type:: WebMapTileService

class pyeodh.resource_catalog.Catalog(client: Client, headers: Headers, data: Any, **kwargs)[source]

Bases: EodhObject

property collections_href: str: URL pointing to collections endpoint.

create_collection(id: str, description: str, extent: Extent, title: str | None = None, license: str | None = None, keywords: list[str] | None = None, providers: list[Provider] | None = None, summaries: Summaries | None = None, assets: dict[str, Asset] | None = None) → Collection[source]

Create a new collection inside a catalog.

Calls: POST /catalogs/{catalog_id}/collections

Parameters:

id (str)
description (str)
extent (Extent)
title (str | None, optional) – Defaults to None.
license (str | None, optional) – Defaults to None.
keywords (list[str] | None, optional) – Defaults to None.
providers (list[Provider] | None, optional) – Defaults to None.
summaries (Summaries | None, optional) – Defaults to None.
assets (dict[str, Asset] | None, optional) – Defaults to None.

Returns:

An initialized collection object.

Return type:

Collection

delete() → None[source]

Delete this catalog.

Calls: DELETE /catalogs/{catalog_id}

get_catalogs() → list[Catalog][source]

Fetches children catalogs of this parent catalog.

Calls: GET /catalogs/{catalog_path}/catalogs

Returns:: List of children catalogs.
Return type:: list[Catalog]

get_collection(collection_id: str) → Collection[source]

Fetches a resource catalog collection.

Calls: GET /catalogs/{catalog_id}/collections/{collection_id}

Parameters:: collection_id (str) – ID of a collection
Returns:: Collection for given ID
Return type:: Collection

get_collections() → list[Collection][source]

Fetches all resource catalog collections.

Calls: GET /catalogs/{catalog_id}/collections

Returns:: List of available collections
Return type:: list[Collection]

search(limit: int = 10, collections: list[str] | None = None, catalog_paths: list[str] | None = None, ids: list[str] | None = None, bbox: list[Any] | None = None, intersects: dict | None = None, datetime: str | None = None, fields: SearchFields | None = None, query: dict[str, Any] | list[str] | None = None, sort_by: list[SearchSortField] | None = None, filter: dict | None = None, filter_crs: str | None = None, filter_lang: Literal['cql-json', 'cql2-json', 'cql2-text'] | None = None) → PaginatedList[Item][source]

update(description: str | None = None, title: str | None = None) → None[source]

Updates catalog.

Calls: PUT /catalogs/{catalog_id}

Parameters:

description (str | None, optional) – New description. Defaults to None.
title (str | None, optional) – New title. Defaults to None.

class pyeodh.resource_catalog.CatalogService(client: Client, headers: Headers, data: Any, **kwargs)[source]

Bases: Catalog

check_conforms_to(conformance_uri: str | Conformance) → bool[source]

Test if API conforms to a specification.

Parameters:: conformance_uri (str | Conformance) – URI of the specification
Returns:: bool

collection_search(limit: int = 10, bbox: list[Any] | None = None, datetime: str | None = None, query: str | None = None) → PaginatedList[Collection][source]

Searches the catalog for collections.

Parameters:

limit (int, optional) – Number of results per page. Defaults to consts.PAGINATION_LIMIT.
bbox (list[Any] | None, optional) – Bounding box. Defaults to None.
datetime (str | None, optional) – Datetime. Defaults to None.
query (str | None, optional) – Query string. Defaults to None.

Returns:

Iterable list of collections.

Return type:

PaginatedList[Collection]

create_catalog(id: str, description: str, title: str | None = None) → Catalog[source]

Creates a new catalog

Calls: POST /catalogs

Parameters:

id (str) – New catalog ID
description (str) – Catalog description
title (str | None, optional) – Catalog title. Defaults to None.

Returns:

An initialized catalog object.

Return type:

Catalog

discovery_search(query: str, limit: int = 10) → PaginatedList[Catalog][source]

Searches the catalog for catalogs and collections.

Parameters:

limit (int, optional) – Number of results per page. Defaults to consts.PAGINATION_LIMIT.
query (str | None, optional) – Query string. Defaults to None.

Returns:

Iterable list of collections.

Return type:

PaginatedList[Collection]

get_catalog(catalog_id: str) → Catalog[source]

Fetches a catalog.

Calls: GET /catalogs/{catalog_id}

Parameters:: catalog_id (str) – Catalog ID
Returns:: An initialized resource catalog object.
Return type:: Catalog

get_catalogs() → list[Catalog][source]

Fetches all catalogs.

Calls: GET /catalogs

Returns:: List of all catalogs available.
Return type:: list[Catalog]

get_collections() → list[Collection][source]

Fetches all resource catalog collections.

Calls: GET /collections

Returns:: List of available collections
Return type:: list[Collection]

get_conformance() → list[str][source]

Fetches list of standards the API conforms to.

Calls: GET /conformance

Returns:: Standards.
Return type:: list[str]

ping() → str | None[source]

Pings API.

Calls: GET /_mgmt/ping

Returns:: Pong.
Return type:: str | None

class pyeodh.resource_catalog.Collection(client: Client, headers: Headers, data: Any, **kwargs)[source]

Bases: EodhObject

Create an item as part of this collection.

Calls: POST /catalogs/{catalog_id}/collections/{collection_id}/items/{item_id}

Parameters:

id (str) – A unique ID.
geometry (dict[str, Any] | None) – Geometry
bbox (list[float] | None) – Bounding box
datetime (Datetime | None) – Datetime
properties (dict[str, Any] | None) – Properties
collection (str | pystac.Collection | None, optional) – ID of a parent collection. Defaults to None.
assets (dict[str, Any] | None, optional) – Assets. Defaults to None.

Returns:

_description_

Return type:

Item

delete() → None[source]

Delete this collection.

Calls: DELETE /catalogs/{catalog_id}/collections/{collection_id}

get_item(item_id: str) → Item[source]

Fetches a collection item.

Calls: GET /catalogs/{catalog_id}/collections/{collection_id}/items/{item_id}

Parameters:: item_id (str) – ID of a collection item.
Returns:: Initialized item object.
Return type:: Item

get_items() → PaginatedList[Item][source]

Fetches all items within a collection.

Calls: GET /catalogs/{catalog_id}/collections/{collection_id}/items

Returns:

Iterable list of items. Automatically handles: paginated results.

Return type:

PaginatedList[Item]

property items_href: str: URL pointing to items endpoint.

Update collection attributes with new values. Provide only the values you want to update.

Calls: PUT /catalogs/{catalog_id}/collections/{collection_id}

Parameters:

description (str | None, optional) – Description. Defaults to None.
extent (Extent | None, optional) – Extent. Defaults to None.
title (str | None, optional) – Title. Defaults to None.
license (str | None, optional) – License. Defaults to None.
keywords (list[str] | None, optional) – Keywords. Defaults to None.
providers (list[Provider] | None, optional) – Providers. Defaults to None.
summaries (Summaries | None, optional) – Summaries. Defaults to None.
assets (dict[str, Asset] | None, optional) – Assets. Defaults to None.

class pyeodh.resource_catalog.Conformance(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

TRANSACTION_EXTENSION = 'https://api.stacspec.org/v1.0.0/ogcapi-features/extensions/transaction'

class pyeodh.resource_catalog.Item(client: Client, headers: Headers, data: Any, **kwargs)[source]

Bases: EodhObject

delete() → None[source]

Delete this item.

Calls: DELETE /catalogs/{catalog_id}/collections/{collection_id}/items/{item_id}

Update this item’s attributes with new values. Only provide the values you want to update.

Calls: PUT /catalogs/{catalog_id}/collections/{collection_id}/items/{item_id}

Parameters:

geometry (dict[str, Any] | None, optional) – Geometry. Defaults to None.
bbox (list[float] | None, optional) – Bounding box. Defaults to None.
datetime (Datetime | None, optional) – Datetime. Defaults to None.
properties (dict[str, Any] | None, optional) – Properties. Defaults to None.
collection (str | pystac.Collection | None, optional) – Collection. Defaults to None.
assets (dict[str, Any] | None, optional) – Assets. Defaults to None.

class pyeodh.ades.AdditionalParameters(title: str | None, role: str | None, href: str | None, parameters: list[Parameter])[source]

Bases: object

Represents additional parameters of the process.

classmethod from_dict(data: dict)[source]

href: str | None

parameters: list[Parameter]

role: str | None

title: str | None

class pyeodh.ades.Ades(client: Client, headers: Headers, data: Any)[source]

Bases: EodhObject

Represents ADES API service.

Parameters:

client (Client) – Instance of pyeodh request client
headers (Headers) – Response headers received when requesting this record
data (Any) – Raw response data received when requesting this record

deploy_process(cwl_url: str | None = None, cwl_yaml: str | None = None) → Process[source]

Deploy a process.

Calls: POST /processes

Parameters:

cwl_url (str | None, optional) – Location of the cwl workflow file. Mutually exclusive with cwl_yaml arg. Defaults to None.
cwl_yaml (str | None, optional) – CWL workflow in yaml format. Mutually exclusive with cwl_url. Defaults to None.

Returns:

The newly created process as returned by the API.

Return type:

Process

get_job(job_id) → Job[source]

Fetches an individual job.

Parameters:: job_id (_type_) – Job ID.
Returns:: Initialized job object.
Return type:: Job

get_jobs() → list[Job][source]

Fetches a list of jobs triggered by the user.

Calls: GET /jobs

Returns:: List of user’s jobs.
Return type:: list[Job]

get_process(process_id: str) → Process[source]

Fetch an individual process

Calls: GET /processes/{process_id}

Parameters:: process_id (str) – Process ID
Returns:: Initialized process object.
Return type:: Process

get_processes() → list[Process][source]

Fetches available processes

Calls: GET /processes

Returns:: List of available processes.
Return type:: list[Process]

property jobs_href: str: URL pointing to jobs endpoint.

property processes_href: str: URL pointing to processes endpoint.

property self_href: str: URL of this record.

property title: str

class pyeodh.ades.AdesJobStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

ACCEPTED = 'accepted'

DISMISSED = 'dismissed'

FAILED = 'failed'

RUNNING = 'running'

SUCCESSFUL = 'successful'

class pyeodh.ades.AdesRelType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

EXECUTE = 'http://www.opengis.net/def/rel/ogc/1.0/execute'

JOBS = 'http://www.opengis.net/def/rel/ogc/1.0/job-list'

PROCESSES = 'http://www.opengis.net/def/rel/ogc/1.0/processes'

RESULTS = 'http://www.opengis.net/def/rel/ogc/1.0/results'

RESULTS_NOT_READY = 'http://www.opengis.net/def/rel/ogc/1.0/exception/result-not-ready'

SELF = 'self'

STATUS = 'monitor'

class pyeodh.ades.Job(client: Client, headers: Headers, data: Any)[source]

Bases: EodhObject

Represents ADES Job record.

Parameters:

client (Client) – Instance of pyeodh request client
headers (Headers) – Response headers received when requesting this record
data (Any) – Raw response data received when requesting this record

delete() → None[source]: Stop and delete this job.

get_result_items() → list[Item][source]

Get job outputs

Returns:: List of items generated by job
Return type:: list[Item]

refresh() → None[source]: Refresh this object with the latest data from the API.

property self_href: str: URL of this object.

class pyeodh.ades.Metadata(title: str | None, role: str | None, href: str | None)[source]

Bases: object

Represents process metadata object.

classmethod from_dict(data: dict[str, str])[source]

href: str | None

role: str | None

title: str | None

class pyeodh.ades.Parameter(name: str, value: list[Any])[source]

Bases: object

Represents a single process parameter.

name: str

value: list[Any]

class pyeodh.ades.Process(client: Client, headers: Headers, data: Any, parent_url: str)[source]

Bases: EodhObject

Represents ADES Process record.

Parameters:

client (Client) – Instance of pyeodh request client
headers (Headers) – Response headers received when requesting this record
data (Any) – Raw response data received when requesting this record
parent_url (str) – URL of the parent API endpoint, usually /processes

delete() → None[source]: Delete this record.

execute(inputs: dict, workspace: str | None = None) → Job[source]

Trigger process workflow execution.

Calls: POST /processes/{process_id}/execution

Parameters:: inputs (dict) – A dictionary containing inputs structured according to schema defined by the process.
Returns:: Object representing the triggered job.
Return type:: Job

property execute_href: str: URL pointing to processes execution endpoint.

update(cwl_url: str | None = None, cwl_yaml: str | None = None) → None[source]

Update the process workflow. cwl_url and cwl_yaml parameters are mutually exclusive.

Calls: PUT /processes/{process_id}

Parameters:

cwl_url (str | None, optional) – URL pointing to the workflow CWL file. Defaults to None.
cwl_yaml (str | None, optional) – Content of the CWL file in yaml format. Defaults to None.

exception pyeodh.ades.ResultsNotReadyError[source]

Bases: Exception

Exception raised when job results are not ready.