16
16
17
17
package za .co .absa .atum .server .model
18
18
19
- import io .circe .{ DecodingFailure , Json }
19
+ import io .circe .Json
20
20
import za .co .absa .atum .model .dto .{
21
21
CheckpointWithPartitioningDTO ,
22
22
MeasureDTO ,
@@ -50,6 +50,26 @@ object CheckpointItemWithPartitioningFromDB {
50
50
private def fromItemsToCheckpointWithPartitioningDTO (
51
51
checkpointItems : Seq [CheckpointItemWithPartitioningFromDB ]
52
52
): Either [Throwable , CheckpointWithPartitioningDTO ] = {
53
+ for {
54
+ measurements <- extractMeasurements(checkpointItems)
55
+ partitioning <- extractPartitioning(checkpointItems)
56
+ } yield {
57
+ CheckpointWithPartitioningDTO (
58
+ id = checkpointItems.head.idCheckpoint,
59
+ name = checkpointItems.head.checkpointName,
60
+ author = checkpointItems.head.author,
61
+ measuredByAtumAgent = checkpointItems.head.measuredByAtumAgent,
62
+ processStartTime = checkpointItems.head.checkpointStartTime,
63
+ processEndTime = checkpointItems.head.checkpointEndTime,
64
+ measurements = measurements.toSet,
65
+ partitioning
66
+ )
67
+ }
68
+ }
69
+
70
+ private def extractMeasurements (
71
+ checkpointItems : Seq [CheckpointItemWithPartitioningFromDB ]
72
+ ): Either [Throwable , Seq [MeasurementDTO ]] = {
53
73
val measurementsOrErr = checkpointItems.map { checkpointItem =>
54
74
checkpointItem.measurementValue.as[MeasureResultDTO ].map { measureResult =>
55
75
MeasurementDTO (
@@ -61,40 +81,22 @@ object CheckpointItemWithPartitioningFromDB {
61
81
)
62
82
}
63
83
}
64
- val partitioningOrErr : Either [DecodingFailure , PartitioningWithIdDTO ] = {
65
- val decodingResult = checkpointItems.head.partitioning.as[PartitioningForDB ]
66
- decodingResult.map{ partitioningForDB =>
67
- val partitioningDTO = partitioningForDB.keys.map { key =>
68
- PartitionDTO (key, partitioningForDB.keysToValuesMap(key))
69
- }
70
- PartitioningWithIdDTO (
71
- id = checkpointItems.head.idPartitioning,
72
- partitioning = partitioningDTO,
73
- author = checkpointItems.head.partitioningAuthor
74
- )
75
- }
76
- }
77
-
78
- }
79
-
80
- val measurementsErrors = measurementsOrErr.collect { case Left (err) => err }
81
- val errors = measurementsErrors ++ partitioningOrErr.left.toSeq
84
+ measurementsOrErr
85
+ .collectFirst { case Left (err) => Left (err) }
86
+ .getOrElse(Right (measurementsOrErr.collect { case Right (measurement) => measurement }))
87
+ }
82
88
83
- if (errors.nonEmpty) {
84
- Left (measurementsErrors.head)
85
- } else {
86
- val measurements = measurementsOrErr.collect { case Right (measurement) => measurement }.toSet
87
- Right (
88
- CheckpointWithPartitioningDTO (
89
- id = checkpointItems.head.idCheckpoint,
90
- name = checkpointItems.head.checkpointName,
91
- author = checkpointItems.head.author,
92
- measuredByAtumAgent = checkpointItems.head.measuredByAtumAgent,
93
- processStartTime = checkpointItems.head.checkpointStartTime,
94
- processEndTime = checkpointItems.head.checkpointEndTime,
95
- measurements = measurements,
96
- partitioningOrErr.toOption.get
97
- )
89
+ private def extractPartitioning (
90
+ checkpointItems : Seq [CheckpointItemWithPartitioningFromDB ]
91
+ ): Either [Throwable , PartitioningWithIdDTO ] = {
92
+ checkpointItems.head.partitioning.as[PartitioningForDB ].map { partitioningForDB =>
93
+ val partitioningDTO = partitioningForDB.keys.map { key =>
94
+ PartitionDTO (key, partitioningForDB.keysToValuesMap(key))
95
+ }
96
+ PartitioningWithIdDTO (
97
+ id = checkpointItems.head.idPartitioning,
98
+ partitioning = partitioningDTO,
99
+ author = checkpointItems.head.partitioningAuthor
98
100
)
99
101
}
100
102
}
0 commit comments