#247 Implement basics of FlowReader

server/src/main/scala/za/co/absa/atum/server/api/http/ApiPaths.scala → model/src/main/scala/za/co/absa/atum/model/ApiPaths.scala

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.atum.server.api.http
+package za.co.absa.atum.model
 
 object ApiPaths {
 

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

@@ -18,19 +18,20 @@ package za.co.absa.atum.model.dto
 
 import io.circe.{Decoder, Encoder}
 import io.circe.generic.semiauto.{deriveDecoder, deriveEncoder}
+import za.co.absa.atum.model.dto.traits.CheckpointCore
 
 import java.time.ZonedDateTime
 import java.util.UUID
 
-case class CheckpointV2DTO(
+case class CheckpointV2DTO (
   id: UUID,
   name: String,
   author: String,
   measuredByAtumAgent: Boolean = false,
   processStartTime: ZonedDateTime,
   processEndTime: Option[ZonedDateTime],
   measurements: Set[MeasurementDTO]
-)
+) extends CheckpointCore
 
 object CheckpointV2DTO {
   implicit val decodeCheckpointDTO: Decoder[CheckpointV2DTO] = deriveDecoder

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

@@ -18,6 +18,7 @@ package za.co.absa.atum.model.dto
 
 import io.circe.{Decoder, Encoder}
 import io.circe.generic.semiauto.{deriveDecoder, deriveEncoder}
+import za.co.absa.atum.model.dto.traits.CheckpointCore
 
 import java.time.ZonedDateTime
 import java.util.UUID
@@ -31,7 +32,7 @@ case class CheckpointWithPartitioningDTO(
   processEndTime: Option[ZonedDateTime],
   measurements: Set[MeasurementDTO],
   partitioning: PartitioningWithIdDTO
-)
+) extends CheckpointCore
 
 object CheckpointWithPartitioningDTO {
 

model/src/main/scala/za/co/absa/atum/model/dto/traits/CheckpointCore.scala

@@ -0,0 +1,32 @@
+/*
+ * 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.traits
+
+import za.co.absa.atum.model.dto.MeasurementDTO
+
+import java.time.ZonedDateTime
+import java.util.UUID
+
+trait CheckpointCore {
+  def id: UUID
+  def name: String
+  def author: String
+  def measuredByAtumAgent: Boolean
+  def processStartTime: ZonedDateTime
+  def processEndTime: Option[ZonedDateTime]
+  def measurements: Set[MeasurementDTO]
+}

model/src/main/scala/za/co/absa/atum/model/types/AtumPartitionsCheckpoint.scala

@@ -0,0 +1,24 @@
+/*
+ * 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.types
+
+import za.co.absa.atum.model.types.basic.AtumPartitions
+
+case class AtumPartitionsCheckpoint(
+                                     partitioning: AtumPartitions,
+                                     checkpoint: Checkpoint
+                                  )

model/src/main/scala/za/co/absa/atum/model/types/Checkpoint.scala

@@ -0,0 +1,46 @@
+/*
+ * 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.types
+
+import za.co.absa.atum.model.dto.traits.CheckpointCore
+
+import java.time.ZonedDateTime
+import java.util.UUID
+
+case class Checkpoint (
+                        id: UUID,
+                        name: String,
+                        author: String,
+                        measuredByAtumAgent: Boolean = false,
+                        processStartTime: ZonedDateTime,
+                        processEndTime: Option[ZonedDateTime],
+                        measurements: Set[Measurement]
+                      )
+
+object Checkpoint {
+  def apply(from: CheckpointCore): Checkpoint = {
+    new Checkpoint(
+      id = from.id,
+      name = from.name,
+      author = from.author,
+      measuredByAtumAgent = from.measuredByAtumAgent,
+      processStartTime = from.processStartTime,
+      processEndTime = from.processEndTime,
+      measurements = from.measurements.map(Measurement(_))
+    )
+  }
+}

model/src/main/scala/za/co/absa/atum/model/types/Measurement.scala

@@ -0,0 +1,82 @@
+/*
+ * 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.types
+
+import za.co.absa.atum.model.ResultValueType
+import za.co.absa.atum.model.dto.MeasurementDTO
+
+trait Measurement {
+  type T
+  def measureName: String
+  def measuredColumns: Seq[String]
+  def valueType: ResultValueType
+  def value: T
+  def stringValue: String
+}
+
+object Measurement {
+
+  def apply[T](from: MeasurementDTO): Measurement = {
+    from.result.mainValue.valueType match {
+      case ResultValueType.StringValue => StringMeasurement(from.measure.measureName, from.measure.measuredColumns, from.result.mainValue.value)
+      case ResultValueType.LongValue => LongMeasurement(from.measure.measureName, from.measure.measuredColumns, from.result.mainValue.value.toLong)
+      case ResultValueType.BigDecimalValue => BigDecimalMeasurement(from.measure.measureName, from.measure.measuredColumns, BigDecimal(from.result.mainValue.value))
+      case ResultValueType.DoubleValue => DoubleMeasurement(from.measure.measureName, from.measure.measuredColumns, from.result.mainValue.value.toDouble)
+    }
+  }
+
+  case class StringMeasurement(
+                                measureName: String,
+                                measuredColumns: Seq[String],
+                                value: String
+                              ) extends Measurement {
+    override type T = String
+    override def valueType: ResultValueType = ResultValueType.StringValue
+    override def stringValue: String = value
+  }
+
+  case class LongMeasurement(
+    measureName: String,
+    measuredColumns: Seq[String],
+    value: Long
+  ) extends Measurement {
+    override type T = Long
+    override def valueType: ResultValueType = ResultValueType.LongValue
+    override def stringValue: String = value.toString
+  }
+
+  case class BigDecimalMeasurement(
+    measureName: String,
+    measuredColumns: Seq[String],
+    value: BigDecimal
+  ) extends Measurement {
+    override type T = BigDecimal
+    override def valueType: ResultValueType = ResultValueType.BigDecimalValue
+    override def stringValue: String = value.toString
+  }
+
+  case class DoubleMeasurement(
+    measureName: String,
+    measuredColumns: Seq[String],
+    value: Double
+  ) extends Measurement {
+    override type T = Double
+    override def valueType: ResultValueType = ResultValueType.DoubleValue
+    override def stringValue: String = value.toString
+  }
+
+}

model/src/test/scala/za/co/absa/atum/model/types/AtumPartitionsUnitTests.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 ABSA Group Limited
+ * 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.

model/src/test/scala/za/co/absa/atum/model/utils/JsonDeserializationSyntaxUnitTests.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 ABSA Group Limited
+ * 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.

model/src/test/scala/za/co/absa/atum/model/utils/JsonSerializationSyntaxUnitTests.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 ABSA Group Limited
+ * 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.

reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala

@@ -18,9 +18,20 @@ package za.co.absa.atum.reader
 
 import sttp.client3.SttpBackend
 import sttp.monad.MonadError
-import za.co.absa.atum.model.types.basic.AtumPartitions
+import sttp.monad.syntax._
+import za.co.absa.atum.model.dto.{CheckpointWithPartitioningDTO, FlowDTO}
+import za.co.absa.atum.model.envelopes.SuccessResponse.{PaginatedResponse, SingleSuccessResponse}
+import za.co.absa.atum.model.types.basic.{AtumPartitions, PartitioningDTOOps}
+import za.co.absa.atum.reader.basic.RequestResult.{RequestPageResultOps, RequestResult}
 import za.co.absa.atum.reader.basic.{PartitioningIdProvider, Reader}
+import za.co.absa.atum.model.ApiPaths._
+import za.co.absa.atum.model.types.{AtumPartitionsCheckpoint, Checkpoint}
+import za.co.absa.atum.reader.implicits.PaginatedResponseImplicits.PaginatedResponseMonadEnhancements
+import za.co.absa.atum.reader.implicits.EitherImplicits.EitherMonadEnhancements
+import za.co.absa.atum.reader.implicits.PaginatedResponseImplicits
+import za.co.absa.atum.reader.result.Page
 import za.co.absa.atum.reader.server.ServerConfig
+import za.co.absa.atum.reader.result.Page.PageRoller
 
 /**
  * This class is a reader that reads data tight to a flow.
@@ -35,4 +46,47 @@ class FlowReader[F[_]](val mainFlowPartitioning: AtumPartitions)
                       (implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any], ev: MonadError[F])
   extends Reader[F] with PartitioningIdProvider[F]{
 
+  private def flowId(mainPartitioningId: Long): F[RequestResult[Long]] = {
+    val endpoint = s"/$Api/$V2/${V2Paths.Partitionings}/$mainPartitioningId/${V2Paths.MainFlow}"
+    val queryResult = getQuery[SingleSuccessResponse[FlowDTO]](endpoint)
+    queryResult.map{ result =>
+      result.map(_.data.id)
+    }
+  }
+
+  private def queryCheckpoints(flowId: Long,
+                               checkpointName: Option[String],
+                               pageSize: Int,
+                               offset: Long): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = {
+    val endpoint = s"/$Api/$V2/${V2Paths.Flows}/$flowId/${V2Paths.Checkpoints}"
+    val params = Map(
+      "limit" -> pageSize.toString,
+      "offset" -> offset.toString
+    ) ++ checkpointName.map("checkpoint-name" -> _)
+    getQuery(endpoint, params)
+  }
+
+  private def geetCheckpointDTOs(checkpointName: Option[String], pageSize: Int = 10, offset: Long = 0): F[RequestResult[Page[CheckpointWithPartitioningDTO, F]]] = {
+    val pageRoller: PageRoller[CheckpointWithPartitioningDTO, F] = geetCheckpointDTOs(checkpointName, _, _)
+
+    for {
+      mainPartitioningId <- partitioningId(mainFlowPartitioning)
+      flowId <- mainPartitioningId.project(flowId)
+      checkpoints <- flowId.project(queryCheckpoints(_, checkpointName, pageSize, offset))
+    } yield checkpoints.map(_.toPage(pageRoller))
+
+  }
+
+  def getCheckpoints(pageSize: Int = 10, offset: Long = 0): F[RequestResult[Page[AtumPartitionsCheckpoint, F]]] = {
+    def checkpointMapper(data: CheckpointWithPartitioningDTO): AtumPartitionsCheckpoint = {
+      val atumPartitions = data.partitioning.partitioning.toAtumPartitions
+      val checkpoint = Checkpoint(data)
+      AtumPartitionsCheckpoint(atumPartitions, checkpoint)
+    }
+    geetCheckpointDTOs(None, pageSize, offset).map(_.pageMap((checkpointMapper)))
+  }
+
+  def getCheckpointsOfName(name: String, pageSize: Int = 10, offset: Int = 0): F[RequestResult[Page[CheckpointWithPartitioningDTO, F]]] = {
+    geetCheckpointDTOs(Some(name), pageSize, offset)
+  }
 }

reader/src/main/scala/za/co/absa/atum/reader/basic/PartitioningIdProvider.scala

@@ -18,6 +18,7 @@ package za.co.absa.atum.reader.basic
 
 import sttp.monad.MonadError
 import sttp.monad.syntax._
+import za.co.absa.atum.model.ApiPaths._
 import za.co.absa.atum.model.dto.PartitioningWithIdDTO
 import za.co.absa.atum.model.envelopes.SuccessResponse.SingleSuccessResponse
 import za.co.absa.atum.model.types.basic.AtumPartitions
@@ -28,8 +29,8 @@ import za.co.absa.atum.reader.basic.RequestResult.RequestResult
 trait PartitioningIdProvider[F[_]] {self: Reader[F] =>
   def partitioningId(partitioning: AtumPartitions)(implicit monad: MonadError[F]): F[RequestResult[Long]] = {
     val encodedPartitioning = partitioning.toPartitioningDTO.asBase64EncodedJsonString
-    val queryResult = getQuery[SingleSuccessResponse[PartitioningWithIdDTO]]("/api/v2/partitionings", Map("partitioning" -> encodedPartitioning))
-    queryResult.map{result =>
+    val queryResult = getQuery[SingleSuccessResponse[PartitioningWithIdDTO]](s"/$Api/$V2/${V2Paths.Partitionings}", Map("partitioning" -> encodedPartitioning))
+    queryResult.map{ result =>
       result.map(_.data.id)
     }
   }

reader/src/main/scala/za/co/absa/atum/reader/basic/Reader.scala

@@ -24,6 +24,7 @@ import sttp.monad.MonadError
 import sttp.monad.syntax._
 import za.co.absa.atum.reader.server.ServerConfig
 import za.co.absa.atum.reader.basic.RequestResult._
+import za.co.absa.atum.reader.exceptions.RequestException.CirceError
 
 /**
  * Reader is a base class for reading data from a remote server.

reader/src/main/scala/za/co/absa/atum/reader/basic/RequestResult.scala

@@ -17,22 +17,33 @@
 package za.co.absa.atum.reader.basic
 
 import sttp.client3.{DeserializationException, HttpError, Response, ResponseException}
+import sttp.monad.MonadError
 import za.co.absa.atum.model.envelopes.ErrorResponse
+import za.co.absa.atum.reader.exceptions.RequestException.{CirceError, HttpException, ParsingException}
+import za.co.absa.atum.reader.exceptions.{ReaderException, RequestException}
+import za.co.absa.atum.reader.result.{GroupedPage, Page}
 
 object RequestResult {
-  type CirceError = io.circe.Error
-  type RequestResult[R] = Either[ResponseException[ErrorResponse, CirceError], R]
+  type RequestResult[R] = Either[RequestException, R]
+
+  def RequestOK[T](value: T): RequestResult[T] = Right(value)
+  def RequestFail[T](error: RequestException): RequestResult[T] = Left(error)
 
   implicit class ResponseOps[R](val response: Response[Either[ResponseException[String, CirceError], R]]) extends AnyVal {
     def toRequestResult: RequestResult[R] = {
       response.body.left.map {
         case he: HttpError[String] =>
           ErrorResponse.basedOnStatusCode(he.statusCode.code, he.body) match {
-            case Right(er) => HttpError(er, he.statusCode)
-            case Left(ce)  => DeserializationException(he.body, ce)
+            case Right(er) => HttpException(he.getMessage, he.statusCode, er, response.request.uri)
+            case Left(ce)  => ParsingException.fromCirceError(ce, he.body)
           }
-        case de: DeserializationException[CirceError] => de
+        case de: DeserializationException[CirceError] => ParsingException.fromCirceError(de.error, de.body)
       }
     }
   }
+
+  implicit class RequestPageResultOps[A, F[_]: MonadError](requestResult: RequestResult[Page[A, F]]) {
+    def pageMap[B](f: A => B): RequestResult[Page[B, F]] = requestResult.map(_.map(f))
+  }
+
 }