Merge branch 'feature/246-implement-basics-of-paritioningreader' of https://github.com/AbsaOSS/atum-service into feature/246-implement-basics-of-paritioningreader

Author: benedeki

.github/workflows/jacoco_report.yml

@@ -24,7 +24,7 @@ on:
 env:
   scalaLong: 2.13.11
   scalaShort: "2.13"
-  coverage-overall: 58.0
+  coverage-overall: 56.0
   coverage-changed-files: 80.0
   check-overall-coverages: true
 

build.sbt

@@ -136,8 +136,7 @@ lazy val reader = (projectMatrix in file("reader"))
   .settings(
     Setup.commonSettings ++ Seq(
       name := "atum-reader",
-      javacOptions ++= Setup.clientJavacOptions,
-      publish / skip := true // module is not yet finished, so we don't want to publish it
+      javacOptions ++= Setup.clientJavacOptions
     ): _*
   )
   .addScalaCrossBuild(Setup.clientSupportedScalaVersions, Dependencies.readerDependencies)

database/src/main/postgres/runs/V0.3.0.11__alter_checkpoints.ddl

@@ -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.
+ */
+
+ UPDATE runs.checkpoints
+ SET measured_by_atum_agent = FALSE
+ WHERE measured_by_atum_agent IS NULL;
+
+ALTER TABLE runs.checkpoints
+    ALTER COLUMN measured_by_atum_agent SET DEFAULT FALSE,
+    ALTER COLUMN measured_by_atum_agent SET NOT NULL;

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

@@ -36,8 +36,8 @@ import za.co.absa.atum.reader.server.ServerConfig
  * @param backend              - sttp backend, that will be executing the requests
  * @tparam F                   - the effect type (e.g. Future, IO, Task, etc.)
  */
-class FlowReader[F[_]: MonadError](val mainFlowPartitioning: AtumPartitions)
-                      (implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any])
+case class FlowReader[F[_]: MonadError](mainFlowPartitioning: AtumPartitions)
+                                       (implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any])
   extends Reader[F] with PartitioningIdProvider[F] {
 
   /**
@@ -50,8 +50,8 @@ class FlowReader[F[_]: MonadError](val mainFlowPartitioning: AtumPartitions)
    */
   def getCheckpointsPage(pageSize: Int = 10, offset: Long = 0): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = {
     for {
-      mainPartitioningIdOrErrror <- partitioningId(mainFlowPartitioning)
-      flowIdOrError <- mapRequestResultF(mainPartitioningIdOrErrror, queryFlowId)
+      mainPartitioningIdOrError <- partitioningId(mainFlowPartitioning)
+      flowIdOrError <- mapRequestResultF(mainPartitioningIdOrError, queryFlowId)
       checkpointsOrError <- mapRequestResultF(flowIdOrError, queryCheckpoints(_, None, pageSize, offset))
     } yield checkpointsOrError
   }
@@ -67,10 +67,10 @@ class FlowReader[F[_]: MonadError](val mainFlowPartitioning: AtumPartitions)
    */
   def getCheckpointsOfNamePage(checkpointName: String, pageSize: Int = 10, offset: Long = 0): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = {
     for {
-      mainPartitioningId <- partitioningId(mainFlowPartitioning)
-      flowId <- mapRequestResultF(mainPartitioningId, queryFlowId)
-      checkpoints <- mapRequestResultF(flowId, queryCheckpoints(_, Some(checkpointName), pageSize, offset))
-    } yield checkpoints
+      mainPartitioningIdOrError <- partitioningId(mainFlowPartitioning)
+      flowIdOrError <- mapRequestResultF(mainPartitioningIdOrError, queryFlowId)
+      checkpointsOrError <- mapRequestResultF(flowIdOrError, queryCheckpoints(_, Some(checkpointName), pageSize, offset))
+    } yield checkpointsOrError
   }
 
   private def queryFlowId(mainPartitioningId: Long): F[RequestResult[Long]] = {

reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala

@@ -117,7 +117,7 @@ class FlowReaderUnitTests extends AnyFunSuiteLike {
       requestId = UUID.fromString("29ce91a7-b668-41d2-a160-26402551fb0b")
     )
 
-    val reader = new FlowReader(atumPartitions)
+    val reader = FlowReader(atumPartitions)
     val result = reader.getCheckpointsPage()
     assert(result == Right(expectedData))
   }
@@ -190,7 +190,7 @@ class FlowReaderUnitTests extends AnyFunSuiteLike {
       requestId = UUID.fromString("29ce91a7-b668-41d2-a160-26402551fb0b")
     )
 
-    val reader = new FlowReader(atumPartitions)
+    val reader = FlowReader(atumPartitions)
     val result = reader.getCheckpointsOfNamePage("Test checkpoints 1")
     assert(result == Right(expectedData))
   }