From 9a6d4968e9632e2511112381b47796a7b589eedf Mon Sep 17 00:00:00 2001 From: Ladislav Sulak Date: Thu, 23 Jul 2026 12:57:41 +0200 Subject: [PATCH] #328: making partitioning author available for flor and partitioning reader --- .../za/co/absa/atum/reader/FlowReader.scala | 7 +++++++ .../absa/atum/reader/PartitioningReader.scala | 7 +++++++ .../reader/core/PartitioningIdProvider.scala | 15 +++++++++++++-- .../atum/reader/FlowReaderUnitTests.scala | 18 ++++++++++++++++++ .../reader/PartitioningReaderUnitTests.scala | 19 +++++++++++++++++++ .../PartitioningIdProviderUnitTests.scala | 7 +++++++ 6 files changed, 71 insertions(+), 2 deletions(-) diff --git a/reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala b/reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala index 9d26e83a9..05e630a77 100644 --- a/reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala +++ b/reader/src/main/scala/za/co/absa/atum/reader/FlowReader.scala @@ -122,6 +122,13 @@ case class FlowReader[F[_]](mainFlowPartitioning: AtumPartitions)(implicit } yield checkpointsOrError } + /** + * Returns the author of the partitioning identifying the flow. + * + * @return - the author of the main flow partitioning + */ + def getAuthor: F[RequestResult[String]] = partitioningAuthor(mainFlowPartitioning) + private def queryFlowId(mainPartitioningId: Long): F[RequestResult[Long]] = { val endpoint = s"/$Api/$V2/${V2Paths.Partitionings}/$mainPartitioningId/${V2Paths.MainFlow}" val queryResult = getQuery[SingleSuccessResponse[FlowDTO]](endpoint) diff --git a/reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala b/reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala index a43f81311..a406f0035 100644 --- a/reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala +++ b/reader/src/main/scala/za/co/absa/atum/reader/PartitioningReader.scala @@ -141,6 +141,13 @@ case class PartitioningReader[F[_]](partitioning: AtumPartitions)(implicit } } + /** + * Returns the author of the partitioning. + * + * @return - the author of the partitioning identifying this reader + */ + def getAuthor: F[RequestResult[String]] = partitioningAuthor(partitioning) + private def queryCheckpoints( partitioningId: Long, checkpointName: Option[String], diff --git a/reader/src/main/scala/za/co/absa/atum/reader/core/PartitioningIdProvider.scala b/reader/src/main/scala/za/co/absa/atum/reader/core/PartitioningIdProvider.scala index 3398c7401..ad3ae6492 100644 --- a/reader/src/main/scala/za/co/absa/atum/reader/core/PartitioningIdProvider.scala +++ b/reader/src/main/scala/za/co/absa/atum/reader/core/PartitioningIdProvider.scala @@ -28,14 +28,25 @@ import za.co.absa.atum.reader.core.RequestResult.RequestResult trait PartitioningIdProvider[F[_]] { self: Reader[F] => - def partitioningId(partitioning: AtumPartitions)(implicit monad: MonadError[F]): F[RequestResult[Long]] = { + + def partitioningWithId(partitioning: AtumPartitions)( + implicit monad: MonadError[F] + ): F[RequestResult[PartitioningWithIdDTO]] = { val encodedPartitioning = partitioning.toPartitioningDTO.asBase64EncodedJsonString val queryResult = getQuery[SingleSuccessResponse[PartitioningWithIdDTO]]( s"/$Api/$V2/${V2Paths.Partitionings}", Map("partitioning" -> encodedPartitioning) ) queryResult.map { result => - result.map(_.data.id) + result.map(_.data) } } + + def partitioningId(partitioning: AtumPartitions)(implicit monad: MonadError[F]): F[RequestResult[Long]] = { + partitioningWithId(partitioning).map(_.map(_.id)) + } + + def partitioningAuthor(partitioning: AtumPartitions)(implicit monad: MonadError[F]): F[RequestResult[String]] = { + partitioningWithId(partitioning).map(_.map(_.author)) + } } diff --git a/reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala b/reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala index bc79f2912..b30611050 100644 --- a/reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala +++ b/reader/src/test/scala/za/co/absa/atum/reader/FlowReaderUnitTests.scala @@ -314,6 +314,24 @@ class FlowReaderUnitTests extends AnyFunSuiteLike { assert(flowReader.isInstanceOf[FlowReader[Identity]]) } + test("The flow partitioning author is properly queried and delivered") { + implicit val server: SttpBackendStub[Identity, capabilities.WebSockets] = SttpBackendStub.synchronous + .whenRequestMatchesPartial { + case r if r.uri.path.endsWith(List("partitionings")) => + assert(r.uri.querySegments.contains(KeyValue("partitioning", partitioningEncoded))) + Response.ok(partitioningResponse) + } + + val atumPartitions: AtumPartitions = AtumPartitions(List( + "a" -> "b", + "c" -> "d" + )) + + val reader = FlowReader(atumPartitions) + val result = reader.getAuthor + assert(result == Right("James Bond")) + } + } object FlowReaderUnitTests { diff --git a/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala b/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala index e08ce76cc..37745ed07 100644 --- a/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala +++ b/reader/src/test/scala/za/co/absa/atum/reader/PartitioningReaderUnitTests.scala @@ -301,6 +301,25 @@ class PartitioningReaderUnitTests extends AnyFunSuiteLike { val result = reader.getAdditionalData assert(result == Right(expected)) } + + test("The partitioning author is properly queried and delivered") { + val atumPartitions: AtumPartitions = AtumPartitions( + List( + "a" -> "b", + "c" -> "d" + ) + ) + implicit val server: SttpBackendStub[Identity, capabilities.WebSockets] = SttpBackendStub.synchronous + .whenRequestMatchesPartial { + case r if r.uri.path.endsWith(List(V2Paths.Partitionings)) => + assert(r.uri.querySegments.contains(KeyValue("partitioning", partitioningEncoded))) + Response.ok(partitioningResponse) + } + + val reader = PartitioningReader[Identity](atumPartitions) + val result = reader.getAuthor + assert(result == Right("James Bond")) + } } object PartitioningReaderUnitTests { diff --git a/reader/src/test/scala/za/co/absa/atum/reader/core/PartitioningIdProviderUnitTests.scala b/reader/src/test/scala/za/co/absa/atum/reader/core/PartitioningIdProviderUnitTests.scala index 22672eabf..01ebfbabe 100644 --- a/reader/src/test/scala/za/co/absa/atum/reader/core/PartitioningIdProviderUnitTests.scala +++ b/reader/src/test/scala/za/co/absa/atum/reader/core/PartitioningIdProviderUnitTests.scala @@ -72,6 +72,13 @@ class PartitioningIdProviderUnitTests extends AnyFunSuiteLike { assert(result == 1) } + test("Gets the partitioning author") { + val reader = ReaderWithPartitioningIdForTest(atumPartitionsToReply) + val response = reader.partitioningAuthor(atumPartitionsToReply) + val result: String = response.getOrElse(throw new Exception("Failed to get partitioning author")) + assert(result == "Gimli") + } + test("Not found on the partitioning id") { val reader = ReaderWithPartitioningIdForTest(atumPartitionsToNotFound) val result = reader.partitioningId(atumPartitionsToNotFound)