#247: Paging support in Reader module

.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'

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

@@ -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/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,7 +14,7 @@
  * 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._
@@ -23,7 +23,7 @@ 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]] = {

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,36 @@
  * 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
+import za.co.absa.atum.reader.result.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))
+  }
+
 }

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,49 @@
+/*
+ * 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
+
+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)
+    }
+  }
+
+
+  final case class NoDataException(
+                                    message: String
+                                  ) extends RequestException(message)
+}

reader/src/main/scala/za/co/absa/atum/reader/result/AbstractPage.scala

@@ -0,0 +1,31 @@
+/*
+ * 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.result
+
+import sttp.monad.MonadError
+
+abstract class AbstractPage [T <: Iterable[_], F[_]: MonadError] {
+    def items: T
+    def hasNext: Boolean
+    def limit: Int
+    def pageStart: Long
+    def pageEnd: Long
+
+    def pageSize: Int = (pageEnd - pageStart).toInt + 1
+    def hasPrior: Boolean = pageStart > 0
+}
+

reader/src/main/scala/za/co/absa/atum/reader/result/GroupedPage.scala

@@ -0,0 +1,93 @@
+/*
+ * 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.result
+
+import sttp.monad.MonadError
+import sttp.monad.syntax._
+import za.co.absa.atum.reader.core.RequestResult.{RequestFail, RequestResult}
+import za.co.absa.atum.reader.exceptions.RequestException.NoDataException
+import za.co.absa.atum.reader.result.GroupedPage.GroupPageRoller
+
+import scala.collection.immutable.ListMap
+
+case class GroupedPage[K, V, F[_]: MonadError](
+                                                items: ListMap[K, Vector[V]],
+                                                hasNext: Boolean,
+                                                limit: Int,
+                                                pageStart: Long,
+                                                pageEnd: Long,
+                                                private[reader] val pageRoller: GroupPageRoller[K, V, F]
+                                    ) extends AbstractPage[Map[K, Vector[V]], F] {
+
+  def apply(key: K): Vector[V] = items(key)
+  def keys: Iterable[K] = items.keys
+  def groupCount: Int = items.size
+
+  def map[K1, V1](f: ((K, Vector[V])) => (K1, Vector[V1])): GroupedPage[K1, V1, F] = {
+    val newItems = items.map(f)
+    val newPageRoller: GroupPageRoller[K1, V1, F] = (limit, offset) => pageRoller(limit, offset).map(_.map(_.map(f)))
+    this.copy(items = newItems, pageRoller = newPageRoller)
+  }
+
+  def mapValues[B](f: V => B): GroupedPage[K, B, F] = {
+    def mapper(item: (K, Vector[V])): (K, Vector[B]) = (item._1, item._2.map(f))
+
+    val newItems = items.map(mapper)
+    val newPageRoller: GroupPageRoller[K, B, F] = (limit, offset) => pageRoller(limit, offset).map(_.map(_.mapValues(f)))
+    this.copy(items = newItems, pageRoller = newPageRoller)
+
+  }
+
+  def prior(newPageSize: Int): F[RequestResult[GroupedPage[K, V, F]]] = {
+    if (hasPrior) {
+      val newOffset = (pageStart - limit).max(0)
+      pageRoller(newPageSize, newOffset)
+    } else {
+      MonadError[F].unit(RequestFail(NoDataException("No prior page")))
+    }
+  }
+
+  def prior: F[RequestResult[GroupedPage[K, V, F]]] = prior(limit)
+
+  def next(newPageSize: Int): F[RequestResult[GroupedPage[K, V, F]]] = {
+    if (hasNext) {
+      pageRoller(newPageSize, pageStart + limit)
+    } else {
+      MonadError[F].unit(RequestFail(NoDataException("No next page")))
+    }
+  }
+
+  def next: F[RequestResult[GroupedPage[K, V, F]]] = next(limit)
+
+  def +(other: GroupedPage[K, V, F]): GroupedPage[K, V, F] = {
+    val newItems = other.items.foldLeft(items) { case (acc, (k, v)) =>
+      if (acc.contains(k)) {
+        acc.updated(k, acc(k) ++ v)
+      } else {
+        acc + (k -> v)
+      }
+    }
+    val newHasNext = hasNext && other.hasNext
+    val newPageStart = pageStart min other.pageStart
+    val newPageEnd = pageEnd max other.pageEnd
+    this.copy(items = newItems, hasNext = newHasNext, pageStart = newPageStart, pageEnd = newPageEnd)
+  }
+}
+
+object GroupedPage {
+  type GroupPageRoller[K, V, F[_]] = (Int, Long) => F[RequestResult[GroupedPage[K, V, F]]]
+}