#247 Implement basics of FlowReader #306

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

FlowReadeis now a simple class to just retrieve checkpoints of the identified flow.
Checkpoints are in the form of DTO without any extra functionality
Checkpoints are requested and returned in the form of pages with page offset and its size as parameters
ApiPaths moved from_Server_ to Model
Some minor renames in package names basic -> core
Fixes in License header years

Closes #247

Release notes:

FlowReader implemented to return the flow's checkpoints (requested and returned in a paginated form)

* 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

…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

No reuse, and not that many. Imho OK to be here.

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

* addressed PR comments

salamonpavel · 2025-02-21T15:41:02Z

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

+    }
+  }
+
+  private def queryCheckpoints(flowId: Long,


scalafmt formatting hasn't been applied

I changed it, but the difference was an empty line, one missing and one surplus white space
I though, the formatting was off making it harder to read. This is hardly the case.
Adhering strictly to scalafmt prescribed format is not the adopted team policy.
(same in the other cases bellow)

salamonpavel · 2025-02-21T15:44:34Z

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

+import za.co.absa.atum.model.ApiPaths._
+import za.co.absa.atum.reader.core.{PartitioningIdProvider, Reader}
+import za.co.absa.atum.reader.implicits.EitherImplicits.EitherMonadEnhancements
+import za.co.absa.atum.reader.requests.QueryParamNames
 import za.co.absa.atum.reader.server.ServerConfig

 /**


I would prefer this

class FlowReader[F[_]: MonadError](val mainFlowPartitioning: AtumPartitions)(implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any], ) extends Reader[F]

over this given the implicit monaderror ev isn't used explicitly

class FlowReader[F[_]](val mainFlowPartitioning: AtumPartitions)(implicit serverConfig: ServerConfig, backend: SttpBackend[F, Any], ev: MonadError[F] ) extends Reader[F]

There was a reason why it was done like this. but maybe with the cut code, that might have dissappeared (I think it was needed for Page creation). I will doublecheck.

salamonpavel · 2025-02-21T15:57:35Z

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

+    getQuery(endpoint, params)
+  }
+
+  def getCheckpointsPage(pageSize: Int = 10, offset: Long = 0): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = {


(For consideration) I would prefer this over the "project". I think it's not even a good name for such mapping operation that lifts the computation into F context.

def getCheckpointsPage( pageSize: Int = 10, offset: Long = 0 ): F[RequestResult[PaginatedResponse[CheckpointWithPartitioningDTO]]] = { partitioningId(mainFlowPartitioning).flatMap { case Left(error) => MonadError[F].unit(Left(error)) case Right(mainPartitioningId) => queryFlowId(mainPartitioningId).flatMap { case Left(error) => MonadError[F].unit(Left(error)) case Right(flowId) => queryCheckpoints(flowId, None, pageSize, offset) } } }

Also, the name of the method could be improved. And the "partitioningId" method could be renamed too.

OK, I get that the project might not be a best name.
But honestly the suggested code seems to much less understandable than the original for-comprehension. What do you see as the benefit in this form?

I guess mainly the project name bothers me, I agree that from readability stand point it's better than nested flatMaps or nested for-comprehensions. How about naming the method liftMapF or sth like that instead? Btw you don't need to explicitly use implicitly :D as there is apply method on the MonadError that does exactly that ...

object EitherImplicits { implicit class EitherMonadEnhancements[A, B](val either: Either[A, B]) extends AnyVal { def liftMapF[C, F[_]: MonadError](f: B => F[Either[A, C]]): F[Either[A, C]] = either match { case Right(b) => f(b) // case Left(a) => implicitly[MonadError[F]].unit(Left(a)) case Left(a) => MonadError[F].unit(Left(a)) } } } object MonadError { def apply[F[_]: MonadError]: MonadError[F] = implicitly[MonadError[F]] }

salamonpavel · 2025-02-21T16:01:19Z

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

 import za.co.absa.atum.model.dto.PartitioningWithIdDTO
 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] =>


scalafmt formatting hasn't been applied

salamonpavel · 2025-02-21T16:05:26Z

reader/src/main/scala/za/co/absa/atum/reader/exceptions/RequestException.scala

+sealed abstract class RequestException(message: String) extends ReaderException(message)
+
+
+object RequestException {


scalafmt formatting hasn't been applied

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

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

* Licenses fix

afd64a6

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

benedeki added 2 commits February 21, 2025 12:29

* complete refactoring

1a56f3d

* added adr

bbed1f0

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

benedeki added 2 commits February 21, 2025 12:50

* removing unnecessary changes

15dd12a

* removed unnecessary changes II

c630fa0

* addressed PR comments

benedeki removed work in progress Work on this item is not yet finished (mainly intended for PRs) dependent The item depends on some other open item (Issue or PR) labels Feb 21, 2025

* Typo fixes

f4ce4d3

salamonpavel reviewed Feb 21, 2025

View reviewed changes

* addressed more PR comments

56cfa09

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 Feb 17, 2025 •

edited

Loading

benedeki Feb 21, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

benedeki Feb 21, 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

benedeki Feb 21, 2025

lsulak Feb 17, 2025

lsulak Feb 17, 2025

salamonpavel Feb 21, 2025

benedeki Feb 21, 2025

benedeki Feb 22, 2025

salamonpavel Feb 21, 2025

benedeki Feb 21, 2025

salamonpavel Feb 21, 2025 •

edited

Loading

benedeki Feb 21, 2025

salamonpavel Feb 21, 2025 •

edited

Loading

salamonpavel Feb 21, 2025

salamonpavel Feb 21, 2025


		private val partitioningEncoded = "W3sia2V5IjoiYSIsInZhbHVlIjoiYiJ9LHsia2V5IjoiYyIsInZhbHVlIjoiZCJ9XQ=="

		private val partitioningResponse =

		sealed abstract class RequestException(message: String) extends ReaderException(message)


		object RequestException {

#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

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

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

salamonpavel Feb 21, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

salamonpavel Feb 21, 2025 • edited Loading

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 Feb 17, 2025 •

edited

Loading

salamonpavel Feb 21, 2025 •

edited

Loading

salamonpavel Feb 21, 2025 •

edited

Loading