#305 Get-Ancestors-Server Functionality

project/JacocoSetup.scala

@@ -50,6 +50,8 @@ object JacocoSetup {
       "za.co.absa.atum.server.api.v2.controller.PartitioningController$",
       "za.co.absa.atum.server.api.v2.service.PartitioningService",
       "za.co.absa.atum.server.api.v2.service.PartitioningService$",
+      "za.co.absa.atum.server.api.common.http.Routes*",
+      "za.co.absa.atum.server.implicits.SeqImplicits*",
       "za.co.absa.atum.model.envelopes.Pagination",
       "za.co.absa.atum.model.dto.PartitioningParentPatchDTO*",
       "za.co.absa.atum.model.ApiPaths*",

server/src/main/scala/za/co/absa/atum/server/Main.scala

@@ -107,6 +107,7 @@ object Main extends ZIOAppDefault {
           GetFlowPartitionings.layer,
           GetPartitioningMainFlow.layer,
           UpdatePartitioningParent.layer,
+          GetPartitioningAncestors.layer,
           PostgresDatabaseProvider.layer,
           TransactorProvider.layer,
           // aws

server/src/main/scala/za/co/absa/atum/server/api/common/http/Routes.scala

@@ -64,6 +64,7 @@ object Routes extends ServerOptions with ServerUtils {
       api.v2.http.Endpoints.getPartitioningMainFlowEndpoint,
       api.v2.http.Endpoints.getFlowPartitioningsEndpoint,
       api.v2.http.Endpoints.getFlowCheckpointsEndpoint,
+      api.v2.http.Endpoints.getPartitioningAncestorsEndpoint,
       api.common.http.Endpoints.healthEndpoint
     )
     ZHttp4sServerInterpreter[HttpEnv.Env](http4sServerOptions(None))

server/src/main/scala/za/co/absa/atum/server/api/database/flows/functions/GetFlowPartitionings.scala

@@ -29,11 +29,9 @@ import za.co.absa.db.fadb.doobie.DoobieFunction.DoobieMultipleResultFunctionWith
 import za.co.absa.db.fadb.status.aggregation.implementations.ByFirstErrorStatusAggregator
 import za.co.absa.db.fadb.status.handling.implementations.StandardStatusHandling
 import zio.{Task, URLayer, ZIO, ZLayer}
-
+import za.co.absa.atum.server.model.PartitioningResult
 import za.co.absa.db.fadb.doobie.postgres.circe.implicits.jsonbGet
 
-import scala.annotation.tailrec
-
 class GetFlowPartitionings(implicit schema: DBSchema, dbEngine: DoobieEngine[Task])
     extends DoobieMultipleResultFunctionWithAggStatus[GetFlowPartitioningsArgs, Option[
       GetFlowPartitioningsResult
@@ -53,30 +51,7 @@ class GetFlowPartitionings(implicit schema: DBSchema, dbEngine: DoobieEngine[Tas
 object GetFlowPartitionings {
   case class GetFlowPartitioningsArgs(flowId: Long, limit: Option[Int], offset: Option[Long])
   case class GetFlowPartitioningsResult(id: Long, partitioningJson: Json, author: String, hasMore: Boolean)
-
-  object GetFlowPartitioningsResult {
-
-    @tailrec def resultsToPartitioningWithIdDTOs(
-      results: Seq[GetFlowPartitioningsResult],
-      acc: Seq[PartitioningWithIdDTO]
-    ): Either[DecodingFailure, Seq[PartitioningWithIdDTO]] = {
-      if (results.isEmpty) Right(acc)
-      else {
-        val head = results.head
-        val tail = results.tail
-        val decodingResult = head.partitioningJson.as[PartitioningForDB]
-        decodingResult match {
-          case Left(decodingFailure) => Left(decodingFailure)
-          case Right(partitioningForDB) =>
-            val partitioningDTO = partitioningForDB.keys.map { key =>
-              PartitionDTO(key, partitioningForDB.keysToValuesMap(key))
-            }
-            resultsToPartitioningWithIdDTOs(tail, acc :+ PartitioningWithIdDTO(head.id, partitioningDTO, head.author))
-        }
-      }
-    }
-
-  }
+    extends PartitioningResult(id, partitioningJson, author)
 
   val layer: URLayer[PostgresDatabaseProvider, GetFlowPartitionings] = ZLayer {
     for {

server/src/main/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningAncestors.scala

@@ -0,0 +1,59 @@
+/*
+ * 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.server.api.database.runs.functions
+
+import doobie.implicits.toSqlInterpolator
+import io.circe.Json
+import za.co.absa.atum.server.api.database.PostgresDatabaseProvider
+import za.co.absa.atum.server.api.database.runs.Runs
+import za.co.absa.atum.server.api.database.runs.functions.GetPartitioningAncestors._
+import za.co.absa.atum.server.model.PartitioningResult
+import za.co.absa.db.fadb.DBSchema
+import za.co.absa.db.fadb.doobie.DoobieEngine
+import za.co.absa.db.fadb.doobie.DoobieFunction.DoobieMultipleResultFunctionWithAggStatus
+import za.co.absa.db.fadb.status.aggregation.implementations.ByFirstErrorStatusAggregator
+import za.co.absa.db.fadb.status.handling.implementations.StandardStatusHandling
+import zio.{Task, URLayer, ZIO, ZLayer}
+import za.co.absa.db.fadb.doobie.postgres.circe.implicits.jsonbGet
+
+class GetPartitioningAncestors(implicit schema: DBSchema, dbEngine: DoobieEngine[Task])
+  extends DoobieMultipleResultFunctionWithAggStatus[GetPartitioningAncestorsArgs, Option[
+    GetPartitioningAncestorsResult
+  ], Task](args =>
+    Seq(
+      fr"${args.partitioningId}",
+      fr"${args.limit}",
+      fr"${args.offset}"
+    )
+  )
+    with StandardStatusHandling
+    with ByFirstErrorStatusAggregator {
+
+  override def fieldsToSelect: Seq[String] = super.fieldsToSelect ++ Seq("ancestor_id", "partitioning", "author", "has_more")
+}
+
+object GetPartitioningAncestors {
+  case class GetPartitioningAncestorsArgs(partitioningId: Long, limit: Option[Int], offset: Option[Long])
+  case class GetPartitioningAncestorsResult(id: Long, partitioningJson: Json, author: String, hasMore: Boolean)
+    extends PartitioningResult(id, partitioningJson, author)
+
+  val layer: URLayer[PostgresDatabaseProvider, GetPartitioningAncestors] = ZLayer {
+    for {
+      dbProvider <- ZIO.service[PostgresDatabaseProvider]
+    } yield new GetPartitioningAncestors()(Runs, dbProvider.dbEngine)
+  }
+}

server/src/main/scala/za/co/absa/atum/server/api/v2/controller/PartitioningController.scala

@@ -63,4 +63,11 @@ trait PartitioningController {
     partitioningParentPatchDTO: PartitioningParentPatchDTO
   ): IO[ErrorResponse, Unit]
 
+
+  def getPartitioningAncestors(
+    partitioningId: Long,
+    limit: Option[Int],
+    offset: Option[Long]
+  ): IO[ErrorResponse, PaginatedResponse[PartitioningWithIdDTO]]
+
 }

server/src/main/scala/za/co/absa/atum/server/api/v2/controller/PartitioningControllerImpl.scala

@@ -133,6 +133,21 @@ class PartitioningControllerImpl(partitioningService: PartitioningService)
           _ => ()
       )
   }
+
+  override def getPartitioningAncestors(
+    partitioningId: Long,
+    limit: Option[Int],
+    offset: Option[Long]
+   ): IO[ErrorResponse, PaginatedResponse[PartitioningWithIdDTO]] = {
+    mapToPaginatedResponse(
+      limit.get,
+      offset.get,
+      serviceCall[PaginatedResult[PartitioningWithIdDTO], PaginatedResult[PartitioningWithIdDTO]](
+        partitioningService.getPartitioningAncestors(partitioningId, limit, offset)
+      )
+    )
+  }
+
 }
 
 object PartitioningControllerImpl {

server/src/main/scala/za/co/absa/atum/server/api/v2/http/Endpoints.scala

@@ -179,6 +179,19 @@ object Endpoints extends BaseEndpoints {
       .errorOutVariantPrepend(notFoundErrorOneOfVariant)
   }
 
+  val getPartitioningAncestorsEndpoint
+  : PublicEndpoint[(Long, Option[Int], Option[Long]), ErrorResponse, PaginatedResponse[
+    PartitioningWithIdDTO
+  ], Any] = {
+    apiV2.get
+      .in(V2Paths.Partitionings / path[Long]("partitioningId") / V2Paths.Ancestors)
+      .in(query[Option[Int]]("limit").default(Some(10)).validateOption(Validator.inRange(1, 1000)))
+      .in(query[Option[Long]]("offset").default(Some(0L)).validateOption(Validator.min(0L)))
+      .out(statusCode(StatusCode.Ok))
+      .out(jsonBody[PaginatedResponse[PartitioningWithIdDTO]])
+      .errorOutVariantPrepend(notFoundErrorOneOfVariant)
+  }
+
   val serverEndpoints: List[ZServerEndpoint[HttpEnv.Env, Any]] = List(
     createServerEndpoint[
       (Long, CheckpointV2DTO),
@@ -259,6 +272,16 @@ object Endpoints extends BaseEndpoints {
         PartitioningController.patchPartitioningParent(partitioningId, partitioningParentPatchDTO)
       }
     ),
+    createServerEndpoint[
+      (Long, Option[Int], Option[Long]),
+      ErrorResponse,
+      PaginatedResponse[PartitioningWithIdDTO]
+    ](
+      getPartitioningAncestorsEndpoint,
+      { case (partitioningId: Long, limit: Option[Int], offset: Option[Long]) =>
+        PartitioningController.getPartitioningAncestors(partitioningId, limit, offset)
+      }
+    )
   )
 
 }

server/src/main/scala/za/co/absa/atum/server/api/v2/repository/PartitioningRepository.scala

@@ -60,4 +60,11 @@ trait PartitioningRepository {
     partitioningId: Long,
     partitioningParentPatchDTO: PartitioningParentPatchDTO
   ): IO[DatabaseError, Unit]
+
+  def getPartitioningAncestors(
+    partitioningId: Long,
+    limit: Option[Int],
+    offset: Option[Long]
+  ): IO[DatabaseError, PaginatedResult[PartitioningWithIdDTO]]
+
 }

server/src/main/scala/za/co/absa/atum/server/api/v2/repository/PartitioningRepositoryImpl.scala

@@ -20,6 +20,7 @@ import za.co.absa.atum.model.dto._
 import za.co.absa.atum.server.api.common.repository.BaseRepository
 import za.co.absa.atum.server.api.database.flows.functions.GetFlowPartitionings._
 import za.co.absa.atum.server.api.database.flows.functions._
+import za.co.absa.atum.server.api.database.runs.functions.GetPartitioningAncestors._
 import za.co.absa.atum.server.api.database.runs.functions.CreateOrUpdateAdditionalData.CreateOrUpdateAdditionalDataArgs
 import za.co.absa.atum.server.api.database.runs.functions.UpdatePartitioningParent.UpdatePartitioningParentArgs
 import za.co.absa.atum.server.api.database.runs.functions._
@@ -46,7 +47,8 @@ class PartitioningRepositoryImpl(
   getPartitioningFn: GetPartitioning,
   getFlowPartitioningsFn: GetFlowPartitionings,
   getPartitioningMainFlowFn: GetPartitioningMainFlow,
-  updatePartitioningParentFn: UpdatePartitioningParent
+  updatePartitioningParentFn: UpdatePartitioningParent,
+  getPartitioningAncestorsFn: GetPartitioningAncestors
 ) extends PartitioningRepository
     with BaseRepository {
 
@@ -138,7 +140,7 @@ class PartitioningRepositoryImpl(
     ).map(_.flatten)
       .flatMap { partitioningResults =>
         ZIO
-          .fromEither(GetFlowPartitioningsResult.resultsToPartitioningWithIdDTOs(partitioningResults, Seq.empty))
+          .fromEither(PartitioningResult.resultsToPartitioningWithIdDTOs(partitioningResults))
           .mapBoth(
             error => GeneralDatabaseError(error.getMessage),
             partitionings => {
@@ -168,6 +170,29 @@ class PartitioningRepositoryImpl(
       "updatePartitioningParent"
     )
   }
+
+  override def getPartitioningAncestors(
+     partitioningId: Long,
+     limit: Option[Int],
+     offset: Option[Long]
+   ): IO[DatabaseError, PaginatedResult[PartitioningWithIdDTO]] = {
+    dbMultipleResultCallWithAggregatedStatus(
+      getPartitioningAncestorsFn(GetPartitioningAncestorsArgs(partitioningId, limit, offset)),
+      "getPartitioningAncestors"
+    ).map(_.flatten)
+      .flatMap { partitioningResults =>
+        ZIO
+          .fromEither(PartitioningResult.resultsToPartitioningWithIdDTOs(partitioningResults))
+          .mapBoth(
+            error => GeneralDatabaseError(error.getMessage),
+            partitionings => {
+              if (partitioningResults.nonEmpty && partitioningResults.head.hasMore) ResultHasMore(partitionings)
+              else ResultNoMore(partitionings)
+            }
+          )
+      }
+  }
+
 }
 
 object PartitioningRepositoryImpl {
@@ -181,7 +206,8 @@ object PartitioningRepositoryImpl {
       with GetPartitioning
       with GetFlowPartitionings
       with GetPartitioningMainFlow
-      with UpdatePartitioningParent,
+      with UpdatePartitioningParent
+      with GetPartitioningAncestors,
     PartitioningRepository
   ] = ZLayer {
     for {
@@ -195,6 +221,7 @@ object PartitioningRepositoryImpl {
       getFlowPartitionings <- ZIO.service[GetFlowPartitionings]
       getPartitioningMainFlow <- ZIO.service[GetPartitioningMainFlow]
       updatePartitioningParent <- ZIO.service[UpdatePartitioningParent]
+      getPartitioningAncestors <- ZIO.service[GetPartitioningAncestors]
     } yield new PartitioningRepositoryImpl(
       createPartitioning,
       getPartitioningMeasures,
@@ -205,7 +232,8 @@ object PartitioningRepositoryImpl {
       getPartitioning,
       getFlowPartitionings,
       getPartitioningMainFlow,
-      updatePartitioningParent
+      updatePartitioningParent,
+      getPartitioningAncestors
     )
   }
 

server/src/main/scala/za/co/absa/atum/server/api/v2/service/PartitioningService.scala

@@ -58,4 +58,10 @@ trait PartitioningService {
     partitioningId: Long,
     partitioningParentPatchDTO: PartitioningParentPatchDTO
   ): IO[ServiceError, Unit]
+
+  def getPartitioningAncestors(
+    partitioningId: Long,
+    limit: Option[Int],
+    offset: Option[Long]
+  ): IO[ServiceError, PaginatedResult[PartitioningWithIdDTO]]
 }

server/src/main/scala/za/co/absa/atum/server/api/v2/service/PartitioningServiceImpl.scala

@@ -107,6 +107,18 @@ class PartitioningServiceImpl(partitioningRepository: PartitioningRepository)
       "updatePartitioningParent"
     )
   }
+
+  override def getPartitioningAncestors(
+     partitioningId: Long,
+     limit: Option[Int],
+     offset: Option[Long]
+   ): IO[ServiceError, PaginatedResult[PartitioningWithIdDTO]] = {
+    repositoryCall(
+      partitioningRepository.getPartitioningAncestors(partitioningId, limit, offset),
+      "getPartitioningAncestors"
+    )
+  }
+
 }
 
 object PartitioningServiceImpl {

server/src/main/scala/za/co/absa/atum/server/implicits/SeqImplicits.scala

@@ -0,0 +1,33 @@
+/*
+ * 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.server.implicits
+
+import io.circe.DecodingFailure
+
+object SeqImplicits {
+  implicit class SeqEnhancements[T](val seq: Seq[T]) extends AnyVal {
+
+    def seqDecode[R](decodingFnc: T => Either[DecodingFailure, R]): Either[DecodingFailure, Seq[R]] = {
+      seq.foldLeft(Right(List.empty[R]): Either[DecodingFailure, List[R]]) { (acc, item) =>
+        for {
+          decodedList <- acc
+          decodedItem <- decodingFnc(item)
+        } yield decodedItem :: decodedList
+      }.map(_.reverse)
+    }
+  }
+}