#247 Implement basics of FlowReader #306

benedeki · 2024-12-11T12:55:16Z

FlowReader methods to read checkpoints - this is the main focus of the whole PR
ApiPaths moved from_Server_ to Model
Checkpoint and Measurement classes in Model to detach user from (versioned) DTOs
Other classes moved from Agent to Model_, as they are used in Reader too.

Depends on #316
Closes #247

* created new module Info * the new modul added to JaCoco and CI routines

* JaCoCo exclusion for model

* created Provider to query the data from server * support for Future, IO, and ZIO based providers * work in progress

* fixed license headers

…endpoints-from-info-module

lsulak · 2024-12-12T11:07:45Z

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


yeah something inside me says that this is not really really a 'model' but in general I'm okay with this; finding a better name for 'model' is difficult and it's backward incompatible change; alternatively, creating a new model for this seems a bit too much, so I prefer your solution

I consider the model as the library to make communication with server "easy". So IMHO it fits, without too much mind-bending 😉

lsulak · 2024-12-12T11:17:26Z

model/src/main/scala/za/co/absa/atum/model/dto/traits/CheckpointCore.scala

+ * limitations under the License.
+ */
+
+package za.co.absa.atum.model.dto.traits


hmm I'm not sure I like the location of this file; the trait/ dir has only this 1 file which doesn't really do much to our service in terms of better 'modularity' / 'structure of the DTO module. Besides, the fact that it's trait can be seen pretty nicely in IntelliJ, as it's visually different (green trait icon)

In fact, I don't really see a big value in doing this at all to be fair. I mean, what if CheckpointV3 will be different? Then we can't do this or will update the trait..I mean it seems potentially limiting without too much benefits.

Also you did it only for checkpoints, not Partitioning DTOs; to me it really seem unncessary hierarchy

The trait is a solution for inability/discouragement of case class inheritance. Because CheckpointWithPartitioningDTO is by definition an enhanced CheckpointDTO. And there's a a point, where I want to be able to handle both with the same function.

Package traits is debatable of course. Naming... 😉

lsulak · 2024-12-12T11:24:46Z

model/src/main/scala/za/co/absa/atum/model/types/Measurement.scala

+import za.co.absa.atum.model.dto.MeasurementDTO
+
+trait Measurement {
+  type T


Suggested change

type T

type V

just a suggestion

lsulak · 2024-12-12T11:29:17Z

model/src/main/scala/za/co/absa/atum/model/types/Measurement.scala

+
+object Measurement {
+
+  def apply[T](from: MeasurementDTO): Measurement = {


this has some similarity with agent's model/MeasureResult - did you check if it can be reused for Reader usages? Maybe slightly changed/moved if needed. But I think if model and agent are doing conceptually similar thing maybe we can unite and reuse some of the existing code

The similarity did occur to me. But they are not 100% same unlike AtumPartitions and AdditionalData and did want to further delay in search for commonality. I think that can be done in future, if found useful.

lsulak · 2024-12-12T14:42:20Z

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

+import za.co.absa.atum.model.types.{AtumPartitionsCheckpoint, Checkpoint}
+import za.co.absa.atum.reader.implicits.PaginatedResponseImplicits.PaginatedResponseMonadEnhancements
+import za.co.absa.atum.reader.implicits.EitherImplicits.EitherMonadEnhancements
+import za.co.absa.atum.reader.implicits.PaginatedResponseImplicits


this one is not needed due to line 29

lsulak · 2024-12-12T14:45:41Z

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

+    getQuery(endpoint, params)
+  }
+
+  private def geetCheckpointDTOs(checkpointName: Option[String], pageSize: Int = 10, offset: Long = 0): F[RequestResult[Page[CheckpointWithPartitioningDTO, F]]] = {


Suggested change

private def geetCheckpointDTOs(checkpointName: Option[String], pageSize: Int = 10, offset: Long = 0): F[RequestResult[Page[CheckpointWithPartitioningDTO, F]]] = {

private def getCheckpointDTOs(checkpointName: Option[String], pageSize: Int = 10, offset: Long = 0): F[RequestResult[Page[CheckpointWithPartitioningDTO, F]]] = {

lsulak · 2024-12-12T14:47:28Z

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

+      val checkpoint = Checkpoint(data)
+      AtumPartitionsCheckpoint(atumPartitions, checkpoint)
+    }
+    geetCheckpointDTOs(None, pageSize, offset).map(_.pageMap((checkpointMapper)))


Suggested change

geetCheckpointDTOs(None, pageSize, offset).map(_.pageMap((checkpointMapper)))

geetCheckpointDTOs(None, pageSize, offset).map(_.pageMap(checkpointMapper))

lsulak · 2024-12-12T14:56:32Z

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

 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.{ReaderException, RequestException}


ReaderException unused

lsulak · 2024-12-12T14:56:41Z

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

 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.{ReaderException, RequestException}
+import za.co.absa.atum.reader.result.{GroupedPage, Page}


GroupedPage unused

lsulak · 2024-12-12T15:12:48Z

model/src/main/scala/za/co/absa/atum/model/types/AtumPartitionsCheckpoint.scala

+
+import za.co.absa.atum.model.types.basic.AtumPartitions
+
+case class AtumPartitionsCheckpoint(


also possibly reader-specific?

lsulak · 2024-12-12T15:21:37Z

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

@@ -18,6 +18,7 @@ package za.co.absa.atum.reader.basic

 import sttp.monad.MonadError


Consider renaming package from basic to core ? just an idea

* Finished UTs

benedeki · 2025-02-05T01:47:15Z

This item depends on:

#247: Paging support in Reader module #316
By Dependent Issues.

…t-basics-of-flowreader

lsulak · 2025-02-17T15:49:30Z

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

+    val params = Map(
+      "limit" -> pageSize.toString,
+      "offset" -> offset.toString
+    ) ++ checkpointName.map("checkpoint-name" -> _)


I would consider to extract this into some sort of a constant, but I don't insist - not sure if this is gonna be used elsewhere in the future

lsulak · 2025-02-17T15:52:20Z

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

+
+    for {
+      mainPartitioningId <- partitioningId(mainFlowPartitioning)
+      flowId <- mainPartitioningId.project(queryFlowId)


for the sake of better readability, I'd consider to reflect either types in the name - for example here it could be flowIdOrError or so

same below etc - all of these 'wrapper' types are a bit easier to read and maintain truth

lsulak · 2025-02-17T15:58:47Z

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

+    for {
+      mainPartitioningId <- partitioningId(mainFlowPartitioning)
+      flowId <- mainPartitioningId.project(queryFlowId)
+      checkpoints <- flowId.project(queryCheckpoints(_, checkpointName, pageSize, offset))


wow, I find this insanely complicated / advanced :D

lsulak · 2025-02-17T16:08:37Z

reader/src/test/scala/za/co/absa/atum/reader/implicits/PaginatedResponseImplicitsTest.scala

+      IdMonad.unit(RequestFail(new RequestException("Not used"){}))
+    }
+
+    val source = PaginatedResponse(Seq(1, 2, 3),


the formatting is really wild here :D

seriously, sometimes it worsens code readability as it's a bit unpredictable / misleading

lsulak · 2025-02-17T16:10:49Z

reader/src/test/scala/za/co/absa/atum/reader/implicits/PaginatedResponseImplicitsTest.scala

+    assert(result.items == Vector(1, 2, 3))
+    assert(!result.hasNext)
+    assert(result.limit == 3)
+    assert(result.pageStart == 1)


consider adding test on pageEnd as well

just to increase test coverage more, computation seems correct

lsulak · 2025-02-17T16:16:57Z

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

+
+  private val partitioningEncoded = "W3sia2V5IjoiYSIsInZhbHVlIjoiYiJ9LHsia2V5IjoiYyIsInZhbHVlIjoiZCJ9XQ=="
+
+  private val partitioningResponse =


consider to extract the test data to another file

lsulak · 2025-02-17T16:20:39Z

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

+          processEndTime = Some(ZonedDateTime.parse("2024-12-30T16:01:36.5052109+01:00[Europe/Budapest]")),
+          measurements = Set(
+            LongMeasurement(
+


unnecessary empty line

lsulak · 2025-02-17T16:22:13Z

reader/src/test/scala/za/co/absa/atum/reader/implicits/PaginatedResponseImplicitsTest.scala

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2024 ABSA Group Limited


2021 - there might be more occurrences like this, but I didn't check

benedeki and others added 30 commits October 31, 2024 01:23

#244: Create the Info module

b97b603

* created new module Info * the new modul added to JaCoco and CI routines

* fixed License headers

e623974

* renamed to _Reader_

5e4eadb

* README.md update

2e1e2ea

* fix

738c904

* JaCoCO action update

df8c9bd

* added dummy code for testing coverage

5affd82

* erroneous class renamed

0f1e121

* JaCoCo exclusion for model

* Deleted wrong files

d773a93

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

0776f9c

* created Provider to query the data from server * support for Future, IO, and ZIO based providers * work in progress

* Work still in progress

38fde1c

* the first working commit

1ac2233

* Removed temporary notes

e6dcb52

* introduced MonadError into the GenericServerConnection

6968b02

* fixed license headers

* Fixed UTs

b9bacef

* trying to get rid of Java 11 dependency

bbb1e7f

* Downgraded sttpClient

33e6628

* further downgrade

f7ced56

* Removed exceptions

ca2116b

* commented out parts of README.md which are not yet part of the code

e5e6f63

- major rework

fe07272

* doc fix

7656f6f

Merge branch 'master' into feature/245-add-the-ability-to-query-rest-…

eb9a678

…endpoints-from-info-module

* disabled failing test

7641c07

* adjustments

bc82a5b

- further cleaning

0e7675e

* tests progress

432716a

* several UTs added

11b0a16

Merge branch 'master' into feature/245-add-the-ability-to-query-rest-…

2c3f145

…endpoints-from-info-module

* last improvements before PR ready

e07dffb

lsulak reviewed Dec 12, 2024

View reviewed changes

* Finished implementation

e63a2e4

* Finished UTs

benedeki changed the base branch from master to feature/247-paging-in-reader February 4, 2025 11:29

benedeki marked this pull request as ready for review February 4, 2025 11:31

benedeki requested review from Zejnilovic, dk1844, salamonpavel and ABLL526 as code owners February 4, 2025 11:31

benedeki added the dependent The item depends on some other open item (Issue or PR) label Feb 5, 2025

benedeki added 2 commits February 5, 2025 09:00

Merge branch 'feature/247-paging-in-reader' into feature/247-implemen…

f70143e

…t-basics-of-flowreader

* uncommented endpoint for Swagger documentation creation

f803ea3

lsulak reviewed Feb 17, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

#247 Implement basics of FlowReader #306

#247 Implement basics of FlowReader #306

benedeki commented Dec 11, 2024 •

edited

Loading

lsulak Dec 12, 2024

benedeki Feb 4, 2025

lsulak Dec 12, 2024 •

edited

Loading

lsulak Dec 12, 2024 •

edited

Loading

benedeki Dec 13, 2024

benedeki Dec 13, 2024

lsulak Dec 12, 2024

lsulak Dec 12, 2024

benedeki Dec 13, 2024

lsulak Dec 12, 2024

lsulak Dec 12, 2024

lsulak Dec 12, 2024

lsulak Dec 12, 2024

lsulak Dec 12, 2024

lsulak Dec 12, 2024

lsulak Dec 12, 2024

benedeki commented Feb 5, 2025

lsulak Feb 17, 2025 •

edited

Loading

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025


		object Measurement {

		def apply[T](from: MeasurementDTO): Measurement = {

	private def geetCheckpointDTOs(checkpointName: Option[String], pageSize: Int = 10, offset: Long = 0): F[RequestResult[Page[CheckpointWithPartitioningDTO, F]]] = {
	private def getCheckpointDTOs(checkpointName: Option[String], pageSize: Int = 10, offset: Long = 0): F[RequestResult[Page[CheckpointWithPartitioningDTO, F]]] = {

	geetCheckpointDTOs(None, pageSize, offset).map(_.pageMap((checkpointMapper)))
	geetCheckpointDTOs(None, pageSize, offset).map(_.pageMap(checkpointMapper))


		import za.co.absa.atum.model.types.basic.AtumPartitions

		case class AtumPartitionsCheckpoint(

		@@ -18,6 +18,7 @@ package za.co.absa.atum.reader.basic

		import sttp.monad.MonadError


		private val partitioningEncoded = "W3sia2V5IjoiYSIsInZhbHVlIjoiYiJ9LHsia2V5IjoiYyIsInZhbHVlIjoiZCJ9XQ=="

		private val partitioningResponse =

#247 Implement basics of FlowReader #306

Are you sure you want to change the base?

#247 Implement basics of FlowReader #306

Conversation

benedeki commented Dec 11, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lsulak Dec 12, 2024 • edited Loading

Choose a reason for hiding this comment

lsulak Dec 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benedeki commented Feb 5, 2025

lsulak Feb 17, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benedeki commented Dec 11, 2024 •

edited

Loading

lsulak Dec 12, 2024 •

edited

Loading

lsulak Dec 12, 2024 •

edited

Loading

lsulak Feb 17, 2025 •

edited

Loading