#188: new Flow service in API v2

database/src/main/postgres/flows/V1.9.1__get_flow_checkpoints.sql

@@ -15,18 +15,20 @@
  */
 
 CREATE OR REPLACE FUNCTION flows.get_flow_checkpoints(
-    IN i_partitioning_of_flow JSONB,
-    IN i_limit                INT DEFAULT 5,
-    IN i_checkpoint_name      TEXT DEFAULT NULL,
-    OUT status                INTEGER,
-    OUT status_text           TEXT,
-    OUT id_checkpoint         UUID,
-    OUT checkpoint_name       TEXT,
-    OUT measure_name          TEXT,
-    OUT measure_columns       TEXT[],
-    OUT measurement_value     JSONB,
-    OUT checkpoint_start_time TIMESTAMP WITH TIME ZONE,
-    OUT checkpoint_end_time   TIMESTAMP WITH TIME ZONE
+    IN i_partitioning_of_flow  JSONB,
+    IN i_limit                 INT DEFAULT 5,
+    IN i_checkpoint_name       TEXT DEFAULT NULL,
+    OUT status                 INTEGER,
+    OUT status_text            TEXT,
+    OUT id_checkpoint          UUID,
+    OUT checkpoint_name        TEXT,
+    OUT author                 TEXT,
+    OUT measured_by_atum_agent BOOLEAN,
+    OUT measure_name           TEXT,
+    OUT measured_columns       TEXT[],
+    OUT measurement_value      JSONB,
+    OUT checkpoint_start_time  TIMESTAMP WITH TIME ZONE,
+    OUT checkpoint_end_time    TIMESTAMP WITH TIME ZONE
 ) RETURNS SETOF record AS
 $$
 -------------------------------------------------------------------------------
@@ -53,14 +55,17 @@ $$
 --          specifying `i_checkpoint_name` parameter
 --
 -- Returns:
---      status              - Status code
---      status_text         - Status text
---      id_checkpoint       - id of retrieved checkpoint
---      checkpoint_name     - name of retrieved checkpoint
---      measure_name        - measure name associated with a given checkpoint
---      measure_columns     - measure columns associated with a given checkpoint
---      measurement_value   - measurement details associated with a given checkpoint
---      checkpoint_time     - time
+--      status                 - Status code
+--      status_text            - Status text
+--      id_checkpoint          - ID of retrieved checkpoint
+--      checkpoint_name        - Name of the retrieved checkpoint
+--      author                 - Author of the checkpoint
+--      measured_by_atum_agent - Flag indicating whether the checkpoint was measured by Atum Agent
+--                               (if false, data supplied manually)
+--      measure_name           - measure name associated with a given checkpoint
+--      measured_columns       - measure columns associated with a given checkpoint
+--      measurement_value      - measurement details associated with a given checkpoint
+--      checkpoint_time        - time
 --
 -- Status codes:
 --      11                  - OK
@@ -89,6 +94,7 @@ BEGIN
     RETURN QUERY
     SELECT 11 AS status, 'OK' AS status_text,
            CP.id_checkpoint, CP.checkpoint_name,
+           CP.created_by AS author, CP.measured_by_atum_agent,
            MD.measure_name, MD.measured_columns,
            M.measurement_value,
            CP.process_start_time AS checkpoint_start_time, CP.process_end_time AS checkpoint_end_time

database/src/test/scala/za/co/absa/atum/database/flows/GetFlowCheckpointsIntegrationTests.scala

@@ -229,12 +229,14 @@ class GetFlowCheckpointsIntegrationTests extends DBTestSuite {
         assert(row1.getString("status_text").contains("OK"))
         assert(row1.getUUID("id_checkpoint").contains(checkpointId))
         assert(row1.getString("checkpoint_name").contains("CheckpointNameCntAndAvg"))
+        assert(row1.getString("author").contains("ObviouslySomeTest"))
+        assert(row1.getBoolean("measured_by_atum_agent").contains(true))
         assert(row1.getOffsetDateTime("checkpoint_start_time").contains(startTime))
         assert(row1.getOffsetDateTime("checkpoint_end_time").contains(endTime))
 
         val measure1 = MeasuredDetails(
           row1.getString("measure_name").get,
-          row1.getArray[String]("measure_columns").map(_.toList).get,
+          row1.getArray[String]("measured_columns").map(_.toList).get,
           row1.getJsonB("measurement_value").get
         )
 
@@ -243,12 +245,14 @@ class GetFlowCheckpointsIntegrationTests extends DBTestSuite {
         assert(row2.getString("status_text").contains("OK"))
         assert(row2.getUUID("id_checkpoint").contains(checkpointId))
         assert(row2.getString("checkpoint_name").contains("CheckpointNameCntAndAvg"))
+        assert(row1.getString("author").contains("ObviouslySomeTest"))
+        assert(row1.getBoolean("measured_by_atum_agent").contains(true))
         assert(row2.getOffsetDateTime("checkpoint_start_time").contains(startTime))
         assert(row2.getOffsetDateTime("checkpoint_end_time").contains(endTime))
 
         val measure2 = MeasuredDetails(
           row2.getString("measure_name").get,
-          row2.getArray[String]("measure_columns").map(_.toList).get,
+          row2.getArray[String]("measured_columns").map(_.toList).get,
           row2.getJsonB("measurement_value").get
         )
 

model/src/main/scala/za/co/absa/atum/model/dto/CheckpointQueryDTO.scala

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.atum.model.dto
+
+case class CheckpointQueryDTO(
+  partitioning: PartitioningDTO,
+  limit: Option[Int],
+  checkpointName: Option[String]
+)

project/Dependencies.scala

@@ -28,6 +28,7 @@ object Dependencies {
     val balta = "0.1.0"
 
     val jacksonModuleScala = "2.14.2"
+    val circeVersion = "0.14.5"
 
     val specs2 = "4.10.0"
     val typesafeConfig = "1.4.2"
@@ -115,12 +116,17 @@ object Dependencies {
     lazy val json4sJackson = "org.json4s" %% "json4s-jackson" % json4sVersion
     lazy val json4sNative = "org.json4s" %% "json4s-native" % json4sVersion % Provided
 
+    lazy val circeCore = "io.circe" %% "circe-core" % Versions.circeVersion
+    lazy val circeParser = "io.circe" %% "circe-parser" % Versions.circeVersion
+
     Seq(
       jacksonModuleScala,
       json4sExt,
       json4sCore,
       json4sJackson,
-      json4sNative
+      json4sNative,
+      circeCore,
+      circeParser,
     )
   }
 
@@ -163,6 +169,7 @@ object Dependencies {
 
     // Fa-db
     lazy val faDbDoobie = faDbOrg %% "doobie" % Versions.fadb
+    lazy val pgCirceDoobie = "org.tpolecat" %% "doobie-postgres-circe" % "1.0.0-RC2"
 
     // aws
     lazy val awsSecretsManagerSdk = awsSdkOrg % "secretsmanager" % Versions.awssdk
@@ -175,6 +182,7 @@ object Dependencies {
 
     Seq(
       faDbDoobie,
+      pgCirceDoobie,
       zioCore,
       zioMacros,
       zioLogging,
@@ -244,7 +252,7 @@ object Dependencies {
       jsonSerdeDependencies(scalaVersion)
   }
 
- def databaseDependencies: Seq[ModuleID] = {
+  def databaseDependencies: Seq[ModuleID] = {
     lazy val scalaTest  = "org.scalatest"   %% "scalatest"  % Versions.scalatest  % Test
     lazy val balta =      "za.co.absa"      %% "balta"      % Versions.balta      % Test
 

server/src/main/scala/za/co/absa/atum/server/Constants.scala

@@ -22,10 +22,12 @@ object Constants {
 
     final val Api = "api"
     final val V1 = "v1"
+    final val V2 = "v2"
 
     final val CreatePartitioning = "createPartitioning"
     final val CreateOrUpdateAdditionalData = "createOrUpdateAdditionalData"
     final val CreateCheckpoint = "createCheckpoint"
+    final val GetFlowCheckpoints = "getFlowCheckpoints"
 
     final val Health = "health"
     final val ZioMetrics = "zio-metrics"

server/src/main/scala/za/co/absa/atum/server/Main.scala

@@ -17,11 +17,12 @@
 package za.co.absa.atum.server
 
 import za.co.absa.atum.server.api.controller._
+import za.co.absa.atum.server.api.database.flows.functions.GetFlowCheckpoints
 import za.co.absa.atum.server.api.database.{PostgresDatabaseProvider, TransactorProvider}
 import za.co.absa.atum.server.api.database.runs.functions._
 import za.co.absa.atum.server.api.http.Server
-import za.co.absa.atum.server.api.repository.{CheckpointRepositoryImpl, PartitioningRepositoryImpl}
-import za.co.absa.atum.server.api.service.{CheckpointServiceImpl, PartitioningServiceImpl}
+import za.co.absa.atum.server.api.repository.{CheckpointRepositoryImpl, FlowRepositoryImpl, PartitioningRepositoryImpl}
+import za.co.absa.atum.server.api.service.{CheckpointServiceImpl, FlowServiceImpl, PartitioningServiceImpl}
 import za.co.absa.atum.server.aws.AwsSecretsProviderImpl
 import za.co.absa.atum.server.config.JvmMonitoringConfig
 import zio._
@@ -42,15 +43,19 @@ object Main extends ZIOAppDefault with Server {
         .provide(
           PartitioningControllerImpl.layer,
           CheckpointControllerImpl.layer,
+          FlowControllerImpl.layer,
           PartitioningServiceImpl.layer,
           CheckpointServiceImpl.layer,
+          FlowServiceImpl.layer,
           PartitioningRepositoryImpl.layer,
           CheckpointRepositoryImpl.layer,
+          FlowRepositoryImpl.layer,
           CreatePartitioningIfNotExists.layer,
           GetPartitioningMeasures.layer,
           GetPartitioningAdditionalData.layer,
           CreateOrUpdateAdditionalData.layer,
           WriteCheckpoint.layer,
+          GetFlowCheckpoints.layer,
           PostgresDatabaseProvider.layer,
           TransactorProvider.layer,
           AwsSecretsProviderImpl.layer,

server/src/main/scala/za/co/absa/atum/server/api/controller/BaseController.scala

@@ -23,6 +23,21 @@ import zio._
 
 trait BaseController {
 
+  def serviceCall[A, B](
+    serviceCall: IO[ServiceError, A],
+    onSuccessFnc: A => B
+  ): IO[ErrorResponse, B] = {
+
+    serviceCall
+      .mapError { serviceError: ServiceError =>
+        InternalServerErrorResponse(serviceError.message)
+      }
+      .flatMap {
+        result => ZIO.succeed(onSuccessFnc(result))
+      }
+
+  }
+
   def serviceCallWithStatus[A, B](
     serviceCall: IO[ServiceError, Either[StatusException, A]],
     onSuccessFnc: A => B

server/src/main/scala/za/co/absa/atum/server/api/controller/FlowController.scala

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.atum.server.api.controller
+
+import za.co.absa.atum.model.dto.{CheckpointQueryDTO, CheckpointDTO}
+import za.co.absa.atum.server.model.ErrorResponse
+import zio.IO
+import zio.macros.accessible
+
+@accessible
+trait FlowController {
+  def getFlowCheckpoints(checkpointQueryDTO: CheckpointQueryDTO): IO[ErrorResponse, Seq[CheckpointDTO]]
+}

server/src/main/scala/za/co/absa/atum/server/api/controller/FlowControllerImpl.scala

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.atum.server.api.controller
+
+import za.co.absa.atum.model.dto.{CheckpointQueryDTO, CheckpointDTO}
+import za.co.absa.atum.server.api.service.FlowService
+import za.co.absa.atum.server.model.ErrorResponse
+import zio._
+
+class FlowControllerImpl(flowService: FlowService)
+  extends FlowController with BaseController {
+
+  override def getFlowCheckpoints(checkpointQueryDTO: CheckpointQueryDTO): IO[ErrorResponse, Seq[CheckpointDTO]] = {
+    serviceCall[Seq[CheckpointDTO], Seq[CheckpointDTO]](
+      flowService.getFlowCheckpoints(checkpointQueryDTO),
+      r => r
+    )
+  }
+
+}
+
+object FlowControllerImpl {
+  val layer: URLayer[FlowService, FlowController] = ZLayer {
+    for {
+      flowService <- ZIO.service[FlowService]
+    } yield new FlowControllerImpl(flowService)
+  }
+}

server/src/main/scala/za/co/absa/atum/server/api/database/DoobieImplicits.scala

@@ -18,12 +18,12 @@ package za.co.absa.atum.server.api.database
 
 import cats.Show
 import cats.data.NonEmptyList
-
-import doobie.{Get, Put}
 import doobie.postgres.implicits._
-
+import doobie.{Get, Put}
+import io.circe.{Decoder, Encoder}
 import org.postgresql.jdbc.PgArray
 import org.postgresql.util.PGobject
+import za.co.absa.atum.model.dto.MeasureResultDTO
 
 import scala.util.{Failure, Success, Try}
 
@@ -155,7 +155,29 @@ object DoobieImplicits {
           }
         )
     }
+  }
 
+  implicit val encodeResultValueType: Encoder[MeasureResultDTO.ResultValueType] = Encoder.encodeString.contramap {
+    case MeasureResultDTO.ResultValueType.String      => "String"
+    case MeasureResultDTO.ResultValueType.Long        => "Long"
+    case MeasureResultDTO.ResultValueType.BigDecimal  => "BigDecimal"
+    case MeasureResultDTO.ResultValueType.Double      => "Double"
   }
 
+  implicit val decodeResultValueType: Decoder[MeasureResultDTO.ResultValueType] = Decoder.decodeString.emap {
+    case "String"     => Right(MeasureResultDTO.ResultValueType.String)
+    case "Long"       => Right(MeasureResultDTO.ResultValueType.Long)
+    case "BigDecimal" => Right(MeasureResultDTO.ResultValueType.BigDecimal)
+    case "Double"     => Right(MeasureResultDTO.ResultValueType.Double)
+    case other        => Left(s"Cannot decode $other as ResultValueType")
+  }
+
+  implicit val encodeTypedValue: Encoder[MeasureResultDTO.TypedValue] =
+    Encoder.forProduct2("value", "valueType")(tv => (tv.value, tv.valueType))
+
+  implicit val decodeTypedValue: Decoder[MeasureResultDTO.TypedValue] =
+    Decoder.forProduct2("value", "valueType")(MeasureResultDTO.TypedValue.apply)
+
+  implicit val decodeMeasureResultDTO: Decoder[MeasureResultDTO] =
+    Decoder.forProduct2("mainValue", "supportValues")(MeasureResultDTO.apply)
 }

server/src/main/scala/za/co/absa/atum/server/api/database/flows/Flows.scala

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.atum.server.api.database.flows
+
+import za.co.absa.fadb.DBSchema
+import za.co.absa.fadb.naming.implementations.SnakeCaseNaming.Implicits._
+
+object Flows extends DBSchema