Add job tagging to API (#2774)

* add job tagging to API

Signed-off-by: sharpd <davidsharp7@gmail.com>

* fix various tests that were failing

Signed-off-by: sharpd <davidsharp7@gmail.com>

* fix merge commit

Signed-off-by: sharpd <davidsharp7@gmail.com>

* fix liniting

Signed-off-by: sharpd <davidsharp7@gmail.com>

* update based on pr feedback

Signed-off-by: sharpd <davidsharp7@gmail.com>

* lint tag test code

Signed-off-by: sharpd <davidsharp7@gmail.com>

* lint tag  code

Signed-off-by: sharpd <davidsharp7@gmail.com>

* fix log typo

Signed-off-by: sharpd <davidsharp7@gmail.com>

* fix logging

Signed-off-by: sharpd <davidsharp7@gmail.com>

* update based on PR feedback.

Signed-off-by: sharpd <davidsharp7@gmail.com>

* correct db field error

Signed-off-by: sharpd <davidsharp7@gmail.com>

---------

Signed-off-by: sharpd <davidsharp7@gmail.com>
Co-authored-by: Willy Lulciuc <willy@datakin.com>

Author: davidsharp7

api/src/main/java/marquez/api/DatasetResource.java

@@ -242,7 +242,7 @@ public Response tagField(
     throwIfNotExists(namespaceName, datasetName);
     throwIfNotExists(namespaceName, datasetName, fieldName);
     log.info(
-        "Tagging field '{}' for dataset '{}' with '{}'.",
+        "Tagging field '{}' on dataset '{}' with '{}'.",
         fieldName.getValue(),
         datasetName.getValue(),
         tagName.getValue());
@@ -270,7 +270,7 @@ public Response deleteTagField(
     throwIfNotExists(namespaceName, datasetName);
     throwIfNotExists(namespaceName, datasetName, fieldName);
     log.info(
-        "Deleting Tag '{}' from field '{}' on dataset '{}' in namepspace '{}'.",
+        "Deleting Tag '{}' from field '{}' on dataset '{}' in namespace '{}'.",
         tagName.getValue(),
         fieldName.getValue(),
         datasetName.getValue(),

api/src/main/java/marquez/api/JobResource.java

@@ -42,6 +42,7 @@
 import marquez.common.models.JobName;
 import marquez.common.models.NamespaceName;
 import marquez.common.models.RunId;
+import marquez.common.models.TagName;
 import marquez.common.models.Version;
 import marquez.db.JobFacetsDao;
 import marquez.db.JobVersionDao;
@@ -273,6 +274,47 @@ public Response getRunFacets(
     return Response.ok(facets).build();
   }
 
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @POST
+  @Path("/namespaces/{namespace}/jobs/{job}/tags/{tag}")
+  @Produces(APPLICATION_JSON)
+  public Response updatetag(
+      @PathParam("namespace") NamespaceName namespaceName,
+      @PathParam("job") JobName jobName,
+      @PathParam("tag") TagName tagName) {
+    throwIfNotExists(namespaceName);
+    throwIfNotExists(namespaceName, jobName);
+
+    jobService.updateJobTags(namespaceName.getValue(), jobName.getValue(), tagName.getValue());
+    Job job =
+        jobService
+            .findJobByName(namespaceName.getValue(), jobName.getValue())
+            .orElseThrow(() -> new JobNotFoundException(jobName));
+    return Response.ok(job).build();
+  }
+
+  @ResponseMetered
+  @ExceptionMetered
+  @DELETE
+  @Path("/namespaces/{namespace}/jobs/{job}/tags/{tag}")
+  @Produces(APPLICATION_JSON)
+  public Response deletetag(
+      @PathParam("namespace") NamespaceName namespaceName,
+      @PathParam("job") JobName jobName,
+      @PathParam("tag") TagName tagName) {
+    throwIfNotExists(namespaceName);
+    throwIfNotExists(namespaceName, jobName);
+
+    jobService.deleteJobTags(namespaceName.getValue(), jobName.getValue(), tagName.getValue());
+    Job job =
+        jobService
+            .findJobByName(namespaceName.getValue(), jobName.getValue())
+            .orElseThrow(() -> new JobNotFoundException(jobName));
+    return Response.ok(job).build();
+  }
+
   @Value
   static class JobVersions {
     @NonNull