Merge branch 'develop' of github.com:MAAP-Project/maap-api-nasa into …

…develop

CHANGELOG.md

@@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [v4.1.0] - 2024-09-10
+- [pull/131](https://github.com/MAAP-Project/maap-api-nasa/pull/131) - Added query params to job list endpoint 
+- [pull/135](https://github.com/MAAP-Project/maap-api-nasa/pull/135) - User secret management 
+- [pull/137](https://github.com/MAAP-Project/maap-api-nasa/pull/137) - Organizations & job queues management
+- [pull/136](https://github.com/MAAP-Project/maap-api-nasa/pull/136) - Add support for DPS sandbox queue
+- [pull/132](https://github.com/MAAP-Project/maap-api-nasa/pull/132) - Remove {username} param from DPS job list endpoint 
+
 ## [v4.0.0] - 2024-06-26
 - [issues/111](https://github.com/MAAP-Project/maap-api-nasa/issues/111) - Implement github actions CICD and convert to poetry based build
 - [pull/110](https://github.com/MAAP-Project/maap-api-nasa/pull/110) - Remove postgres from docker-compose
@@ -16,4 +23,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 
 [unreleased]: https://github.com/MAAP-Project/maap-api-nasa/v4.0.0...HEAD
-[v4.0.0]: https://github.com/MAAP-Project/maap-api-nasa/compare/v3.1.5...v4.0.0
+[v4.1.0]: https://github.com/MAAP-Project/maap-api-nasa/compare/v4.0.0...v4.1.0
+[v4.0.0]: https://github.com/MAAP-Project/maap-api-nasa/compare/v3.1.5...v4.0.0

api/endpoints/admin.py

@@ -2,6 +2,7 @@
 from flask_restx import Resource
 from flask import request
 from flask_api import status
+from api.models.job_queue import JobQueue
 from api.models.role import Role
 from api.restplus import api
 from api.auth.security import login_required
@@ -10,12 +11,107 @@
 from api.schemas.pre_approved_schema import PreApprovedSchema
 from datetime import datetime
 import json
-
+from api.utils import job_queue
 from api.utils.http_util import err_response
 
 log = logging.getLogger(__name__)
 ns = api.namespace('admin', description='Operations related to the MAAP admin')
 
+@ns.route('/job-queues')
+class JobQueuesCls(Resource):
+
+    @api.doc(security='ApiKeyAuth')
+    @login_required(role=Role.ROLE_ADMIN)
+    def get(self):
+        """
+        Lists the job queues and associated organizations
+        :return:
+        """
+        all_queues = job_queue.get_all_queues()
+        return all_queues
+
+
+    @api.doc(security='ApiKeyAuth')
+    @login_required(role=Role.ROLE_ADMIN)
+    def post(self):
+
+        """
+        Create new job queue.
+        """
+
+        req_data = request.get_json()
+        if not isinstance(req_data, dict):
+            return err_response("Valid JSON body object required.")
+
+        queue_name = req_data.get("queue_name", "")
+        if not isinstance(queue_name, str) or not queue_name:
+            return err_response("Valid queue name is required.")
+
+        queue_description = req_data.get("queue_description", "")
+        if not isinstance(queue_description, str) or not queue_description:
+            return err_response("Valid queue description is required.")
+
+        guest_tier = req_data.get("guest_tier", False)
+        is_default = req_data.get("is_default", False)
+        time_limit_minutes = req_data.get("time_limit_minutes", 0)
+        orgs = req_data.get("orgs", [])
+
+        new_queue = job_queue.create_queue(queue_name, queue_description, guest_tier, is_default, time_limit_minutes, orgs)
+        return new_queue
+
+
+@ns.route('/job-queues/<int:queue_id>')
+class JobQueueCls(Resource):
+
+    @api.doc(security='ApiKeyAuth')
+    @login_required()
+    def put(self, queue_id):
+
+        """
+        Update job queue. Only supplied fields are updated.
+        """
+
+        if not queue_id:
+            return err_response("Job queue id is required.")
+
+        req_data = request.get_json()
+        if not isinstance(req_data, dict):
+            return err_response("Valid JSON body object required.")
+
+        queue = db.session.query(JobQueue).filter_by(id=queue_id).first()
+
+        if queue is None:
+            return err_response(msg="No job queue found with id " + queue_id)
+
+        queue.queue_name = req_data.get("queue_name", queue.queue_name)
+        queue.queue_description = req_data.get("queue_description", queue.queue_description)
+        queue.guest_tier = req_data.get("guest_tier", queue.guest_tier)
+        queue.is_default = req_data.get("is_default", queue.is_default)
+        queue.time_limit_minutes = req_data.get("time_limit_minutes", queue.time_limit_minutes)
+        orgs = req_data.get("orgs", [])
+
+        updated_queue = job_queue.update_queue(queue, orgs)
+        return updated_queue
+
+
+    @api.doc(security='ApiKeyAuth')
+    @login_required(role=Role.ROLE_ADMIN)
+    def delete(self, queue_id):
+        """
+        Delete job queue
+        """
+
+        queue = db.session.query(JobQueue).filter_by(id=queue_id).first()
+        queue_name = queue.queue_name
+
+        if queue is None:
+            return err_response(msg="Job queue does not exist")
+
+        job_queue.delete_queue(queue_id)
+
+        return {"code": status.HTTP_200_OK, "message": "Successfully deleted {}.".format(queue_name)}
+
+
 @ns.route('/pre-approved')
 class PreApprovedEmails(Resource):
 

api/endpoints/algorithm.py

@@ -1,5 +1,8 @@
 import logging
 import os
+from collections import namedtuple
+
+import sqlalchemy
 from flask import request, Response
 from flask_restx import Resource, reqparse
 from flask_api import status
@@ -10,6 +13,7 @@
 import traceback
 import api.utils.github_util as git
 import api.utils.hysds_util as hysds
+import api.utils.http_util as http_util
 import api.settings as settings
 import api.utils.ogc_translate as ogc
 from api.auth.security import get_authorized_user, login_required
@@ -19,6 +23,8 @@
 from datetime import datetime
 import json
 
+from api.utils import job_queue
+
 log = logging.getLogger(__name__)
 
 ns = api.namespace('mas', description='Operations to register an algorithm')
@@ -255,12 +261,16 @@ def post(self):
 
         try:
             # validate if input queue is valid
-            if resource is None: resource = settings.DEFAULT_QUEUE
-            if resource not in hysds.get_mozart_queues():
-                response_body["code"] = status.HTTP_500_INTERNAL_SERVER_ERROR
-                response_body["message"] = "The resource {} is invalid. Please select from one of {}".format(resource, hysds.get_mozart_queues())
-                response_body["error"] = "Invalid queue in request: {}".format(req_data)
-                return response_body, 500
+            user = get_authorized_user()
+            if resource is None:
+                resource = job_queue.get_default_queue().queue_name
+            else:
+                valid_queues = job_queue.get_user_queues(user.id)
+                valid_queue_names = list(map(lambda q: q.queue_name, valid_queues))
+                if resource not in valid_queue_names:
+                    return http_util.err_response(msg=f"User does not have permissions for resource {resource}."
+                                                      f"Please select from one of {valid_queue_names}",
+                                                  code=status.HTTP_400_BAD_REQUEST)
             # clean up any old specs from the repo
             repo = git.clean_up_git_repo(repo, repo_name=settings.REPO_NAME)
             # creating hysds-io file
@@ -311,6 +321,7 @@ def post(self):
             hysds.write_file("{}/{}".format(settings.REPO_PATH, settings.REPO_NAME), "job-submission.json",
                              job_submission_json)
             logging.debug("Created spec files")
+
         except Exception as ex:
             tb = traceback.format_exc()
             response_body["code"] = status.HTTP_500_INTERNAL_SERVER_ERROR
@@ -411,6 +422,10 @@ def get(self, algo_id):
         try:
             job_type = "job-{}".format(algo_id)
             response = hysds.get_job_spec(job_type)
+            if response is None:
+                return Response(ogc.get_exception(type="FailedSearch", origin_process="DescribeProcess",
+                                                  ex_message="Algorithm not found. {}".format(job_type)),
+                                status=status.HTTP_404_NOT_FOUND,mimetype='text/xml')
             params = response.get("result").get("params")
             queue = response.get("result").get("recommended-queues")[0]
             response_body = ogc.describe_process_response(algo_id, params, queue)
@@ -450,6 +465,9 @@ def delete(self, algo_id):
 
 @ns.route('/algorithm/resource')
 class ResourceList(Resource):
+
+    @api.doc(security='ApiKeyAuth')
+    @login_required()
     def get(self):
         """
         This function would query DPS to see what resources (named based on memory space) are available for
@@ -458,9 +476,12 @@ def get(self):
         """
         try:
             response_body = {"code": None, "message": None}
-            queues = hysds.get_mozart_queues()
+            user = get_authorized_user()
+            queues = job_queue.get_user_queues(user.id)
+            queue_names = list(map(lambda q: q.queue_name, queues))
+
             response_body["code"] = status.HTTP_200_OK
-            response_body["queues"] = queues
+            response_body["queues"] = queue_names
             response_body["message"] = "success"
             return response_body
         except Exception as ex:

api/endpoints/job.py

@@ -5,6 +5,7 @@
 from api.restplus import api
 import api.utils.hysds_util as hysds
 import api.utils.ogc_translate as ogc
+import api.utils.job_queue as job_queue
 import api.settings as settings
 try:
     import urllib.parse as urlparse
@@ -60,9 +61,13 @@ def post(self):
 
         try:
             dedup = "false" if dedup is None else dedup
-            queue = hysds.get_recommended_queue(job_type=job_type) if queue is None or queue is "" else queue
-            response = hysds.mozart_submit_job(job_type=job_type, params=params, dedup=dedup, queue=queue,
-                                               identifier=identifier)
+            user = get_authorized_user()
+            queue = job_queue.validate_or_get_queue(queue, job_type, user.id)
+            job_time_limit = hysds_io.get("result").get("soft_time_limit", 86400)
+            if job_queue.contains_time_limit(queue):
+                job_time_limit = int(queue.time_limit_minutes) * 60
+            response = hysds.mozart_submit_job(job_type=job_type, params=params, dedup=dedup, queue=queue.queue_name,
+                                               identifier=identifier, job_time_limit=int(job_time_limit))
 
             logging.info("Mozart Response: {}".format(json.dumps(response)))
             job_id = response.get("result")
@@ -78,6 +83,10 @@ def post(self):
                 return Response(ogc.status_response(job_id=job_id, job_status=job_status), mimetype='text/xml')
             else:
                 raise Exception(response.get("message"))
+        except ValueError as ex:
+            logging.error(traceback.format_exc())
+            return Response(ogc.get_exception(type="FailedJobSubmit", origin_process="Execute",
+                                              ex_message=str(ex)), status.HTTP_400_BAD_REQUEST)
         except Exception as ex:
             logging.info("Error submitting job: {}".format(ex))
             return Response(ogc.get_exception(type="FailedJobSubmit", origin_process="Execute",