#246: Implement basics of ParitioningReader

agent/src/main/scala/za/co/absa/atum/agent/AtumContext.scala

@@ -151,7 +151,7 @@ class AtumContext private[agent] (
     val retrievedAD = agent.updateAdditionalData(this.atumPartitions, currAdditionalDataSubmit)
 
     // Could be different from the one that was submitted. Replacing, just to have the most fresh copy possible.
-    this.additionalData = retrievedAD.data.map{ case (k, v) => (k, v.flatMap(_.value)) }
+    this.additionalData = retrievedAD.data.map{ case (k, v) => (k, v.map(_.value)) }
     this
   }
 

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/CapturingDispatcher.scala

@@ -125,7 +125,7 @@ class CapturingDispatcher(config: Config) extends Dispatcher(config) {
   ): AdditionalDataDTO = {
     val result = AdditionalDataDTO(
       additionalDataPatchDTO.data.map { case (key, value) =>
-        key -> Some(AdditionalDataItemDTO(Some(value), additionalDataPatchDTO.byUser))
+        key -> Some(AdditionalDataItemDTO(value, additionalDataPatchDTO.byUser))
       }
     )
 

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/ConsoleDispatcher.scala

@@ -52,7 +52,7 @@ class ConsoleDispatcher(config: Config) extends Dispatcher(config) with Logging
 
     AdditionalDataDTO(
       additionalDataPatchDTO.data.map { case (key, value) =>
-        key -> Some(AdditionalDataItemDTO(Some(value), additionalDataPatchDTO.byUser))
+        key -> Some(AdditionalDataItemDTO(value, additionalDataPatchDTO.byUser))
       }
     )
   }

agent/src/main/scala/za/co/absa/atum/agent/dispatcher/HttpDispatcher.scala

@@ -23,6 +23,7 @@ import sttp.client3._
 import sttp.model.Uri
 import sttp.client3.okhttp.OkHttpSyncBackend
 import za.co.absa.atum.agent.exception.AtumAgentException.HttpException
+import za.co.absa.atum.model.ApiPaths
 import za.co.absa.atum.model.dto._
 import za.co.absa.atum.model.envelopes.SuccessResponse.SingleSuccessResponse
 import za.co.absa.atum.model.utils.JsonSyntaxExtensions._
@@ -31,19 +32,17 @@ import za.co.absa.atum.model.utils.JsonSyntaxExtensions._
 class HttpDispatcher(config: Config) extends Dispatcher(config) with Logging {
   import HttpDispatcher._
 
-  val serverUrl: String = config.getString(UrlKey)
+  private val serverUrl: String = config.getString(UrlKey)
 
-  private val apiV1 = "/api/v1"
-  private val apiV2 = "/api/v2"
+  private val apiV1 = s"/${ApiPaths.Api}/${ApiPaths.V1}"
+  private val apiV2 = s"/${ApiPaths.Api}/${ApiPaths.V2}"
 
-  private val partitioningsPath = "partitionings"
+  private val createPartitioningEndpoint = Uri.unsafeParse(s"$serverUrl$apiV1/${ApiPaths.V1Paths.CreatePartitioning}")
+  private val createCheckpointEndpoint = Uri.unsafeParse(s"$serverUrl$apiV1/${ApiPaths.V1Paths.CreateCheckpoint}")
 
-  private val createPartitioningEndpoint = Uri.unsafeParse(s"$serverUrl$apiV1/createPartitioning")
-  private val createCheckpointEndpoint = Uri.unsafeParse(s"$serverUrl$apiV1/createCheckpoint")
-
-  private val getPartitioningIdEndpoint = Uri.unsafeParse(s"$serverUrl$apiV2/$partitioningsPath")
+  private val getPartitioningIdEndpoint = Uri.unsafeParse(s"$serverUrl$apiV2/${ApiPaths.V2Paths.Partitionings}")
   private def createAdditionalDataEndpoint(partitioningId: Long): Uri =
-    Uri.unsafeParse(s"$serverUrl$apiV2/$partitioningsPath/$partitioningId/additional-data")
+    Uri.unsafeParse(s"$serverUrl$apiV2/${ApiPaths.V2Paths.Partitionings}/$partitioningId/${ApiPaths.V2Paths.AdditionalData}")
 
   private val commonAtumRequest = basicRequest
     .header("Content-Type", "application/json")

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

@@ -20,7 +20,7 @@ import io.circe.generic.semiauto.{deriveDecoder, deriveEncoder}
 import io.circe.{Decoder, Encoder}
 
 case class AdditionalDataItemDTO(
-  value: Option[String],
+  value: String,
   author: String
 )
 

model/src/test/scala/za/co/absa/atum/model/dto/SerializationUtilsUnitTests.scala

@@ -31,8 +31,8 @@ class SerializationUtilsUnitTests extends AnyFlatSpecLike {
   "asJsonString" should "serialize AdditionalDataDTO into json string" in {
     val additionalDataDTO = AdditionalDataDTO(
       Map[String, Option[AdditionalDataItemDTO]](
-        "key1" -> Some(AdditionalDataItemDTO(Some("val1"), "testAuthor")),
-        "key2" -> Some(AdditionalDataItemDTO(Some("val2"), "testAuthor")),
+        "key1" -> Some(AdditionalDataItemDTO("val1", "testAuthor")),
+        "key2" -> Some(AdditionalDataItemDTO("val2", "testAuthor")),
         "key3" -> None
       )
     )
@@ -55,8 +55,8 @@ class SerializationUtilsUnitTests extends AnyFlatSpecLike {
 
     val expectedAdditionalDataDTO = AdditionalDataDTO(
       Map[String, Option[AdditionalDataItemDTO]](
-        "key1" -> Some(AdditionalDataItemDTO(Some("val1"), "testAuthor")),
-        "key2" -> Some(AdditionalDataItemDTO(Some("val2"), "testAuthor"))
+        "key1" -> Some(AdditionalDataItemDTO("val1", "testAuthor")),
+        "key2" -> Some(AdditionalDataItemDTO("val2", "testAuthor"))
       )
     )
     val actualAdditionalDataDTO = additionalDataDTOJson.as[AdditionalDataDTO]

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

@@ -18,24 +18,86 @@ package za.co.absa.atum.reader
 
 import sttp.client3.SttpBackend
 import sttp.monad.MonadError
+import sttp.monad.syntax._
+import za.co.absa.atum.model.ApiPaths.{Api, V2, V2Paths}
+import za.co.absa.atum.model.dto.{AdditionalDataDTO, CheckpointV2DTO}
+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.core.RequestResult.RequestResult
 import za.co.absa.atum.reader.core.{PartitioningIdProvider, Reader}
+import za.co.absa.atum.reader.requests.QueryParamNames
 import za.co.absa.atum.reader.server.ServerConfig
 
 /**
  *
  * @param partitioning  - the Atum partitions to read the information from
  * @param serverConfig  - the Atum server configuration
  * @param backend       - sttp backend, that will be executing the requests
- * @param ev            - using evidence based approach to ensure that the type F is a MonadError instead of using context
- *                      bounds, as it make the imports easier to follow
  * @tparam F            - the effect type (e.g. Future, IO, Task, etc.)
  */
-case class PartitioningReader[F[_]](partitioning: AtumPartitions)
-                                   (implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any], ev: MonadError[F])
+case class PartitioningReader[F[_]: MonadError](partitioning: AtumPartitions)
+                                   (implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any])
   extends Reader[F] with PartitioningIdProvider[F]{
-  def foo(): String = {
-    // just to have some testable content
-    "bar"
+
+  /**
+   * Function to retrieve a page of checkpoints belonging to the partitioning.
+   * The checkpoints are ordered by their creation order.
+   *
+   * @param pageSize  - the size of the page (record count) to be returned
+   * @param offset    - offset of the page (starting position)
+   * @return          - a page of checkpoints
+   */
+  def getCheckpointsPage(pageSize: Int = 10, offset: Long = 0): F[RequestResult[PaginatedResponse[CheckpointV2DTO]]] = {
+    for {
+      partitioningIdOrError <- partitioningId(partitioning)
+      checkpointsOrError <- mapRequestResultF(partitioningIdOrError, queryCheckpoints(_, None, pageSize, offset))
+    } yield checkpointsOrError
+  }
+
+  /**
+   * Function to retrieve a page of checkpoints of the given name belonging to the partitioning. (While the usual logic
+   * would suggest, there would be only one checkpoint of a name, nothing prevents to have checkpoints of the same name;
+   * also during reprocessing the checkpoints might multiply.)
+   * The checkpoints are ordered by their creation order.
+   *
+   * @param checkpointName  - the name to filter with
+   * @param pageSize        - the size of the page (record count) to be returned
+   * @param offset          - offset of the page (starting position)
+   * @return                - a page of checkpoints
+   */
+  def getCheckpointsOfNamePage(checkpointName: String, pageSize: Int = 10, offset: Long = 0): F[RequestResult[PaginatedResponse[CheckpointV2DTO]]] = {
+    for {
+      partitioningIdOrError <- partitioningId(partitioning)
+      checkpointsOrError <- mapRequestResultF(partitioningIdOrError, queryCheckpoints(_, Some(checkpointName), pageSize, offset))
+    } yield checkpointsOrError
+  }
+
+  /**
+   * Returns the additional data associated with the partitioning
+   *
+   * @return - the additional data as they were save for the partitioning
+   */
+  def getAdditionalData: F[RequestResult[AdditionalDataDTO]] = {
+    for{
+      partitioningIdOrError <- partitioningId(partitioning)
+      additionalDataOrError <- mapRequestResultF(partitioningIdOrError, queryAdditionalData)
+    } yield additionalDataOrError.map(_.data)
+  }
+
+  private def queryCheckpoints(partitioningId: Long,
+                               checkpointName: Option[String],
+                               limit: Int,
+                               offset: Long): F[RequestResult[PaginatedResponse[CheckpointV2DTO]]] = {
+    val endpoint = s"/$Api/$V2/${V2Paths.Partitionings}/$partitioningId/${V2Paths.Checkpoints}"
+    val params = Map(
+      QueryParamNames.Limit -> limit.toString,
+      QueryParamNames.Offset -> offset.toString
+    ) ++ checkpointName.map(QueryParamNames.CheckpointName -> _)
+    getQuery(endpoint, params)
+  }
+
+  private def queryAdditionalData(partitioningId: Long): F[RequestResult[SingleSuccessResponse[AdditionalDataDTO]]] = {
+    val endpoint = s"$Api/$V2/${V2Paths.Partitionings}/$partitioningId/${V2Paths.AdditionalData}"
+    getQuery(endpoint)
   }
 }