Merge pull request #818 from checkmarble/feat/ingested-data-reader-en…

…dpoints feat: add endpoint to get ingested object data
checkmarble · Feb 5, 2025 · 2dc71d9 · 2dc71d9
2 parents 5cd69e1 + 984e941
commit 2dc71d9
Show file tree

Hide file tree

Showing 9 changed files with 100 additions and 6 deletions.
diff --git a/api/handle_ingestion.go b/api/handle_ingestion.go
@@ -128,3 +128,29 @@ func handleListUploadLogs(uc usecases.Usecases) func(c *gin.Context) {
 		c.JSON(http.StatusOK, pure_utils.Map(uploadLogs, dto.AdaptUploadLogDto))
 	}
 }
+
+func handleGetIngestedObject(uc usecases.Usecases) func(c *gin.Context) {
+	return func(c *gin.Context) {
+		ctx := c.Request.Context()
+		organizationID, err := utils.OrganizationIdFromRequest(c.Request)
+		if presentError(ctx, c, err) {
+			return
+		}
+
+		objectType := c.Param("object_type")
+		objectId := c.Param("object_id")
+
+		usecase := usecasesWithCreds(ctx, uc).NewIngestedDataReaderUsecase()
+		objects, err := usecase.GetIngestedObject(ctx, organizationID, objectType, objectId)
+		if presentError(ctx, c, err) {
+			return
+		}
+
+		if len(objects) == 0 {
+			c.JSON(http.StatusNotFound, nil)
+			return
+		}
+
+		c.JSON(http.StatusOK, dto.DataModelObject{Data: objects[0].Data, Metadata: objects[0].Metadata})
+	}
+}
diff --git a/api/routes.go b/api/routes.go
@@ -51,6 +51,7 @@ func addRoutes(r *gin.Engine, conf Configuration, uc usecases.Usecases, auth Aut
 	router.POST("/ingestion/:object_type/multiple", tom, handleIngestionMultiple(uc))
 	router.POST("/ingestion/:object_type/batch", timeoutMiddleware(conf.BatchTimeout), handlePostCsvIngestion(uc))
 	router.GET("/ingestion/:object_type/upload-logs", tom, handleListUploadLogs(uc))
+	router.GET("/ingestion/:object_type/:object_id", tom, handleGetIngestedObject(uc))
 
 	router.GET("/scenarios", tom, listScenarios(uc))
 	router.POST("/scenarios", tom, createScenario(uc))

diff --git a/dto/data_model_dto.go b/dto/data_model_dto.go
@@ -116,3 +116,8 @@ type CreateFieldInput struct {
 	IsEnum      bool   `json:"is_enum"`
 	IsUnique    bool   `json:"is_unique"`
 }
+
+type DataModelObject struct {
+	Data     map[string]any `json:"data"`
+	Metadata map[string]any `json:"metadata"`
+}
diff --git a/models/data_model.go b/models/data_model.go
@@ -252,3 +252,8 @@ type DataModelLinkCreateInput struct {
 	ChildTableID   string
 	ChildFieldID   string
 }
+
+type DataModelObject struct {
+	Data     map[string]any
+	Metadata map[string]any
+}
diff --git a/repositories/ingested_data_read_repository.go b/repositories/ingested_data_read_repository.go
@@ -3,6 +3,7 @@ package repositories
 import (
 	"context"
 	"fmt"
+	"slices"
 
 	"github.com/Masterminds/squirrel"
 	"github.com/cockroachdb/errors"
@@ -25,7 +26,7 @@ type IngestedDataReadRepository interface {
 		exec Executor,
 		table models.Table,
 		objectId string,
-	) ([]map[string]any, error)
+	) ([]models.DataModelObject, error)
 	QueryAggregatedValue(
 		ctx context.Context,
 		exec Executor,
@@ -236,7 +237,7 @@ func (repo *IngestedDataReadRepositoryImpl) QueryIngestedObject(
 	exec Executor,
 	table models.Table,
 	objectId string,
-) ([]map[string]any, error) {
+) ([]models.DataModelObject, error) {
 	if err := validateClientDbExecutor(exec); err != nil {
 		return nil, err
 	}
@@ -248,7 +249,7 @@ func (repo *IngestedDataReadRepositoryImpl) QueryIngestedObject(
 		ctx,
 		exec,
 		qualifiedTableName,
-		columnNames,
+		append(columnNames, "valid_from"),
 		[]models.Filter{{
 			LeftSql:    fmt.Sprintf("%s.object_id", qualifiedTableName),
 			Operator:   ast.FUNC_EQUAL,
@@ -259,7 +260,20 @@ func (repo *IngestedDataReadRepositoryImpl) QueryIngestedObject(
 		return nil, err
 	}
 
-	return objectsAsMap, nil
+	ingestedObjects := make([]models.DataModelObject, len(objectsAsMap))
+	for i, object := range objectsAsMap {
+		ingestedObject := models.DataModelObject{Data: map[string]any{}, Metadata: map[string]any{}}
+		for fieldName, fieldValue := range object {
+			if slices.Contains(columnNames, fieldName) {
+				ingestedObject.Data[fieldName] = fieldValue
+			} else {
+				ingestedObject.Metadata[fieldName] = fieldValue
+			}
+		}
+		ingestedObjects[i] = ingestedObject
+	}
+
+	return ingestedObjects, nil
 }
 
 func queryWithDynamicColumnList(

diff --git a/usecases/ingested_data_reader_usecase.go b/usecases/ingested_data_reader_usecase.go
@@ -0,0 +1,35 @@
+package usecases
+
+import (
+	"context"
+
+	"github.com/checkmarble/marble-backend/models"
+	"github.com/checkmarble/marble-backend/repositories"
+	"github.com/checkmarble/marble-backend/usecases/executor_factory"
+)
+
+type IngestedDataReaderUsecase struct {
+	ingestedDataReadRepository repositories.IngestedDataReadRepository
+	dataModelRepository        repositories.DataModelRepository
+	executorFactory            executor_factory.ExecutorFactory
+}
+
+func (usecase *IngestedDataReaderUsecase) GetIngestedObject(ctx context.Context,
+	organizationId string, objectType string, objectId string,
+) ([]models.DataModelObject, error) {
+	exec := usecase.executorFactory.NewExecutor()
+
+	dataModel, err := usecase.dataModelRepository.GetDataModel(ctx, exec, organizationId, true)
+	if err != nil {
+		return nil, err
+	}
+
+	table := dataModel.Tables[objectType]
+
+	db, err := usecase.executorFactory.NewClientDbExecutor(ctx, organizationId)
+	if err != nil {
+		return nil, err
+	}
+
+	return usecase.ingestedDataReadRepository.QueryIngestedObject(ctx, db, table, objectId)
+}
diff --git a/usecases/scheduled_execution/async_decision_job.go b/usecases/scheduled_execution/async_decision_job.go
@@ -275,7 +275,7 @@ func (w *AsyncDecisionWorker) createSingleDecisionForObjectId(
 		return false, nil, nil
 	}
 
-	object := models.ClientObject{TableName: table.Name, Data: objectMap[0]}
+	object := models.ClientObject{TableName: table.Name, Data: objectMap[0].Data}
 
 	evaluationParameters := evaluate_scenario.ScenarioEvaluationParameters{
 		Scenario:          scenario,

diff --git a/usecases/transfers_data_read/transfers_data_read.go b/usecases/transfers_data_read/transfers_data_read.go
@@ -70,7 +70,7 @@ func (usecase TransferDataReader) QueryTransferDataFromMapping(
 		return make([]models.TransferData, 0), nil
 	}
 
-	readPartnerId, transferData := presentTransferData(ctx, objects[0])
+	readPartnerId, transferData := presentTransferData(ctx, objects[0].Data)
 	if err := usecase.enforceSecurity.ReadTransferData(ctx, readPartnerId); err != nil {
 		return nil, err
 	}

diff --git a/usecases/usecases_with_creds.go b/usecases/usecases_with_creds.go
@@ -480,3 +480,11 @@ func (usecases UsecasesWithCreds) NewNewAsyncScheduledExecWorker() *scheduled_ex
 	)
 	return &w
 }
+
+func (usecases UsecasesWithCreds) NewIngestedDataReaderUsecase() IngestedDataReaderUsecase {
+	return IngestedDataReaderUsecase{
+		ingestedDataReadRepository: usecases.Repositories.IngestedDataReadRepository,
+		dataModelRepository:        usecases.Repositories.DataModelRepository,
+		executorFactory:            usecases.NewExecutorFactory(),
+	}
+}