Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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))
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading