ripple1d.api package

Submodules

ripple1d.api.app module

ripple1d.api.log module

ripple1d.api.manager module

Manage Flask and Huey.

class ripple1d.api.manager.RippleManager(args)

Bases: object

Launch Flask and Huey in separate terminals and manage their lifecycles.

print_config()

Pring configuration settings for the Ripple Manager.

start()

Start the Ripple API and Huey consumer in separate terminals.

ripple1d.api.manager.main()

Run Ripple in API mode.

ripple1d.api.run_api_test module

Run API tests.

ripple1d.api.run_api_test.execute_test_process()

Send a POST request to initiate a test process on a local server.

This function constructs the URL for initiating a test process using the FLASK_RUN_PORT specified in the .flaskenv file. It then sends a POST request with an empty JSON payload to this URL.

If the response status code is not HTTPStatus.OK (200), it raises an HTTPError with the response content and status code. Otherwise, it prints the JSON-decoded response content.

ripple1d.api.run_api_test.main()

Execute the ping and test process functions.

This function first calls the ping() function to send a GET request to the /ping endpoint. Then, it calls the execute_test_process() function to initiate a test process by sending a POST request to the relevant endpoint.

ripple1d.api.run_api_test.ping()

Send a GET request to the /ping endpoint of a local server.

This function constructs the URL for the /ping endpoint using the FLASK_RUN_PORT specified in the .flaskenv file. It then sends a GET request to this URL.

If the response status code is not HTTPStatus.OK (200), it raises an HTTPError with the response content and status code. Otherwise, it prints the JSON-decoded response content.

ripple1d.api.tasks module

huey instance for huey + Flask REST API (to be called by huey_consumer.py).

ripple1d.api.tasks.create_and_enqueue_task(func, kwargs={}, priority=None)

Create a task instance, then add it to the custom table “task_status” before adding it to the queue.

Return the Result object associated with the task, which is the same class type that would be returned by calling the @huey.task()-decorated function itself (in this case, _process).

This is needed to avoid a race condition, this ensures that the task exists in the custom table before it starts executing.

Return type:

Result

ripple1d.api.tasks.fetch_logs(task_id)

For given task ID, return logs.

Return type:

str

ripple1d.api.tasks.fetch_ogc_status(task_id)

For given task ID, return its current status in OGC API terms.

Return type:

str

ripple1d.api.tasks.fetch_one_query(task_id, field, table)

Return a fetch one query given table and field given task_id.

Return type:

str

ripple1d.api.tasks.fetch_results(task_id)

For given task ID, return results.

Return type:

str

ripple1d.api.tasks.huey_status(task_id)

For given task ID, return its current status in huey terms.

Return type:

str

ripple1d.api.tasks.job_dismissed(task_id)

For given task ID, return its current status in OGC API terms.

Return type:

bool

ripple1d.api.tasks.noop(task_id=None)

Do nothing except log a message. For ping endpoint and testing.

ripple1d.api.tasks.revoke_task(task_id)

Revoke a task.

ripple1d.api.tasks.revoke_task_by_pid(task_id)

Revoke a task by pid.

ripple1d.api.tasks.subprocess_caller(func, args, task_id, log_dir='', log_level=20, timeout=None)

Call the specified function through a subprocess.

ripple1d.api.tasks.task_results(only_task_id)

Return dictionary of tasks, where key is task ID and value is subdict with val, err, and tb.

If only_task_id is None, all tasks will be returned. If only_task_id is not None, only the provided task will be returned.

Return type:

dict[str, dict]

ripple1d.api.tasks.task_status(only_task_id)

Return dictionary of tasks, where key is task ID and value is subdict (status fields related to each task).

If only_task_id is None, all tasks will be returned. If only_task_id is not None, only the provided task will be returned.

Return type:

dict[str, dict]

ripple1d.api.tasks.task_summary(only_task_id)

Return dictionary of tasks, where key is task ID and value is subdict with status and results.

If only_task_id is None, all tasks will be returned. If only_task_id is not None, only the provided task will be returned.

Return type:

dict[str, dict]

ripple1d.api.tasks.update_p_id(task_id, p_id)

Update p_id.

ripple1d.api.utils module

Generic utilities supporting the huey + Flask REST API.

ripple1d.api.utils.find_huey_consumer()

Find the huey_consumer.py file in the installed packages and return its path.

ripple1d.api.utils.get_unexpected_and_missing_args(func, kwargs_provided)

Inspect func and evaluate whether the provided kwargs will satisfy it, considering its required & optional args.

Return a list of unexpected kwargs (provided, but not called for), :rtype: tuple[list, list]

and return a list of missing kwargs (required but not provided). WARNING: assumes the func does not use *args or **kwargs features.

Module contents

Initialize API package.