#245 Add the ability to query REST endpoints from Reader module

.github/workflows/test_filenames_check.yml

@@ -39,6 +39,7 @@ jobs:
           excludes: |
             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
+            server/src/test/scala/za/co/absa/atum/server/ConfigProviderTest.scala,
+            model/src/test/scala/za/co/absa/atum/model/testing/*
           verbose-logging: 'false'
           fail-on-violation: 'true'

README.md

@@ -15,6 +15,8 @@
         - [Measurement](#measurement)
         - [Checkpoint](#checkpoint)
         - [Data Flow](#data-flow)
+    - [Usage](#usage)
+        - [Reader](#reader-usage)
     - [How to generate Code coverage report](#how-to-generate-code-coverage-report)
     - [How to Run in IntelliJ](#how-to-run-in-intellij)
     - [How to Run Tests](#how-to-run-tests)
@@ -156,6 +158,65 @@ We can even say, that `Checkpoint` is a result of particular `Measurements` (ver
 The journey of a dataset throughout various data transformations and pipelines. It captures the whole journey,
 even if it involves multiple applications or ETL pipelines.
 
+## Usage
+
+### Reader usage
+Reader module support several asynchronous http clients. The dependencies used for these clients are set as _optional_,
+so the user of the module can decide which client to use and include only the necessary dependencies. 
+
+The clients are:
+
+[//]: # (TODO #298 needs Java 11 cross-build)
+
+[//]: # (#### Future based `HttpClientServerConnection`)
+
+[//]: # (Uses `java.net.http.HttpClient` to send requests to the server, therefore requires no additional dependencies. But works )
+
+[//]: # (only with Java 11 or higher. )
+
+#### Future based `ArmeririaServerConnection`
+Add 
+```scala
+"com.softwaremill.sttp.client3" %% "armeria-backend" % "[version]"
+```
+to your dependencies.
+
+#### Cats IO based `ArmeririaServerConnection`
+Add
+```scala
+"org.typelevel." %% "cats-effect" % "[version]"
+"com.softwaremill.sttp.client3" %% "armeria-backend-cats" % "[version]" // for cats-effect 3.x
+// or
+"com.softwaremill.sttp.client3" %% "armeria-backend-cats-ce2" % "[version]" // for cats-effect 2.x
+```
+"
+to your dependencies.
+
+[//]: # (TODO #298 needs Java 11 cross-build)
+
+[//]: # (#### ZIO based `HttpClientServerConnection`)
+
+[//]: # (Add)
+
+[//]: # (```scala)
+
+[//]: # ("com.softwaremill.sttp.client3" %% "zio" % "[version]"  // for ZIO 2.x)
+
+[//]: # ("com.softwaremill.sttp.client3" %% "zio1" % "[version]" // for ZIO 1.x)
+
+[//]: # (```)
+
+[//]: # (to your dependencies.)
+
+#### ZIO based `ArmeririaServerConnection`
+Add
+```scala
+"com.softwaremill.sttp.client3" %% "armeria-backend-zio" % "[version]"  // for ZIO 2.x
+"com.softwaremill.sttp.client3" %% "armeria-backend-zio1" % "[version]" // for ZIO 1.x
+```
+to your dependencies.
+
+
 
 ## How to generate Code coverage report
 ```sbt
@@ -172,6 +233,7 @@ Code coverage wil be generated on path:
 To make this project runnable via IntelliJ, do the following:
 - Make sure that your configuration in `server/src/main/resources/reference.conf` 
   is configured according to your needs
+- When building within the UI be sure to have the option `-language:higherKinds` on in the compiler options
 
 ## How to Run Tests
 

agent/README.md

@@ -7,26 +7,28 @@
 
 ## Usage
 
-Create multiple `AtumContext` with different control measures to be applied 
+Create multiple `AtumContext` with different control measures to be applied
 
 ### Option 1
 ```scala
 val atumContextInstanceWithRecordCount = AtumContext(processor = processor)
+  .withMeasureAdded(RecordCount(MockMeasureNames.recordCount1, controlCol = "id"))
   .withMeasureAdded(RecordCount(MockMeasureNames.recordCount1, measuredColumn = "id"))
 
 val atumContextWithSalaryAbsMeasure = atumContextInstanceWithRecordCount
+  .withMeasureAdded(AbsSumOfValuesOfColumn(controlCol = "salary"))
   .withMeasureAdded(AbsSumOfValuesOfColumn(measuredColumn = "salary"))
 ```
 
-### Option 2 
+### Option 2
 Use `AtumPartitions` to get an `AtumContext` from the service using the `AtumAgent`.
 ```scala
     val atumContext1 = AtumAgent.createAtumContext(atumPartition)
 ```
 
 #### AtumPartitions
-A list of key values that maintains the order of arrival of the items, the `AtumService` 
-is able to deliver the correct `AtumContext` according to the `AtumPartitions` we give it. 
+A list of key values that maintains the order of arrival of the items, the `AtumService`
+is able to deliver the correct `AtumContext` according to the `AtumPartitions` we give it.
 ```scala
     val atumPartitions = AtumPartitions().withPartitions(ListMap("name" -> "partition-name", "country" -> "SA", "gender" -> "female" ))
 

database/src/main/postgres/runs/get_measurements.sql

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2024 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.
+ */
+
+CREATE OR REPLACE FUNCTION postgres.runs.get_measurements(
+    IN i_parameter TEXT,
+    OUT status INTEGER,
+    OUT status_text TEXT
+) RETURNS record AS
+$$
+    -------------------------------------------------------------------------------
+--
+-- Function: postgres.runs.get_measurements([Function_Param_Count])
+--      [Description]
+--
+-- Parameters:
+--      i_parameter         -
+--
+-- Returns:
+--      status              - Status code
+--      status_text         - Status text
+--
+-- Status codes:
+--      10                  - OK
+--
+-------------------------------------------------------------------------------
+DECLARE
+BEGIN
+
+END;
+$$
+    LANGUAGE plpgsql VOLATILE
+                     SECURITY DEFINER;
+
+GRANT EXECUTE ON FUNCTION postgres.runs.get_measurements() TO [user];

model/src/test/scala/za/co/absa/atum/model/testing/implicits/StringImplicits.scala

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2024 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.testing.implicits
+
+object StringImplicits {
+  implicit class StringLinearization(val str: String) extends AnyVal {
+    def linearize: String = {
+      str.stripMargin.replace("\r", "").replace("\n", "")
+    }
+  }
+}

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

@@ -21,7 +21,7 @@ import za.co.absa.atum.model.ResultValueType
 import za.co.absa.atum.model.dto.MeasureResultDTO.TypedValue
 import za.co.absa.atum.model.dto._
 import za.co.absa.atum.model.utils.JsonSyntaxExtensions._
-import za.co.absa.atum.model.utils.SerializationUtilsTest.StringLinearization
+import za.co.absa.atum.model.testing.implicits.StringImplicits.StringLinearization
 
 import java.time.{ZoneId, ZoneOffset, ZonedDateTime}
 import java.util.UUID
@@ -436,11 +436,3 @@ class SerializationUtilsUnitTests extends AnyFlatSpecLike {
   }
 
 }
-
-object SerializationUtilsTest {
-  implicit class StringLinearization(val str: String) extends AnyVal {
-    def linearize: String = {
-      str.stripMargin.replace("\r", "").replace("\n", "")
-    }
-  }
-}

project/Dependencies.scala

@@ -38,7 +38,7 @@ object Dependencies {
 
     val sparkCommons = "0.6.1"
 
-    val sttpClient = "3.5.2"
+    val sttpClient = "3.5.2" //last supported version for Java 8
     val sttpCirceJson = "3.9.7"
 
     val postgresql = "42.6.0"
@@ -64,6 +64,8 @@ object Dependencies {
 
     val absaCommons = "2.0.0"
 
+    val catsEffect = "3.3.14"
+
     def truncateVersion(version: String, parts: Int): String = {
       version.split("\\.").take(parts).mkString(".")
     }
@@ -90,9 +92,9 @@ object Dependencies {
   private def jsonSerdeDependencies: Seq[ModuleID] = {
 
     // Circe dependencies
-    lazy val circeCore = "io.circe" %% "circe-core" % Versions.circeJson
-    lazy val circeParser = "io.circe" %% "circe-parser" % Versions.circeJson
-    lazy val circeGeneric = "io.circe" %% "circe-generic" % Versions.circeJson
+    val circeCore = "io.circe" %% "circe-core" % Versions.circeJson
+    val circeParser = "io.circe" %% "circe-parser" % Versions.circeJson
+    val circeGeneric = "io.circe" %% "circe-generic" % Versions.circeJson
 
     Seq(
       circeCore,
@@ -236,9 +238,46 @@ object Dependencies {
   }
 
   def readerDependencies(scalaVersion: Version): Seq[ModuleID] = {
+    val zioOrg = "dev.zio"
+    val sbtOrg = "com.github.sbt"
+    val sttpClient3Org = "com.softwaremill.sttp.client3"
+    val typeLevelOrg = "org.typelevel"
+
+    // STTP core and Circe integration
+    lazy val sttpCore = sttpClient3Org %% "core" % Versions.sttpClient
+    lazy val sttpCirce = sttpClient3Org %% "circe" % Versions.sttpClient
+
+    // Armeria Future backend
+    lazy val sttpArmeririaFutureBackend = sttpClient3Org %% "armeria-backend" % Versions.sttpClient % Optional
+    // Armeria Cats backend
+    lazy val sttpArmeririaCatsBackend = sttpClient3Org %% "armeria-backend-cats" % Versions.sttpClient % Optional
+    lazy val catsEffect = typeLevelOrg %% "cats-effect" % Versions.catsEffect % Optional
+    // Armeria Zio backend
+    lazy val sttpArmeririaZioBackend = sttpClient3Org %% "armeria-backend-zio" % Versions.sttpClient % Optional
+    // HttpClient Zio backend
+//    lazy val sttpHttpClientZioBackend = sttpClient3Org %% "zio" % Versions.sttpClient % Optional TODO #298 needs Java 11 cross-build
+
+    // testing
+    lazy val zioTest = zioOrg %% "zio-test" % Versions.zio % Test
+    lazy val zioTestSbt = zioOrg %% "zio-test-sbt" % Versions.zio % Test
+    lazy val zioTestJunit = zioOrg %% "zio-test-junit" % Versions.zio % Test
+    lazy val sbtJunitInterface = sbtOrg % "junit-interface" % Versions.sbtJunitInterface % Test
+
     Seq(
+      sttpCore,
+      sttpCirce,
+      sttpArmeririaFutureBackend,
+      sttpArmeririaCatsBackend,
+      catsEffect,
+      sttpArmeririaZioBackend,
+//      sttpHttpClientZioBackend, TODO #298 needs Java 11 cross-build
+      zioTest,
+      zioTestSbt,
+      zioTestJunit,
+      sbtJunitInterface
     ) ++
-      testDependencies
+      testDependencies ++
+      jsonSerdeDependencies
   }
 
   def databaseDependencies: Seq[ModuleID] = {

project/Setup.scala

@@ -40,17 +40,37 @@ object Setup {
   val serverAndDbScalaVersion: Version = scala213 //covers REST server and database modules
   val clientSupportedScalaVersions: Seq[Version] = Seq(scala212, scala213)
 
-  val commonScalacOptions: Seq[String] = Seq("-unchecked", "-deprecation", "-feature", "-Xfatal-warnings")
+  val commonScalacOptions: Seq[String] = Seq(
+    "-unchecked",
+    "-deprecation",
+    "-feature",
+    "-Xfatal-warnings"
+  )
 
-  val serverAndDbJavacOptions: Seq[String] = Seq("-source", "11", "-target", "11", "-Xlint")
-  val serverAndDbScalacOptions: Seq[String] = Seq("-Ymacro-annotations")
+  val serverAndDbJavacOptions: Seq[String] = Seq(
+    "-source", "11",
+    "-target", "11",
+    "-Xlint"
+  )
+  val serverAndDbScalacOptions: Seq[String] = Seq(
+    "-language:higherKinds",
+    "-Ymacro-annotations"
+  )
 
   val clientJavacOptions: Seq[String] = Seq("-source", "1.8", "-target", "1.8", "-Xlint")
   def clientScalacOptions(scalaVersion: Version): Seq[String] = {
     if (scalaVersion >= scala213) {
-      Seq("-release", "8", "-Ymacro-annotations")
+      Seq(
+        "-release", "8",
+        "-language:higherKinds",
+        "-Ymacro-annotations"
+      )
     } else {
-      Seq("-release", "8", "-target:8")
+      Seq(
+        "-release", "8",
+        "-language:higherKinds",
+        "-target:8"
+      )
     }
   }
 

project/VersionAxes.scala

@@ -32,6 +32,11 @@ object VersionAxes {
     override val idSuffix: String = directorySuffix.replaceAll("""\W+""", "_")
   }
 
+  case class JavaVersionAxis(javaVersion: String) extends sbt.VirtualAxis.WeakAxis {
+    override val directorySuffix = s"-jdk$javaVersion"
+    override val idSuffix: String = directorySuffix.replaceAll("""\W+""", "_")
+  }
+
   private def camelCaseToLowerDashCase(origName: String): String = {
     origName
       .replaceAll("([A-Z])", "-$1")

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

@@ -16,7 +16,10 @@
 
 package za.co.absa.atum.reader
 
-class FlowReader {
+import za.co.absa.atum.reader.basic.Reader
+import za.co.absa.atum.reader.server.GenericServerConnection
+
+class FlowReader[F[_]]()(override implicit val serverConnection: GenericServerConnection[F]) extends Reader[F]{
   def foo(): String = {
     // just to have some testable content
     "bar"

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

@@ -16,7 +16,10 @@
 
 package za.co.absa.atum.reader
 
-class PartitioningReader {
+import za.co.absa.atum.reader.basic.Reader
+import za.co.absa.atum.reader.server.GenericServerConnection
+
+class PartitioningReader[F[_]]()(override implicit val serverConnection: GenericServerConnection[F]) extends Reader[F] {
   def foo(): String = {
     // just to have some testable content
     "bar"

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

@@ -0,0 +1,21 @@
+/*
+ * 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.basic
+
+import za.co.absa.atum.reader.server.GenericServerConnection
+
+abstract class Reader[F[_]](implicit val serverConnection: GenericServerConnection[F])