#247 Implement basics of FlowReader

.github/workflows/test_filenames_check.yml

@@ -42,6 +42,7 @@ jobs:
             server/src/test/scala/za/co/absa/atum/server/api/TestData.scala,
             server/src/test/scala/za/co/absa/atum/server/api/TestTransactorProvider.scala,
             server/src/test/scala/za/co/absa/atum/server/ConfigProviderTest.scala,
-            model/src/test/scala/za/co/absa/atum/testing/*
+            model/src/test/scala/za/co/absa/atum/testing/*,
+            reader/src/test/scala/za/co/absa/atum/testing/*
           verbose-logging: 'false'
           fail-on-violation: 'true'

adrs/01_Basics-of-FlowReader-and-PartitioningReader.drawio

@@ -0,0 +1,28 @@
+<mxfile host="app.diagrams.net" agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:135.0) Gecko/20100101 Firefox/135.0" version="24.8.6">
+  <diagram name="Page-1" id="M1M2r3vxqz2qx0Wid1ZL">
+    <mxGraphModel dx="1434" dy="733" grid="1" gridSize="10" guides="1" tooltips="1" connect="1" arrows="1" fold="1" page="1" pageScale="1" pageWidth="827" pageHeight="1169" math="0" shadow="0">
+      <root>
+        <mxCell id="0" />
+        <mxCell id="1" parent="0" />
+        <mxCell id="rmWetNVkiCoe2Ea992-S-1" value="PartitioningReader" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="440" y="40" width="120" height="60" as="geometry" />
+        </mxCell>
+        <mxCell id="rmWetNVkiCoe2Ea992-S-2" value="FlowReader" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="60" y="40" width="120" height="60" as="geometry" />
+        </mxCell>
+        <mxCell id="rmWetNVkiCoe2Ea992-S-4" value="&lt;div align=&quot;left&quot;&gt;&lt;ul&gt;&lt;li&gt;&lt;b&gt;Identified by the main paritioining&lt;/b&gt;&lt;/li&gt;&lt;li&gt;ability to retrieve &lt;i&gt;id_flow&lt;/i&gt; based on the main partitioing &lt;br&gt;&lt;/li&gt;&lt;li&gt;Getting the checkpoints belonging to the flow&lt;/li&gt;&lt;li&gt;Getting checkpoints of certain name only&lt;/li&gt;&lt;li&gt;Getting it&#39;s partitioning (phase 2)&lt;/li&gt;&lt;/ul&gt;&lt;/div&gt;" style="rounded=0;whiteSpace=wrap;html=1;align=left;" vertex="1" parent="1">
+          <mxGeometry x="80" y="110" width="240" height="160" as="geometry" />
+        </mxCell>
+        <mxCell id="rmWetNVkiCoe2Ea992-S-5" value="&lt;div align=&quot;left&quot;&gt;Data returned&lt;ul&gt;&lt;li&gt;Checkpoints including partitioiing info in form of DTOs&lt;/li&gt;&lt;li&gt;returns just the dto as pagination&lt;/li&gt;&lt;/ul&gt;&lt;/div&gt;" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="80" y="310" width="250" height="130" as="geometry" />
+        </mxCell>
+        <mxCell id="rmWetNVkiCoe2Ea992-S-7" value="&lt;blockquote&gt;&lt;div align=&quot;left&quot;&gt;&lt;li&gt;&lt;b&gt;Identified by the paritioining&lt;/b&gt;&lt;/li&gt;&lt;/div&gt;&lt;div align=&quot;left&quot;&gt;&lt;li&gt;ability to retrieve &lt;i&gt;id_partitioing&lt;/i&gt; based on the partitioing &lt;/li&gt;&lt;li&gt;Getting the checkpoints&lt;/li&gt;&lt;li&gt;Getting checkpoint(s) filtered by name&lt;/li&gt;&lt;li&gt;Getting the additional data&lt;/li&gt;&lt;li&gt;Getting the checkpoints + additional data in form of _INFO file format (phase 2)&lt;/li&gt;&lt;/div&gt;&lt;/blockquote&gt;" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="440" y="110" width="240" height="160" as="geometry" />
+        </mxCell>
+        <mxCell id="rmWetNVkiCoe2Ea992-S-8" value="&lt;div align=&quot;left&quot;&gt;Data returned&lt;ul&gt;&lt;li&gt;Checkpoints in the form of DTO in Pagination DTO&lt;/li&gt;&lt;li&gt;additional data as their DTO&lt;/li&gt;&lt;/ul&gt;&lt;/div&gt;" style="rounded=0;whiteSpace=wrap;html=1;" vertex="1" parent="1">
+          <mxGeometry x="440" y="320" width="240" height="120" as="geometry" />
+        </mxCell>
+      </root>
+    </mxGraphModel>
+  </diagram>
+</mxfile>

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/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,8 +18,15 @@ package za.co.absa.atum.reader
 
 import sttp.client3.SttpBackend
 import sttp.monad.MonadError
+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
-import za.co.absa.atum.reader.basic.{PartitioningIdProvider, Reader}
+import za.co.absa.atum.reader.core.RequestResult.RequestResult
+import za.co.absa.atum.model.ApiPaths._
+import za.co.absa.atum.reader.core.{PartitioningIdProvider, Reader}
+import za.co.absa.atum.reader.implicits.EitherImplicits.EitherMonadEnhancements
+import za.co.absa.atum.reader.requests.QueryParamNames
 import za.co.absa.atum.reader.server.ServerConfig
 
 /**
@@ -35,4 +42,40 @@ class FlowReader[F[_]](val mainFlowPartitioning: AtumPartitions)
                       (implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any], ev: MonadError[F])
   extends Reader[F] with PartitioningIdProvider[F]{
 
+  private def queryFlowId(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],
+                               limit: Int,
+                               offset: Long): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = {
+    val endpoint = s"/$Api/$V2/${V2Paths.Flows}/$flowId/${V2Paths.Checkpoints}"
+    val params = Map(
+      QueryParamNames.limit -> limit.toString,
+      QueryParamNames.offset -> offset.toString
+    ) ++ checkpointName.map(QueryParamNames.checkpointName -> _)
+    getQuery(endpoint, params)
+  }
+
+  def getCheckpointsPage(pageSize: Int = 10, offset: Long = 0): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = {
+    for {
+      mainPartitioningIdOrErrror <- partitioningId(mainFlowPartitioning)
+      flowIdOrError <- mainPartitioningIdOrErrror.project(queryFlowId)
+      checkpointsOrError <- flowIdOrError.project(queryCheckpoints(_, None, pageSize, offset))
+    } yield checkpointsOrError
+  }
+
+  def getCheckpointsOfNamePage(checkpointName: String, pageSize: Int = 10, offset: Long = 0): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = {
+    for {
+      mainPartitioningId <- partitioningId(mainFlowPartitioning)
+      flowId <- mainPartitioningId.project(queryFlowId)
+      checkpoints <- flowId.project(queryCheckpoints(_, Some(checkpointName), pageSize, offset))
+    } yield checkpoints
+  }
+
 }

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

@@ -19,7 +19,7 @@ package za.co.absa.atum.reader
 import sttp.client3.SttpBackend
 import sttp.monad.MonadError
 import za.co.absa.atum.model.types.basic.AtumPartitions
-import za.co.absa.atum.reader.basic.{PartitioningIdProvider, Reader}
+import za.co.absa.atum.reader.core.{PartitioningIdProvider, Reader}
 import za.co.absa.atum.reader.server.ServerConfig
 
 /**

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

@@ -14,22 +14,23 @@
  * limitations under the License.
  */
 
-package za.co.absa.atum.reader.basic
+package za.co.absa.atum.reader.core
 
 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
 import za.co.absa.atum.model.types.basic.AtumPartitionsOps
 import za.co.absa.atum.model.utils.JsonSyntaxExtensions.JsonSerializationSyntax
-import za.co.absa.atum.reader.basic.RequestResult.RequestResult
+import 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 → reader/src/main/scala/za/co/absa/atum/reader/core/Reader.scala

@@ -14,16 +14,17 @@
  * limitations under the License.
  */
 
-package za.co.absa.atum.reader.basic
+package za.co.absa.atum.reader.core
 
 import io.circe.Decoder
 import sttp.client3.{Identity, RequestT, ResponseException, SttpBackend, basicRequest}
 import sttp.client3.circe.asJson
 import sttp.model.Uri
 import sttp.monad.MonadError
 import sttp.monad.syntax._
+import za.co.absa.atum.reader.core.RequestResult._
 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 → reader/src/main/scala/za/co/absa/atum/reader/core/RequestResult.scala

@@ -14,25 +14,31 @@
  * limitations under the License.
  */
 
-package za.co.absa.atum.reader.basic
+package za.co.absa.atum.reader.core
 
 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.RequestException
 
 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)
       }
     }
   }
+
 }

reader/src/main/scala/za/co/absa/atum/reader/exceptions/ReaderException.scala

@@ -0,0 +1,19 @@
+/*
+ * 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.reader.exceptions
+
+class ReaderException(message: String) extends Exception(message)

reader/src/main/scala/za/co/absa/atum/reader/exceptions/RequestException.scala

@@ -0,0 +1,45 @@
+/*
+ * 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.reader.exceptions
+
+import sttp.model.{StatusCode, Uri}
+import za.co.absa.atum.model.envelopes.ErrorResponse
+
+sealed abstract class RequestException(message: String) extends ReaderException(message)
+
+
+object RequestException {
+  type CirceError = io.circe.Error
+
+  final case class HttpException(
+                                  message: String,
+                                  statusCode: StatusCode,
+                                  errorResponse: ErrorResponse,
+                                  request: Uri
+                                ) extends RequestException(message)
+
+  final case class ParsingException(
+                                     message: String,
+                                     body: String
+                                    ) extends RequestException(message)
+  object ParsingException {
+    def fromCirceError(error: CirceError, body: String): ParsingException = {
+      ParsingException(error.getMessage, body)
+    }
+  }
+
+}

reader/src/main/scala/za/co/absa/atum/reader/implicits/EitherImplicits.scala

@@ -0,0 +1,30 @@
+/*
+ * 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.reader.implicits
+
+import sttp.monad.MonadError
+
+object EitherImplicits {
+
+  implicit class EitherMonadEnhancements[A, B](val either: Either[A, B]) extends AnyVal {
+    def project[C, F[_]: MonadError](f: B => F[Either[A, C]]): F[Either[A, C]] = either match {
+      case Right(b) => f(b)
+      case Left(a) => implicitly[MonadError[F]].unit(Left(a))
+    }
+  }
+
+}

reader/src/main/scala/za/co/absa/atum/reader/requests/QueryParamNames.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.reader.requests
+
+object QueryParamNames {
+  val limit = "limit"
+  val offset = "offset"
+  val checkpointName = "checkpoint-name"
+}