Skip to content

# Cuiman API Reference

The Cuiman Python API is provided by the cuiman package.

The Client class provides a synchronous API. If you want an asynchronous version, use the AsyncClient class instead. It provides the same interface, but using asynchronous server calls.

Both clients return their configuration as a ClientConfig object.

Methods of the Client and AsyncClient may raise a ClientError if a server call fails.

cuiman.Client

Bases: ClientMixin

The client API for the web service (synchronous mode).

Parameters:

Name Type Description Default
config Optional[ClientConfig]

Optional client configuration object. If given, other configuration arguments are ignored.

 None
config_path Optional[str]

Optional path of the configuration file to be loaded

 None
api_url Optional[str]

The service URL of the OGC API - Processes.

 None
config_kwargs Any

Configuration settings as keyword arguments.

 {}
Source code in cuiman\src\cuiman\api\client.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
class Client(ClientMixin):
    """
    The client API for the web service (synchronous mode).

    Args:
      config: Optional client configuration object. If given,
        other configuration arguments are ignored.
      config_path: Optional path of the configuration file to be loaded
      api_url: The service URL of the OGC API - Processes.
      config_kwargs: Configuration settings as keyword arguments.
    """

    def __init__(
        self,
        *,
        config: Optional[ClientConfig] = None,
        config_path: Optional[str] = None,
        api_url: Optional[str] = None,
        _debug: bool = False,
        _transport: Optional[Transport] = None,
        **config_kwargs: Any,
    ):
        self._config = ClientConfig.create(
            config=config,
            config_path=config_path,
            api_url=api_url,
            **config_kwargs,
        )
        if not self._config.api_url:
            raise ValueError("Required setting 'api_url' not configured")
        self._transport = (
            HttpxTransport(
                api_url=self._config.api_url,
                headers=self._config.auth_headers,
                return_type_map=self._config.return_type_map,
                debug=_debug,
            )
            if _transport is None
            else _transport
        )

    @property
    def config(self) -> ClientConfig:
        return self._config

    def _repr_json_(self):
        # noinspection PyProtectedMember
        return self.config._repr_json_()

    def get_capabilities(self, **kwargs: Any) -> Capabilities:
        """
        The landing page provides links to the:
          * The OpenAPI-definition (no fixed path),
          * The Conformance statements (path /conformance),
          * The processes metadata (path /processes),
          * The endpoint for job monitoring (path /jobs).

        For more information, see [OGC API — Processes — Part 1
        Section
        7.2](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_landing_page).

        Returns:
          Capabilities: The landing page provides links to the API definition
            (link relations `service-desc` and `service-doc`),
            the Conformance declaration (path `/conformance`,
            link relation `http://www.opengis.net/def/rel/ogc/1.0/conformance`),
            and to other resources.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `500`: A server error occurred.
        """
        return self._transport.call(
            TransportArgs(
                path="/",
                method="get",
                return_types={"200": Capabilities},
                error_types={"500": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def get_conformance(self, **kwargs: Any) -> ConformanceDeclaration:
        """
        A list of all conformance classes, specified in a standard, that the
        server conforms to.

        | Conformance class | URI |
        |-----------|-------|
        |Core|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/core|
        |OGC Process Description|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/ogc-process-description|
        |JSON|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/json|
        |HTML|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/html|
        |OpenAPI Specification 3.0|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/oas30|
        |Job list|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/job-list|
        |Callback|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/callback|
        |Dismiss|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/dismiss|

        For more information, see [OGC API — Processes — Part 1
        Section 7.4](https://docs.ogc.org/is/18-062r2/18-
        062r2.html#sc_conformance_classes).


        Returns:
          ConformanceDeclaration: The URIs of all conformance classes supported
            by the server. To support "generic" clients that want
            to access multiple OGC API - Processes implementations - and
            not "just" a specific API / server, the server declares
            the conformance classes it implements and conforms to.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `500`: A server error occurred.
        """
        return self._transport.call(
            TransportArgs(
                path="/conformance",
                method="get",
                return_types={"200": ConformanceDeclaration},
                error_types={"500": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def get_processes(self, **kwargs: Any) -> ProcessList:
        """
        The list of processes contains a summary of each process the OGC API -
        Processes offers, including the link to a more detailed description of
        the process.

        For more information, see [OGC API — Processes — Part 1
        Section
        7.9](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_process_list).


        Returns:
          ProcessList: Information about the available processes

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.
        """
        return self._transport.call(
            TransportArgs(
                path="/processes",
                method="get",
                return_types={"200": ProcessList},
                extra_kwargs=kwargs,
            )
        )

    def get_process(self, process_id: str, **kwargs: Any) -> ProcessDescription:
        """
        The process description contains information about inputs and outputs
        and a link to the execution-endpoint for the process. The Core does not
        mandate the use of a specific process description to specify the
        interface of a process. That said, the Core requirements class makes the
        following recommendation:

        Implementations **should** consider supporting the OGC process
        description.

        For more information, see [OGC API — Processes — Part 1
        Section 7.10](https://docs.ogc.org/is/18-062r2/18-
        062r2.html#sc_process_description).

        Args:
          process_id:
          kwargs: Optional keyword arguments that may be
            used by the underlying transport.

        Returns:
          ProcessDescription: A process description.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `404`: The requested URI was not found.
        """
        return self._transport.call(
            TransportArgs(
                path="/processes/{processID}",
                method="get",
                path_params={"processID": process_id},
                return_types={"200": ProcessDescription},
                error_types={"404": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def execute_process(
        self, process_id: str, request: ProcessRequest, **kwargs: Any
    ) -> JobInfo:
        """
        Create a new job.

        For more information, see [OGC API — Processes — Part 1
        Section
        7.11](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_create_job).

        Args:
          process_id:
          kwargs: Optional keyword arguments that may be
            used by the underlying transport.
          request: Mandatory request JSON

        Returns:
          JobInfo: Started asynchronous execution. Created job.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `404`: The requested URI was not found.
            - `500`: A server error occurred.
        """
        return self._transport.call(
            TransportArgs(
                path="/processes/{processID}/execution",
                method="post",
                path_params={"processID": process_id},
                request=request,
                return_types={"201": JobInfo},
                error_types={"404": ApiError, "500": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def get_jobs(self, **kwargs: Any) -> JobList:
        """
        List available jobs.

        For more information, see [OGC API — Processes — Part 1
        Section 11](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_job_list).


        Returns:
          JobList: A list of jobs for this process.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `404`: The requested URI was not found.
        """
        return self._transport.call(
            TransportArgs(
                path="/jobs",
                method="get",
                return_types={"200": JobList},
                error_types={"404": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def get_job(self, job_id: str, **kwargs: Any) -> JobInfo:
        """
        Show the status of a job.

        For more information, see [OGC API — Processes — Part 1
        Section 7.12](https://docs.ogc.org/is/18-062r2/18-
        062r2.html#sc_retrieve_status_info).

        Args:
          job_id: Local identifier of a job
          kwargs: Optional keyword arguments that may be
            used by the underlying transport.

        Returns:
          JobInfo: The status of a job.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `404`: The requested URI was not found.
            - `500`: A server error occurred.
        """
        return self._transport.call(
            TransportArgs(
                path="/jobs/{jobId}",
                method="get",
                path_params={"jobId": job_id},
                return_types={"200": JobInfo},
                error_types={"404": ApiError, "500": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def dismiss_job(self, job_id: str, **kwargs: Any) -> JobInfo:
        """
        Cancel a job execution and removes it from the jobs list.

        For more information, see [OGC API — Processes — Part 1
        Section 13](https://docs.ogc.org/is/18-062r2/18-062r2.html#Dismiss).

        Args:
          job_id: Local identifier of a job
          kwargs: Optional keyword arguments that may be
            used by the underlying transport.

        Returns:
          JobInfo: Information about the job.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `404`: The requested URI was not found.
            - `500`: A server error occurred.
        """
        return self._transport.call(
            TransportArgs(
                path="/jobs/{jobId}",
                method="delete",
                path_params={"jobId": job_id},
                return_types={"200": JobInfo},
                error_types={"404": ApiError, "500": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def get_job_results(self, job_id: str, **kwargs: Any) -> JobResults:
        """
        List available results of a job. In case of a failure, list errors
        instead.

        For more information, see [OGC API — Processes — Part 1
        Section 7.13](https://docs.ogc.org/is/18-062r2/18-
        062r2.html#sc_retrieve_job_results).

        Args:
          job_id: Local identifier of a job
          kwargs: Optional keyword arguments that may be
            used by the underlying transport.

        Returns:
          JobResults: The results of a job.

        Raises:
          ClientError: If the call to the web service fails
            with a status code != `2xx`.

            - `404`: The requested URI was not found.
            - `500`: A server error occurred.
        """
        return self._transport.call(
            TransportArgs(
                path="/jobs/{jobId}/results",
                method="get",
                path_params={"jobId": job_id},
                return_types={"200": JobResults},
                error_types={"404": ApiError, "500": ApiError},
                extra_kwargs=kwargs,
            )
        )

    def close(self):
        """Close this client."""
        if self._transport is not None:
            self._transport.close()

get_capabilities(**kwargs)

The landing page provides links to the
  • The OpenAPI-definition (no fixed path),
  • The Conformance statements (path /conformance),
  • The processes metadata (path /processes),
  • The endpoint for job monitoring (path /jobs).

For more information, see OGC API — Processes — Part 1 Section 7.2.

Returns:

Name Type Description
Capabilities Capabilities

The landing page provides links to the API definition (link relations service-desc and service-doc), the Conformance declaration (path /conformance, link relation http://www.opengis.net/def/rel/ogc/1.0/conformance), and to other resources.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 500: A server error occurred.
Source code in cuiman\src\cuiman\api\client.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def get_capabilities(self, **kwargs: Any) -> Capabilities:
    """
    The landing page provides links to the:
      * The OpenAPI-definition (no fixed path),
      * The Conformance statements (path /conformance),
      * The processes metadata (path /processes),
      * The endpoint for job monitoring (path /jobs).

    For more information, see [OGC API — Processes — Part 1
    Section
    7.2](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_landing_page).

    Returns:
      Capabilities: The landing page provides links to the API definition
        (link relations `service-desc` and `service-doc`),
        the Conformance declaration (path `/conformance`,
        link relation `http://www.opengis.net/def/rel/ogc/1.0/conformance`),
        and to other resources.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `500`: A server error occurred.
    """
    return self._transport.call(
        TransportArgs(
            path="/",
            method="get",
            return_types={"200": Capabilities},
            error_types={"500": ApiError},
            extra_kwargs=kwargs,
        )
    )

get_conformance(**kwargs)

A list of all conformance classes, specified in a standard, that the server conforms to.

Conformance class URI
Core http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/core
OGC Process Description http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/ogc-process-description
JSON http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/json
HTML http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/html
OpenAPI Specification 3.0 http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/oas30
Job list http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/job-list
Callback http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/callback
Dismiss http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/dismiss

For more information, see OGC API — Processes — Part 1 Section 7.4.

Returns:

Name Type Description
ConformanceDeclaration ConformanceDeclaration

The URIs of all conformance classes supported by the server. To support "generic" clients that want to access multiple OGC API - Processes implementations - and not "just" a specific API / server, the server declares the conformance classes it implements and conforms to.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 500: A server error occurred.
Source code in cuiman\src\cuiman\api\client.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def get_conformance(self, **kwargs: Any) -> ConformanceDeclaration:
    """
    A list of all conformance classes, specified in a standard, that the
    server conforms to.

    | Conformance class | URI |
    |-----------|-------|
    |Core|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/core|
    |OGC Process Description|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/ogc-process-description|
    |JSON|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/json|
    |HTML|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/html|
    |OpenAPI Specification 3.0|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/oas30|
    |Job list|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/job-list|
    |Callback|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/callback|
    |Dismiss|http://www.opengis.net/spec/ogcapi-processes-1/1.0/conf/dismiss|

    For more information, see [OGC API — Processes — Part 1
    Section 7.4](https://docs.ogc.org/is/18-062r2/18-
    062r2.html#sc_conformance_classes).


    Returns:
      ConformanceDeclaration: The URIs of all conformance classes supported
        by the server. To support "generic" clients that want
        to access multiple OGC API - Processes implementations - and
        not "just" a specific API / server, the server declares
        the conformance classes it implements and conforms to.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `500`: A server error occurred.
    """
    return self._transport.call(
        TransportArgs(
            path="/conformance",
            method="get",
            return_types={"200": ConformanceDeclaration},
            error_types={"500": ApiError},
            extra_kwargs=kwargs,
        )
    )

get_processes(**kwargs)

The list of processes contains a summary of each process the OGC API - Processes offers, including the link to a more detailed description of the process.

For more information, see OGC API — Processes — Part 1 Section 7.9.

Returns:

Name Type Description
ProcessList ProcessList

Information about the available processes

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.
Source code in cuiman\src\cuiman\api\client.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def get_processes(self, **kwargs: Any) -> ProcessList:
    """
    The list of processes contains a summary of each process the OGC API -
    Processes offers, including the link to a more detailed description of
    the process.

    For more information, see [OGC API — Processes — Part 1
    Section
    7.9](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_process_list).


    Returns:
      ProcessList: Information about the available processes

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.
    """
    return self._transport.call(
        TransportArgs(
            path="/processes",
            method="get",
            return_types={"200": ProcessList},
            extra_kwargs=kwargs,
        )
    )

get_process(process_id, **kwargs)

The process description contains information about inputs and outputs and a link to the execution-endpoint for the process. The Core does not mandate the use of a specific process description to specify the interface of a process. That said, the Core requirements class makes the following recommendation:

Implementations should consider supporting the OGC process description.

For more information, see OGC API — Processes — Part 1 Section 7.10.

Parameters:

Name Type Description Default
process_id str
required
kwargs Any

Optional keyword arguments that may be used by the underlying transport.

 {}

Returns:

Name Type Description
ProcessDescription ProcessDescription

A process description.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 404: The requested URI was not found.
Source code in cuiman\src\cuiman\api\client.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def get_process(self, process_id: str, **kwargs: Any) -> ProcessDescription:
    """
    The process description contains information about inputs and outputs
    and a link to the execution-endpoint for the process. The Core does not
    mandate the use of a specific process description to specify the
    interface of a process. That said, the Core requirements class makes the
    following recommendation:

    Implementations **should** consider supporting the OGC process
    description.

    For more information, see [OGC API — Processes — Part 1
    Section 7.10](https://docs.ogc.org/is/18-062r2/18-
    062r2.html#sc_process_description).

    Args:
      process_id:
      kwargs: Optional keyword arguments that may be
        used by the underlying transport.

    Returns:
      ProcessDescription: A process description.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `404`: The requested URI was not found.
    """
    return self._transport.call(
        TransportArgs(
            path="/processes/{processID}",
            method="get",
            path_params={"processID": process_id},
            return_types={"200": ProcessDescription},
            error_types={"404": ApiError},
            extra_kwargs=kwargs,
        )
    )

execute_process(process_id, request, **kwargs)

Create a new job.

For more information, see OGC API — Processes — Part 1 Section 7.11.

Parameters:

Name Type Description Default
process_id str
required
kwargs Any

Optional keyword arguments that may be used by the underlying transport.

 {}
request ProcessRequest

Mandatory request JSON

 required

Returns:

Name Type Description
JobInfo JobInfo

Started asynchronous execution. Created job.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 404: The requested URI was not found.
  • 500: A server error occurred.
Source code in cuiman\src\cuiman\api\client.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def execute_process(
    self, process_id: str, request: ProcessRequest, **kwargs: Any
) -> JobInfo:
    """
    Create a new job.

    For more information, see [OGC API — Processes — Part 1
    Section
    7.11](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_create_job).

    Args:
      process_id:
      kwargs: Optional keyword arguments that may be
        used by the underlying transport.
      request: Mandatory request JSON

    Returns:
      JobInfo: Started asynchronous execution. Created job.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `404`: The requested URI was not found.
        - `500`: A server error occurred.
    """
    return self._transport.call(
        TransportArgs(
            path="/processes/{processID}/execution",
            method="post",
            path_params={"processID": process_id},
            request=request,
            return_types={"201": JobInfo},
            error_types={"404": ApiError, "500": ApiError},
            extra_kwargs=kwargs,
        )
    )

get_jobs(**kwargs)

List available jobs.

For more information, see OGC API — Processes — Part 1 Section 11.

Returns:

Name Type Description
JobList JobList

A list of jobs for this process.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 404: The requested URI was not found.
Source code in cuiman\src\cuiman\api\client.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
def get_jobs(self, **kwargs: Any) -> JobList:
    """
    List available jobs.

    For more information, see [OGC API — Processes — Part 1
    Section 11](https://docs.ogc.org/is/18-062r2/18-062r2.html#sc_job_list).


    Returns:
      JobList: A list of jobs for this process.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `404`: The requested URI was not found.
    """
    return self._transport.call(
        TransportArgs(
            path="/jobs",
            method="get",
            return_types={"200": JobList},
            error_types={"404": ApiError},
            extra_kwargs=kwargs,
        )
    )

get_job(job_id, **kwargs)

Show the status of a job.

For more information, see OGC API — Processes — Part 1 Section 7.12.

Parameters:

Name Type Description Default
job_id str

Local identifier of a job

 required
kwargs Any

Optional keyword arguments that may be used by the underlying transport.

 {}

Returns:

Name Type Description
JobInfo JobInfo

The status of a job.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 404: The requested URI was not found.
  • 500: A server error occurred.
Source code in cuiman\src\cuiman\api\client.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
def get_job(self, job_id: str, **kwargs: Any) -> JobInfo:
    """
    Show the status of a job.

    For more information, see [OGC API — Processes — Part 1
    Section 7.12](https://docs.ogc.org/is/18-062r2/18-
    062r2.html#sc_retrieve_status_info).

    Args:
      job_id: Local identifier of a job
      kwargs: Optional keyword arguments that may be
        used by the underlying transport.

    Returns:
      JobInfo: The status of a job.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `404`: The requested URI was not found.
        - `500`: A server error occurred.
    """
    return self._transport.call(
        TransportArgs(
            path="/jobs/{jobId}",
            method="get",
            path_params={"jobId": job_id},
            return_types={"200": JobInfo},
            error_types={"404": ApiError, "500": ApiError},
            extra_kwargs=kwargs,
        )
    )

dismiss_job(job_id, **kwargs)

Cancel a job execution and removes it from the jobs list.

For more information, see OGC API — Processes — Part 1 Section 13.

Parameters:

Name Type Description Default
job_id str

Local identifier of a job

 required
kwargs Any

Optional keyword arguments that may be used by the underlying transport.

 {}

Returns:

Name Type Description
JobInfo JobInfo

Information about the job.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 404: The requested URI was not found.
  • 500: A server error occurred.
Source code in cuiman\src\cuiman\api\client.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def dismiss_job(self, job_id: str, **kwargs: Any) -> JobInfo:
    """
    Cancel a job execution and removes it from the jobs list.

    For more information, see [OGC API — Processes — Part 1
    Section 13](https://docs.ogc.org/is/18-062r2/18-062r2.html#Dismiss).

    Args:
      job_id: Local identifier of a job
      kwargs: Optional keyword arguments that may be
        used by the underlying transport.

    Returns:
      JobInfo: Information about the job.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `404`: The requested URI was not found.
        - `500`: A server error occurred.
    """
    return self._transport.call(
        TransportArgs(
            path="/jobs/{jobId}",
            method="delete",
            path_params={"jobId": job_id},
            return_types={"200": JobInfo},
            error_types={"404": ApiError, "500": ApiError},
            extra_kwargs=kwargs,
        )
    )

get_job_results(job_id, **kwargs)

List available results of a job. In case of a failure, list errors instead.

For more information, see OGC API — Processes — Part 1 Section 7.13.

Parameters:

Name Type Description Default
job_id str

Local identifier of a job

 required
kwargs Any

Optional keyword arguments that may be used by the underlying transport.

 {}

Returns:

Name Type Description
JobResults JobResults

The results of a job.

Raises:

Type Description
ClientError

If the call to the web service fails with a status code != 2xx.

  • 404: The requested URI was not found.
  • 500: A server error occurred.
Source code in cuiman\src\cuiman\api\client.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
def get_job_results(self, job_id: str, **kwargs: Any) -> JobResults:
    """
    List available results of a job. In case of a failure, list errors
    instead.

    For more information, see [OGC API — Processes — Part 1
    Section 7.13](https://docs.ogc.org/is/18-062r2/18-
    062r2.html#sc_retrieve_job_results).

    Args:
      job_id: Local identifier of a job
      kwargs: Optional keyword arguments that may be
        used by the underlying transport.

    Returns:
      JobResults: The results of a job.

    Raises:
      ClientError: If the call to the web service fails
        with a status code != `2xx`.

        - `404`: The requested URI was not found.
        - `500`: A server error occurred.
    """
    return self._transport.call(
        TransportArgs(
            path="/jobs/{jobId}/results",
            method="get",
            path_params={"jobId": job_id},
            return_types={"200": JobResults},
            error_types={"404": ApiError, "500": ApiError},
            extra_kwargs=kwargs,
        )
    )

close()

Close this client.

Source code in cuiman\src\cuiman\api\client.py 
389
390
391
392
def close(self):
    """Close this client."""
    if self._transport is not None:
        self._transport.close()

cuiman.ClientConfig

Bases: AuthConfig, BaseSettings

Client configuration.

Parameters:

Name Type Description Default
api_url

a URL pointing to a service compliant with the OCG API - Processes.

 required
Source code in cuiman\src\cuiman\api\config.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
class ClientConfig(AuthConfig, BaseSettings):
    """Client configuration.

    Args:
        api_url: a URL pointing to a service compliant with
            the OCG API - Processes.
    """

    model_config = SettingsConfigDict(
        env_prefix="EOZILLA_",
        extra="forbid",
    )

    default_config: ClassVar["ClientConfig"]
    """
    Default instance. 
    Used to create pre-configured instances of this class.
    Designed to be overridden by library clients.
    """

    default_path: ClassVar[Path]
    """
    Name of the configuration's local default path. 
    Used for configuration persistence in `~/.<config_name>/`.
    Designed to be overridden by library clients.
    """

    return_type_map: ClassVar[dict[type, type]] = {}
    """
    A mapping from a hard-coded client return type to a 
    custom return type. The hard-coded return type is usually a 
    model class from `gavicore.models`. The custom return type 
    typically extends the model class.  
    Designed to be configured by library clients.
    The default mapping is empty.
    """

    api_url: Annotated[Optional[str], Field(title="Process API URL")] = None
    """
    The URL of the server that provides a web API compliant with
    OGC API - Processes, Part 1 - Core.
    """

    def _repr_json_(self):
        return self.model_dump(mode="json", by_alias=True), dict(
            root="Client configuration:"
        )

    @classmethod
    def create(
        cls,
        *,
        config: Optional["ClientConfig"] = None,
        config_path: Optional[Path | str] = None,
        **config_kwargs,
    ) -> "ClientConfig":
        # 0. from defaults
        config_dict = cls.default_config.to_dict()

        # 1. from file
        file_config = cls.from_file(config_path=config_path)
        if file_config is not None:
            _update_if_not_none(config_dict, file_config.to_dict())

        # 2. from env
        env_config = cls()
        _update_if_not_none(config_dict, env_config.to_dict())

        # 3. from config
        if config is not None:
            _update_if_not_none(config_dict, config.to_dict())

        # 4. from kwargs
        _update_if_not_none(config_dict, config_kwargs)

        return cls.new_instance(**config_dict)

    @classmethod
    def from_file(
        cls, config_path: Optional[str | Path] = None
    ) -> Optional["ClientConfig"]:
        config_path_: Path = cls.normalize_config_path(config_path)
        if not config_path_.exists():
            return None
        with config_path_.open("rt") as stream:
            # Note, we may switch TOML
            config_dict = yaml.safe_load(stream)
        return cls.new_instance(**config_dict)

    def write(self, config_path: Optional[str | Path] = None) -> Path:
        config_path = self.normalize_config_path(config_path)
        config_path.parent.mkdir(exist_ok=True)
        with config_path.open("wt") as stream:
            yaml.dump(
                self.model_dump(mode="json", by_alias=True, exclude_none=True), stream
            )
        return config_path

    @classmethod
    def normalize_config_path(cls, config_path) -> Path:
        return (
            config_path
            if isinstance(config_path, Path)
            else (Path(config_path) if config_path else cls.default_path)
        )

    @classmethod
    def new_instance(
        cls,
        **kwargs: Any,
    ) -> "ClientConfig":
        config_cls = type(ClientConfig.default_config)
        assert issubclass(config_cls, ClientConfig)
        return config_cls(**kwargs)

    def to_dict(self):
        return self.model_dump(
            mode="json",
            by_alias=True,
            exclude_none=True,
            exclude_defaults=True,
            exclude_unset=True,
        )

    # noinspection PyMethodParameters
    @field_validator("api_url")
    def validate_api_url(cls, v: str | None) -> str | None:
        return None if v is None or v == "" else str(HttpUrl(v))

    # noinspection PyUnusedLocal
    @classmethod
    def accept_process(
        cls, process_summary: ProcessSummary, **filter_kwargs: Any
    ) -> bool:
        """
        Predicate function that is used to filter the list of processes.
        The function is intended to be overridden by subclasses in order to allow
        for evaluating the given `process_summary` in an application-specific way.
        This includes the using custom fields in the given
        [ProcessSummary][gavicore.models.ProcessSummary] instance.

        Applications may use the [extend_model()][gavicore.util.model.extend_model]
        function to enhance existing model classes by their custom fields.

        The default implementation unconditionally returns `True`.

        Args:
            process_summary: A process summary.
            filter_kwargs: Implementation specific arguments passed
                by a user of this class.

        Returns:
            `True` to accept the given process, otherwise `False`.
        """
        return True

    # noinspection PyUnusedLocal
    @classmethod
    def accept_input(
        cls,
        process_description: ProcessDescription,
        input_name: str,
        input_description: InputDescription,
        **filter_kwargs: Any,
    ) -> bool:
        """
        Predicate function that is used to filter the list of inputs of a process.
        The function is intended to be overridden by subclasses in order to allow
        for evaluating the given `input_description` in an application-specific way.
        This includes the using custom fields in the given
        [InputDescription][gavicore.models.InputDescription] instance.

        Applications may use the [extend_model()][gavicore.util.model.extend_model]
        function to enhance existing model classes by their custom fields.

        The default implementation unconditionally returns `True`.

        Args:
            process_description: The process description.
            input_name: The input's name.
            input_description: A description of an
                input of the given `process_description`.
            filter_kwargs: Implementation specific arguments passed
                by a user of this class.

        Returns:
            `True` to accept the given input, otherwise `False`.
        """
        return True

    # noinspection PyUnusedLocal
    @classmethod
    def is_advanced_input(
        cls,
        process_description: ProcessDescription,
        input_name: str,
        input_description: InputDescription,
    ) -> bool:
        """
        Experimental method, do not use!

        Designed to be overridden by a custom `ClientConfig` class
        from which an instance will be assigned to `ClientConfig.default_config`
        to become effective.

        The default implementations checks if the given `input_description`
        has `additionalParameters`, and if so, if a parameter with name
        `"level"` has value `["advanced"]` (a list!).

        Args:
            process_description: The process description.
            input_name: The input's name.
            input_description: A description of an
                input of the given `process_description`.

        Returns:
            `True` if the input is advanced
            (e.g. for advanced process users only).
        """
        additional_parameters = input_description.additionalParameters
        if additional_parameters:
            parameters = additional_parameters.parameters
            if parameters:
                for p in parameters:
                    if p.name == "level" and p.value == ["advanced"]:
                        return True
        return False

default_config class-attribute

Default instance. Used to create pre-configured instances of this class. Designed to be overridden by library clients.

default_path class-attribute

Name of the configuration's local default path. Used for configuration persistence in ~/.<config_name>/. Designed to be overridden by library clients.

return_type_map = {} class-attribute

A mapping from a hard-coded client return type to a custom return type. The hard-coded return type is usually a model class from gavicore.models. The custom return type typically extends the model class.
Designed to be configured by library clients. The default mapping is empty.

api_url = None class-attribute instance-attribute

The URL of the server that provides a web API compliant with OGC API - Processes, Part 1 - Core.

accept_process(process_summary, **filter_kwargs) classmethod

Predicate function that is used to filter the list of processes. The function is intended to be overridden by subclasses in order to allow for evaluating the given process_summary in an application-specific way. This includes the using custom fields in the given [ProcessSummary][gavicore.models.ProcessSummary] instance.

Applications may use the [extend_model()][gavicore.util.model.extend_model] function to enhance existing model classes by their custom fields.

The default implementation unconditionally returns True.

Parameters:

Name Type Description Default
process_summary ProcessSummary

A process summary.

 required
filter_kwargs Any

Implementation specific arguments passed by a user of this class.

 {}

Returns:

Type Description
bool

True to accept the given process, otherwise False.
Source code in cuiman\src\cuiman\api\config.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
@classmethod
def accept_process(
    cls, process_summary: ProcessSummary, **filter_kwargs: Any
) -> bool:
    """
    Predicate function that is used to filter the list of processes.
    The function is intended to be overridden by subclasses in order to allow
    for evaluating the given `process_summary` in an application-specific way.
    This includes the using custom fields in the given
    [ProcessSummary][gavicore.models.ProcessSummary] instance.

    Applications may use the [extend_model()][gavicore.util.model.extend_model]
    function to enhance existing model classes by their custom fields.

    The default implementation unconditionally returns `True`.

    Args:
        process_summary: A process summary.
        filter_kwargs: Implementation specific arguments passed
            by a user of this class.

    Returns:
        `True` to accept the given process, otherwise `False`.
    """
    return True

accept_input(process_description, input_name, input_description, **filter_kwargs) classmethod

Predicate function that is used to filter the list of inputs of a process. The function is intended to be overridden by subclasses in order to allow for evaluating the given input_description in an application-specific way. This includes the using custom fields in the given [InputDescription][gavicore.models.InputDescription] instance.

Applications may use the [extend_model()][gavicore.util.model.extend_model] function to enhance existing model classes by their custom fields.

The default implementation unconditionally returns True.

Parameters:

Name Type Description Default
process_description ProcessDescription

The process description.

 required
input_name str

The input's name.

 required
input_description InputDescription

A description of an input of the given process_description.

 required
filter_kwargs Any

Implementation specific arguments passed by a user of this class.

 {}

Returns:

Type Description
bool

True to accept the given input, otherwise False.
Source code in cuiman\src\cuiman\api\config.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
@classmethod
def accept_input(
    cls,
    process_description: ProcessDescription,
    input_name: str,
    input_description: InputDescription,
    **filter_kwargs: Any,
) -> bool:
    """
    Predicate function that is used to filter the list of inputs of a process.
    The function is intended to be overridden by subclasses in order to allow
    for evaluating the given `input_description` in an application-specific way.
    This includes the using custom fields in the given
    [InputDescription][gavicore.models.InputDescription] instance.

    Applications may use the [extend_model()][gavicore.util.model.extend_model]
    function to enhance existing model classes by their custom fields.

    The default implementation unconditionally returns `True`.

    Args:
        process_description: The process description.
        input_name: The input's name.
        input_description: A description of an
            input of the given `process_description`.
        filter_kwargs: Implementation specific arguments passed
            by a user of this class.

    Returns:
        `True` to accept the given input, otherwise `False`.
    """
    return True

is_advanced_input(process_description, input_name, input_description) classmethod

Experimental method, do not use!

Designed to be overridden by a custom ClientConfig class from which an instance will be assigned to ClientConfig.default_config to become effective.

The default implementations checks if the given input_description has additionalParameters, and if so, if a parameter with name "level" has value ["advanced"] (a list!).

Parameters:

Name Type Description Default
process_description ProcessDescription

The process description.

 required
input_name str

The input's name.

 required
input_description InputDescription

A description of an input of the given process_description.

 required

Returns:

Type Description
bool

True if the input is advanced
bool

(e.g. for advanced process users only).
Source code in cuiman\src\cuiman\api\config.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
@classmethod
def is_advanced_input(
    cls,
    process_description: ProcessDescription,
    input_name: str,
    input_description: InputDescription,
) -> bool:
    """
    Experimental method, do not use!

    Designed to be overridden by a custom `ClientConfig` class
    from which an instance will be assigned to `ClientConfig.default_config`
    to become effective.

    The default implementations checks if the given `input_description`
    has `additionalParameters`, and if so, if a parameter with name
    `"level"` has value `["advanced"]` (a list!).

    Args:
        process_description: The process description.
        input_name: The input's name.
        input_description: A description of an
            input of the given `process_description`.

    Returns:
        `True` if the input is advanced
        (e.g. for advanced process users only).
    """
    additional_parameters = input_description.additionalParameters
    if additional_parameters:
        parameters = additional_parameters.parameters
        if parameters:
            for p in parameters:
                if p.name == "level" and p.value == ["advanced"]:
                    return True
    return False

cuiman.ClientError

Bases: Exception

Raised if a web API call failed.

The failure can have several reasons such as

  • the request failed with a status code that is not 2xx, or
  • the received JSON response is not parsable.

Parameters:

Name Type Description Default
message str

The error message

 required
api_error ApiError

The details describing the error that occurred on the server or the details that describe a non-expected response from the server.

 required
Source code in cuiman\src\cuiman\api\exceptions.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class ClientError(Exception):
    """Raised if a web API call failed.

     The failure can have several reasons such as

    - the request failed with a status code that is not 2xx, or
    - the received JSON response is not parsable.

    Args:
        message: The error message
        api_error: The details describing the error that occurred on the server
            or the details that describe a non-expected response from the server.
    """

    def __init__(self, message: str, api_error: ApiError):
        super().__init__(message)
        self.api_error = api_error