Add Endpoint for Submitting Jobs with File POSTing

Makefile

@@ -16,6 +16,7 @@ export PYTHONPATH = ${API}:${CHECK_PROCESSING_TIME}:${GET_FILES}:${HANDLE_BATCH_
 
 
 build: render
+	cp -r job_spec apps/api/src/hyp3_api/job_spec
 	python -m pip install --upgrade -r requirements-apps-api.txt -t ${API}; \
 	python -m pip install --upgrade -r requirements-apps-api-binary.txt --platform manylinux2014_x86_64 --only-binary=:all: -t ${API}; \
 	python -m pip install --upgrade -r requirements-apps-handle-batch-event.txt -t ${HANDLE_BATCH_EVENT}; \

apps/api/api-cf.yml.j2

@@ -182,6 +182,11 @@ Resources:
             Action:
               - dynamodb:GetItem
             Resource: !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${AccessCodesTable}*"
+          - Effect: Allow
+            Action:
+              - s3:PutObject
+              - s3:PutObjectTagging
+            Resource: !Sub "arn:aws:s3:::${ContentBucket}/*"
 
   Lambda:
     Type: AWS::Lambda::Function

apps/api/src/hyp3_api/api-spec/job_parameters.yml.j2

@@ -15,14 +15,27 @@ components:
         - {{ job_type }}
         {% endfor %}
 
+    upload_job_type:
+      description: Type of process to run this job.
+      type: string
+      example: {{ job_types.__iter__().__next__() }}
+      enum:
+        {% for job_type, job_spec in job_types.items() %}
+        {% if 'files' in job_spec.keys() %}
+        - {{ job_type }}
+        {% endif %}
+        {% endfor %}
+
     job_parameters:
+      type: object
       anyOf:
         {% for job_type in job_types %}
         - $ref: "#/components/schemas/{{ job_type }}Parameters"
         {% endfor %}
 
     {% for job_type, job_spec in job_types.items() %}
     {{ job_type }}Parameters:
+      title: {{ job_type }}Parameters
       description: Parameters for running {{ job_type }} jobs
       type: object
       additionalProperties: false
@@ -41,12 +54,13 @@ components:
       {% endfor %}
 
     {{ job_type }}Job:
+      title: {{ job_type }}Job
       description: Contains user provided information on running a new {{ job_type }} job.
       type: object
-      additionalProperties: false
       required:
         - job_type
         - job_parameters
+      additionalProperties: false
       properties:
         job_type:
           type: string
@@ -60,5 +74,4 @@ components:
           $ref: "./openapi-spec.yml#/components/schemas/bucket_prefix"
         job_parameters:
           $ref: "#/components/schemas/{{ job_type }}Parameters"
-
     {% endfor %}

apps/api/src/hyp3_api/api-spec/openapi-spec.yml.j2

@@ -16,7 +16,6 @@ servers:
 {% endif %}
 
 paths:
-
   /costs:
     get:
       description: Get table of job costs.
@@ -28,8 +27,54 @@ paths:
               schema:
                 $ref: "#/components/schemas/costs_response"
 
-  /jobs:
+  /upload-job:
+    post:
+      description: Submits a list of jobs for processing.
+      requestBody:
+        content:
+          multipart/form-data:
+            schema:
+              type: object
+              required:
+                - job_type
+              properties:
+                job_type:
+                  $ref: "./job_parameters.yml#/components/schemas/upload_job_type"
+                name:
+                  $ref: "#/components/schemas/name"
+                bucket:
+                  $ref: "#/components/schemas/bucket"
+                bucket_prefix: 
+                  $ref: "#/components/schemas/bucket_prefix"
+                job_parameters:
+                  $ref: "./job_parameters.yml#/components/schemas/job_parameters"
+                {% for job_type, job_spec in job_types.items() %}
+                {% if 'files' in job_spec.keys() %}
+                {% for parameter, parameter_spec in job_spec['files'].items() %}
+                {{ parameter }}:
+                  {{ json.dumps(parameter_spec['api_schema']) }}
+                {% endfor %}  
+                {% endif %}
+                {% endfor %}
+            encoding:
+            {% for job_type, job_spec in job_types.items() %}
+            {% if 'files' in job_spec.keys() %}
+            {% for parameter, parameter_spec in job_spec['files'].items() %}
+              {{ parameter }}:
+                contentType: {{ parameter_spec['allowed_types'] | join(', ') }}
+            {% endfor %}  
+            {% endif %}
+            {% endfor %}
+        required: true
+      responses:
+        "200":
+          description: 200 response
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/jobs_response"
 
+  /jobs:
     post:
       description: Submits a list of jobs for processing.
       requestBody:

apps/api/src/hyp3_api/handlers.py

@@ -1,6 +1,8 @@
+import json
 from http.client import responses
+from uuid import uuid4
 
-from flask import Response, abort, jsonify, request
+from flask import Request, Response, abort, jsonify, request
 
 import dynamo
 from dynamo.exceptions import (
@@ -13,7 +15,7 @@
 )
 from hyp3_api import util
 from hyp3_api.multi_burst_validation import MultiBurstValidationError
-from hyp3_api.validation import CmrError, ValidationError, validate_jobs
+from hyp3_api.validation import CmrError, ValidationError, validate_files, validate_job_parameters, validate_jobs
 
 
 def problem_format(status: int, message: str) -> Response:
@@ -23,6 +25,50 @@ def problem_format(status: int, message: str) -> Response:
     return response
 
 
+def _get_request_dict(request: Request) -> dict:
+    """Retrieve and sanitize request dictionary from `/upload-job` form."""
+    request_form = dict(request.form)
+    allowed_params = ['job_type', 'name', 'bucket', 'bucket_prefix', 'job_parameters']
+
+    # File params from other job types with files will be included in the request and need to be removed
+    params = list(request_form.keys())
+    for param in params:
+        if param not in allowed_params:
+            request_form.pop(param)
+
+    request_form['job_parameters'] = json.loads(request_form['job_parameters'])
+
+    return request_form
+
+
+def post_upload_job(request: Request, user: str) -> dict:
+    request_dict = _get_request_dict(request)
+    request_dict['job_id'] = str(uuid4())
+    try:
+        validate_files(request)
+        validate_job_parameters(request_dict)
+        validate_jobs([request_dict])
+    except CmrError as e:
+        abort(problem_format(503, str(e)))
+    except (ValidationError, MultiBurstValidationError) as e:
+        abort(problem_format(400, str(e)))
+    try:
+        request_dict = dynamo.jobs.handle_content_bucket(request_dict)
+        for file_param, file_obj in request.files.items():
+            s3_uri = util.save_and_upload_to_s3(
+                file_obj=file_obj, bucket=request_dict['bucket'], bucket_prefix=request_dict['bucket_prefix']
+            )
+            request_dict['job_parameters'][file_param] = s3_uri
+        request_dict = dynamo.jobs.put_jobs(user, [request_dict])[0]
+    except UnexpectedApplicationStatusError as e:
+        abort(problem_format(403, str(e)))
+    except InsufficientCreditsError as e:
+        abort(problem_format(400, str(e)))
+    except CustomPrefixForDefaultBucketError as e:
+        abort(problem_format(400, str(e)))
+    return request_dict
+
+
 def post_jobs(body: dict, user: str) -> dict:
     print(body)
     try:

apps/api/src/hyp3_api/routes.py

@@ -26,7 +26,7 @@
 
 
 JWKS_CLIENT = auth.get_jwks_client()
-AUTHENTICATED_ROUTES = ['/jobs', '/user']
+AUTHENTICATED_ROUTES = ['/jobs', '/user', '/upload-job']
 
 
 @app.before_request
@@ -139,6 +139,12 @@ def costs_get() -> Response:
     return jsonify(dynamo.jobs.COSTS)
 
 
+@app.route('/upload-job', methods=['POST'])
+@openapi
+def upload_job_post() -> Response:
+    return jsonify(handlers.post_upload_job(request, g.user))
+
+
 @app.route('/jobs', methods=['POST'])
 @openapi
 def jobs_post() -> Response:

apps/api/src/hyp3_api/util.py

@@ -1,10 +1,17 @@
 import binascii
 import json
 from base64 import b64decode, b64encode
+from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import Any
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
 import boto3
+from boto3.s3.transfer import TransferConfig
+from werkzeug.datastructures import FileStorage
+
+
+S3_CLIENT = boto3.client('s3')
 
 
 class TokenDeserializeError(Exception):
@@ -53,3 +60,22 @@ def build_next_url(url: str, start_token: str, x_forwarded_host: str | None = No
 def get_current_account_arn() -> str:
     sts = boto3.client('sts')
     return sts.get_caller_identity()['Account']
+
+
+def _upload_file_to_s3(
+    path_to_file: Path, content_type: str, bucket: str, prefix: str = '', chunk_size: int = 8_388_608
+) -> None:
+    key = str(Path(prefix) / path_to_file.name)
+    extra_args = {'ContentType': content_type}
+    config = TransferConfig(multipart_threshold=chunk_size, multipart_chunksize=chunk_size)
+    S3_CLIENT.upload_file(str(path_to_file), bucket, key, extra_args, Config=config)
+
+
+def save_and_upload_to_s3(file_obj: FileStorage, bucket: str, bucket_prefix: str) -> str:
+    filename = file_obj.filename
+    assert filename
+    with TemporaryDirectory() as temp_dir:
+        filepath = Path(temp_dir) / filename
+        file_obj.save(filepath)
+        _upload_file_to_s3(filepath, file_obj.mimetype, bucket, bucket_prefix)
+    return f's3://{bucket}/{bucket_prefix}/{filename}'

apps/api/src/hyp3_api/validation.py

@@ -6,9 +6,12 @@
 
 import requests
 import yaml
+from flask import Request
+from jsonschema import Draft7Validator
 from shapely.geometry import MultiPolygon, Polygon, shape
 
 from hyp3_api import CMR_URL, multi_burst_validation
+from hyp3_api.openapi import get_spec_yaml
 from hyp3_api.util import get_granules
 
 
@@ -94,6 +97,70 @@ def _make_sure_granules_exist(granules: Iterable[str], granule_metadata: list[di
         raise ValidationError(f'Some requested scenes could not be found: {", ".join(not_found_granules)}')
 
 
+def validate_job_parameters(request_dict: dict) -> None:
+    """Manual validation of job parameters for `/upload-job`.
+
+    Args:
+        request_dict: Sanitized request dictionary to be validated
+    """
+    api_spec_dict = get_spec_yaml(Path(__file__).parent / 'api-spec' / 'openapi-spec.yml')
+    job_parameters_by_title = {
+        x['title']: x for x in api_spec_dict['components']['schemas']['job']['properties']['job_parameters']['anyOf']
+    }
+
+    job_type = request_dict['job_type']
+    job_parameter_schema = job_parameters_by_title.get(f'{job_type}Parameters')
+
+    validator = Draft7Validator(job_parameter_schema)  # type: ignore
+    errors = sorted(validator.iter_errors(request_dict['job_parameters']), key=lambda e: e.path)
+
+    if errors:
+        raise ValidationError(errors[-1])
+
+
+def _check_for_missing_files(request_files: dict, file_spec: dict) -> None:
+    missing_files = []
+    for key in file_spec.keys():
+        if 'required' in file_spec[key].keys() and file_spec[key]['required']:
+            if key not in request_files.keys():
+                missing_files.append(key)
+    if len(missing_files) > 0:
+        msg = f'Missing required file(s): {", ".join(missing_files)}'
+        raise ValidationError(msg)
+
+
+def _check_for_irrelevant_files(request_files: dict, file_spec: dict, job_type: str) -> None:
+    for file_param in request_files.keys():
+        if file_param not in file_spec.keys():
+            msg = f'Invalid file included for {job_type} job type: ({file_param}) {request_files[file_param].filename}'
+            raise ValidationError(msg)
+
+
+def _check_correct_file_type(request_files: dict, file_spec: dict) -> None:
+    for param, file_obj in request_files.items():
+        filetype = file_obj.mimetype
+        allowed_types = file_spec[param]['allowed_types']
+        if filetype not in allowed_types:
+            msg = f"Invalid file type for {param}: '{filetype}' is not one of {allowed_types}."
+            raise ValidationError(msg)
+
+
+def validate_files(request: Request) -> None:
+    """Check that files have been included correctly in an `/upload-job` request."""
+    job_type = request.form['job_type']
+    job_spec_path = Path(__file__).parent / f'job_spec/{job_type}.yml'
+
+    with Path.open(job_spec_path) as file:
+        job_spec = yaml.safe_load(file)
+
+    file_spec = job_spec[job_type]['files']
+    request_files = dict(request.files)
+
+    _check_for_missing_files(request_files, file_spec)
+    _check_for_irrelevant_files(request_files, file_spec, job_type)
+    _check_correct_file_type(request_files, file_spec)
+
+
 def check_cmr_query_succeeded(job: dict, granule_metadata: list[dict]) -> None:
     if not granule_metadata:
         raise CmrError(

apps/render_cf.py

@@ -236,9 +236,11 @@ def render_costs(job_types: dict, cost_profile: str) -> None:
 
 
 def validate_job_spec(job_type: str, job_spec: dict) -> None:
-    expected_fields = sorted(['required_parameters', 'parameters', 'cost_profiles', 'validators', 'steps'])
-    actual_fields = sorted(job_spec.keys())
-    if actual_fields != expected_fields:
+    expected_fields = {'required_parameters', 'parameters', 'cost_profiles', 'validators', 'steps'}
+    optional_fields = {'files'}
+    actual_fields = set(job_spec.keys())
+
+    if actual_fields - optional_fields != expected_fields:
         raise ValueError(f'{job_type} has fields {actual_fields} but should have {expected_fields}')
 
     if 'job_id' in job_spec['parameters']:

job_spec/AK_FIRE_SAFE.yml

@@ -54,6 +54,22 @@ AK_FIRE_SAFE:
         description: Prefix to pull the fire detection products.
         type: string
         example: "txt"
+    ak_fire_safe_file:
+      api_schema:
+        description: S3 uri for the file
+        type: string
+        nullable: true
+        default: null
+        example: ""
+  files:
+    ak_fire_safe_file:
+      allowed_types: ["image/jpeg", "image/png"]
+      required: true
+      api_schema:
+        description: Test file to upload
+        type: string
+        format: binary
+
   cost_profiles:
     DEFAULT:
       cost: 1.0