Fixes #120 - Fixing bugs

AbsaOSS · Mar 27, 2024 · c5c5a28 · c5c5a28
1 parent c063480
commit c5c5a28
Show file tree

Hide file tree

Showing 7 changed files with 46 additions and 40 deletions.
diff --git a/server/src/main/scala/za/co/absa/atum/server/api/controller/PartitioningControllerImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/controller/PartitioningControllerImpl.scala
@@ -18,35 +18,36 @@ package za.co.absa.atum.server.api.controller
 
 
 import za.co.absa.atum.model.dto._
-import za.co.absa.atum.server.api.service.PartitioningServiceImpl
-import za.co.absa.atum.server.model.ErrorResponse
+import za.co.absa.atum.server.api.exception.ServiceError
+import za.co.absa.atum.server.api.service.PartitioningService
+import za.co.absa.atum.server.model.{ErrorResponse, InternalServerErrorResponse}
 import zio._
 
-class PartitioningControllerImpl(partitioningServiceImpl: PartitioningServiceImpl)
+class PartitioningControllerImpl(partitioningService: PartitioningService)
   extends PartitioningController with BaseController {
 
   override def createPartitioningIfNotExists(partitioning: PartitioningSubmitDTO): IO[ErrorResponse, AtumContextDTO] = {
-    serviceCallWithStatus[Unit, AtumContextDTO](
-      partitioningServiceImpl.returnAtumContext(partitioning),
-      _ => AtumContextDTO(partitioning = partitioning.partitioning, measures = Set.empty, additionalData = Map.empty)
-    )
+    partitioningService.returnAtumContext(partitioning)
+      .mapError { serviceError: ServiceError =>
+        InternalServerErrorResponse(serviceError.message)
+      }
   }
 
   override def createOrUpdateAdditionalData(
     additionalData: AdditionalDataSubmitDTO
   ): IO[ErrorResponse, AdditionalDataSubmitDTO] = {
     serviceCallWithStatus[Unit, AdditionalDataSubmitDTO](
-      partitioningServiceImpl.createOrUpdateAdditionalData(additionalData),
+      partitioningService.createOrUpdateAdditionalData(additionalData),
       _ => additionalData
     )
   }
 
 }
 
 object PartitioningControllerImpl {
-  val layer: URLayer[PartitioningServiceImpl, PartitioningController] = ZLayer {
+  val layer: URLayer[PartitioningService, PartitioningController] = ZLayer {
     for {
-      partitioningServiceImpl <- ZIO.service[PartitioningServiceImpl]
-    } yield new PartitioningControllerImpl(partitioningServiceImpl)
+      partitioningService <- ZIO.service[PartitioningService]
+    } yield new PartitioningControllerImpl(partitioningService)
   }
 }
diff --git a/...la/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningAdditionalData.scala b/...la/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningAdditionalData.scala
@@ -23,19 +23,20 @@ import play.api.libs.json.Json
 import za.co.absa.atum.model.dto.{AdditionalDataDTO, PartitioningSubmitDTO}
 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.exception.DatabaseError
 import za.co.absa.atum.server.model.PartitioningForDB
 import za.co.absa.fadb.DBSchema
-import za.co.absa.fadb.doobie.DoobieFunction.DoobieSingleResultFunctionWithStatus
-import za.co.absa.fadb.doobie.{DoobieEngine, StatusWithData}
+import za.co.absa.fadb.doobie.DoobieFunction.DoobieMultipleResultFunction
+import za.co.absa.fadb.doobie.DoobieEngine
 import za.co.absa.fadb.status.handling.implementations.StandardStatusHandling
 import zio.interop.catz.asyncInstance
 import zio.{Task, URLayer, ZIO, ZLayer}
 
 class GetPartitioningAdditionalData (implicit schema: DBSchema, dbEngine: DoobieEngine[Task])
-  extends DoobieSingleResultFunctionWithStatus[PartitioningSubmitDTO, Seq[AdditionalDataDTO], Task]
-    with StandardStatusHandling {
+  extends DoobieMultipleResultFunction[PartitioningSubmitDTO, AdditionalDataDTO, Task]
+  {
 
-  override def sql(values: PartitioningSubmitDTO)(implicit read: Read[StatusWithData[Seq[AdditionalDataDTO]]]): Fragment = {
+  override def sql(values: PartitioningSubmitDTO)(implicit read: Read[AdditionalDataDTO]): Fragment = {
     val partitioning = PartitioningForDB.fromSeqPartitionDTO(values.partitioning)
     val partitioningJsonString = Json.toJson(partitioning).toString
 

diff --git a/...in/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningMeasures.scala b/...in/scala/za/co/absa/atum/server/api/database/runs/functions/GetPartitioningMeasures.scala
@@ -23,19 +23,19 @@ import play.api.libs.json.Json
 import za.co.absa.atum.model.dto.{MeasureDTO, PartitioningSubmitDTO}
 import za.co.absa.atum.server.model.PartitioningForDB
 import za.co.absa.fadb.DBSchema
-import za.co.absa.fadb.doobie.{DoobieEngine, StatusWithData}
-import za.co.absa.fadb.doobie.DoobieFunction.DoobieSingleResultFunctionWithStatus
+import za.co.absa.fadb.doobie.DoobieEngine
+import za.co.absa.fadb.doobie.DoobieFunction.DoobieMultipleResultFunction
 import za.co.absa.fadb.status.handling.implementations.StandardStatusHandling
 import za.co.absa.atum.server.api.database.PostgresDatabaseProvider
 import za.co.absa.atum.server.api.database.runs.Runs
 import zio._
 import zio.interop.catz._
 
 class GetPartitioningMeasures (implicit schema: DBSchema, dbEngine: DoobieEngine[Task])
-  extends DoobieSingleResultFunctionWithStatus[PartitioningSubmitDTO, Seq[MeasureDTO], Task]
-    with StandardStatusHandling {
+  extends DoobieMultipleResultFunction[PartitioningSubmitDTO, Seq[MeasureDTO], Task]
+  {
 
-  override def sql(values: PartitioningSubmitDTO)(implicit read: Read[StatusWithData[Seq[MeasureDTO]]]): Fragment = {
+  override def sql(values: PartitioningSubmitDTO)(implicit read: Read[Seq[MeasureDTO]]): Fragment = {
     val partitioning = PartitioningForDB.fromSeqPartitionDTO(values.partitioning)
     val partitioningJsonString = Json.toJson(partitioning).toString
 

diff --git a/server/src/main/scala/za/co/absa/atum/server/api/repository/PartitioningRepository.scala b/server/src/main/scala/za/co/absa/atum/server/api/repository/PartitioningRepository.scala
@@ -30,11 +30,11 @@ trait PartitioningRepository {
 
   def getPartitioningMeasures(
     partitioning: PartitioningSubmitDTO
-  ): IO[DatabaseError, Either[StatusException, Seq[MeasureDTO]]]
+  ): IO[DatabaseError, Seq[MeasureDTO]]
 
   def getPartitioningAdditionalData(
     partitioning: PartitioningSubmitDTO
-  ): IO[DatabaseError, Either[StatusException,  AdditionalDataDTO]]
+  ): IO[DatabaseError, Seq[AdditionalDataDTO]]
 
   def createOrUpdateAdditionalData(additionalData: AdditionalDataSubmitDTO):
     IO[DatabaseError, Either[StatusException, Unit]]

diff --git a/server/src/main/scala/za/co/absa/atum/server/api/repository/PartitioningRepositoryImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/repository/PartitioningRepositoryImpl.scala
@@ -42,13 +42,13 @@ class PartitioningRepositoryImpl(
   }
 
   override def getPartitioningMeasures(partitioning: PartitioningSubmitDTO):
-    IO[DatabaseError, Either[StatusException, Seq[MeasureDTO]]] = {
-    dbCallWithStatus(getPartitioningMeasuresFn(partitioning), "getPartitioningMeasures")
+    IO[DatabaseError, Seq[MeasureDTO]] = {
+    getPartitioningMeasuresFn(partitioning)
   }
 
   override def getPartitioningAdditionalData(partitioning: PartitioningSubmitDTO):
-    IO[DatabaseError, Either[StatusException, Seq[AdditionalDataDTO]]] = {
-    dbCallWithStatus(getPartitioningAdditionalDataFn(partitioning), "getPartitioningAdditionalData")
+    IO[DatabaseError, Seq[AdditionalDataDTO]] = {
+    getPartitioningAdditionalDataFn(partitioning)
   }
 }
 

diff --git a/server/src/main/scala/za/co/absa/atum/server/api/service/PartitioningService.scala b/server/src/main/scala/za/co/absa/atum/server/api/service/PartitioningService.scala
@@ -16,7 +16,7 @@
 
 package za.co.absa.atum.server.api.service
 
-import za.co.absa.atum.model.dto.{AdditionalDataDTO, AdditionalDataSubmitDTO, MeasureDTO, PartitioningSubmitDTO}
+import za.co.absa.atum.model.dto.{AdditionalDataDTO, AdditionalDataSubmitDTO, AtumContextDTO, MeasureDTO, PartitioningSubmitDTO}
 import za.co.absa.atum.server.api.exception.ServiceError
 import za.co.absa.fadb.exceptions.StatusException
 import zio.IO
@@ -28,12 +28,14 @@ trait PartitioningService {
     IO[ServiceError, Either[StatusException, Unit]]
 
   def getPartitioningMeasures(partitioning: PartitioningSubmitDTO):
-    IO[ServiceError, Either[StatusException, Seq[MeasureDTO]]]
+    IO[ServiceError, Seq[MeasureDTO]]
 
   def getPartitioningAdditionalData(partitioning: PartitioningSubmitDTO):
-    IO[ServiceError, Either[StatusException, AdditionalDataDTO]]
+    IO[ServiceError, Seq[AdditionalDataDTO]]
 
   def createOrUpdateAdditionalData(additionalData: AdditionalDataSubmitDTO):
     IO[ServiceError, Either[StatusException, Unit]]
 
+  def returnAtumContext(partitioning: PartitioningSubmitDTO): IO[ServiceError, AtumContextDTO]
+
 }
diff --git a/server/src/main/scala/za/co/absa/atum/server/api/service/PartitioningServiceImpl.scala b/server/src/main/scala/za/co/absa/atum/server/api/service/PartitioningServiceImpl.scala
@@ -19,8 +19,8 @@ package za.co.absa.atum.server.api.service
 import za.co.absa.atum.model.dto.{AdditionalDataDTO, AdditionalDataSubmitDTO, AtumContextDTO, MeasureDTO, PartitioningSubmitDTO}
 import za.co.absa.atum.server.api.exception.ServiceError
 import za.co.absa.atum.server.api.repository.PartitioningRepository
-import za.co.absa.atum.server.model.ErrorResponse
 import za.co.absa.fadb.exceptions.StatusException
+import za.co.absa.atum.server.api.exception.DatabaseError
 import zio._
 
 class PartitioningServiceImpl(partitioningRepository: PartitioningRepository)
@@ -42,16 +42,18 @@ class PartitioningServiceImpl(partitioningRepository: PartitioningRepository)
     )
   }
 
-  override def getPartitioningMeasures(partitioning: PartitioningSubmitDTO): IO[ServiceError, Either[StatusException, Seq[MeasureDTO]]] = {
-    repositoryCallWithStatus(
-      partitioningRepository.getPartitioningMeasures(partitioning), "getPartitioningMeasures"
-    )
+  override def getPartitioningMeasures(partitioning: PartitioningSubmitDTO): IO[ServiceError, Seq[MeasureDTO]] = {
+    partitioningRepository.getPartitioningMeasures(partitioning)
+      .mapError { case DatabaseError(message) =>
+        ServiceError(s"Failed to retrieve partitioning measures': $message")
+      }
   }
 
-  override def getPartitioningAdditionalData(partitioning: PartitioningSubmitDTO): IO[ServiceError, Either[StatusException, AdditionalDataDTO]] = {
-    repositoryCallWithStatus(
-      partitioningRepository.getPartitioningAdditionalData(partitioning), "getPartitioningAdditionalData"
-    )
+  override def getPartitioningAdditionalData(partitioning: PartitioningSubmitDTO): IO[ServiceError, Seq[AdditionalDataDTO]] = {
+    partitioningRepository.getPartitioningAdditionalData(partitioning)
+      .mapError { case DatabaseError(message) =>
+        ServiceError(s"Failed to retrieve partitioning additional data': $message")
+      }
   }
 
   def returnAtumContext(partitioning: PartitioningSubmitDTO): IO[ServiceError, AtumContextDTO] = {
@@ -72,7 +74,7 @@ class PartitioningServiceImpl(partitioningRepository: PartitioningRepository)
 
       measures <- getPartitioningMeasures(partitioning)
         .flatMap {
-          case Left(_) => ZIO.succeed(Set.empty[MeasureDTO])
+          case Left(_) => ZIO.succeed(Seq.empty[MeasureDTO])
           case Right(value) => ZIO.succeed(value.toSet)
         }
         .mapError(error => ServiceError(error.message))