feat: add filtering in assets api

.gitignore

@@ -10,6 +10,7 @@
 
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out
+coverage.txt
 
 # Dependency directories (remove the comment below to include it)
 vendor/

api/httpapi/handlers/asset_handler.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"strings"
 
 	"github.com/gorilla/mux"
 	"github.com/odpf/salt/log"
@@ -19,6 +20,10 @@ import (
 	"github.com/odpf/columbus/user"
 )
 
+var (
+	dataFilterPrefix = "data"
+)
+
 // AssetHandler exposes a REST interface to types
 type AssetHandler struct {
 	logger        log.Logger
@@ -47,8 +52,13 @@ func NewAssetHandler(
 }
 
 func (h *AssetHandler) GetAll(w http.ResponseWriter, r *http.Request) {
-	config := h.buildAssetConfig(r.URL.Query())
-	assets, err := h.assetRepo.GetAll(r.Context(), config)
+	cfg, err := h.buildAssetConfig(r.URL.Query())
+	if err != nil {
+		WriteJSONError(w, http.StatusBadRequest, bodyParserErrorMsg(err))
+		return
+	}
+
+	assets, err := h.assetRepo.GetAll(r.Context(), cfg)
 	if err != nil {
 		internalServerError(w, h.logger, err.Error())
 		return
@@ -61,9 +71,15 @@ func (h *AssetHandler) GetAll(w http.ResponseWriter, r *http.Request) {
 	withTotal, ok := r.URL.Query()["with_total"]
 	if ok && len(withTotal) > 0 && withTotal[0] != "false" && withTotal[0] != "0" {
 		total, err := h.assetRepo.GetCount(r.Context(), asset.Config{
-			Type:    config.Type,
-			Service: config.Service,
-			Text:    config.Text,
+			Types:         cfg.Types,
+			Services:      cfg.Services,
+			Size:          cfg.Size,
+			Offset:        cfg.Offset,
+			SortBy:        cfg.SortBy,
+			SortDirection: cfg.SortDirection,
+			QueryFields:   cfg.QueryFields,
+			Query:         cfg.Query,
+			Data:          cfg.Data,
 		})
 		if err != nil {
 			internalServerError(w, h.logger, err.Error())
@@ -253,12 +269,16 @@ func (h *AssetHandler) GetStargazers(w http.ResponseWriter, r *http.Request) {
 }
 
 func (h *AssetHandler) GetVersionHistory(w http.ResponseWriter, r *http.Request) {
-	config := h.buildAssetConfig(r.URL.Query())
+	cfg, err := h.buildAssetConfig(r.URL.Query())
+	if err != nil {
+		WriteJSONError(w, http.StatusBadRequest, bodyParserErrorMsg(err))
+		return
+	}
 
 	pathParams := mux.Vars(r)
 	assetID := pathParams["id"]
 
-	assetVersions, err := h.assetRepo.GetVersionHistory(r.Context(), config, assetID)
+	assetVersions, err := h.assetRepo.GetVersionHistory(r.Context(), cfg, assetID)
 	if err != nil {
 		if errors.As(err, new(asset.InvalidError)) {
 			WriteJSONError(w, http.StatusBadRequest, err.Error())
@@ -363,29 +383,88 @@ func (h *AssetHandler) validatePatchPayload(assetPayload map[string]interface{})
 	return
 }
 
-func (h *AssetHandler) buildAssetConfig(query url.Values) asset.Config {
-	config := asset.Config{
-		Text:    query.Get("text"),
-		Type:    asset.Type(query.Get("type")),
-		Service: query.Get("service"),
+func (h *AssetHandler) buildAssetConfig(query url.Values) (cfg asset.Config, err error) {
+	cfg = asset.Config{
+		SortBy:        query.Get("sort"),
+		SortDirection: query.Get("direction"),
+		Query:         query.Get("q"),
+	}
+
+	types := query.Get("types")
+	if types != "" {
+		typ := strings.Split(types, ",")
+		for _, typeVal := range typ {
+			cfg.Types = append(cfg.Types, asset.Type(typeVal))
+		}
+	}
+
+	services := query.Get("services")
+	if services != "" {
+		cfg.Services = strings.Split(services, ",")
+	}
+
+	queriesFields := query.Get("q_fields")
+	if queriesFields != "" {
+		cfg.QueryFields = strings.Split(queriesFields, ",")
 	}
 
 	sizeString := query.Get("size")
 	if sizeString != "" {
 		size, err := strconv.Atoi(sizeString)
 		if err == nil {
-			config.Size = size
+			cfg.Size = size
 		}
 	}
+
 	offsetString := query.Get("offset")
 	if offsetString != "" {
 		offset, err := strconv.Atoi(offsetString)
 		if err == nil {
-			config.Offset = offset
+			cfg.Offset = offset
 		}
 	}
 
-	return config
+	cfg.Data = dataAssetConfigValue(query)
+	if err = cfg.Validate(); err != nil {
+		return asset.Config{}, err
+	}
+
+	return cfg, nil
+}
+
+func dataAssetConfigValue(queryString url.Values) map[string]string {
+	dataFilter := make(map[string]string)
+	preChar := "["
+	postChar := "]"
+
+	// Get substring between two strings.
+	for key, values := range queryString {
+		if !strings.HasPrefix(key, dataFilterPrefix) {
+			continue
+		}
+
+		posFirst := strings.Index(key, preChar)
+		if posFirst == -1 {
+			return nil
+		}
+		posLast := strings.Index(key, postChar)
+		if posLast == -1 {
+			return nil
+		}
+		posFirstAdjusted := posFirst + len(preChar)
+		if posFirstAdjusted >= posLast {
+			return nil
+		}
+
+		filterKey := key[posFirstAdjusted:posLast]
+		dataFilter[filterKey] = values[0] // cannot have duplicate query key, always get the first one
+	}
+
+	if len(dataFilter) == 0 {
+		return nil
+	}
+
+	return dataFilter
 }
 
 func (h *AssetHandler) saveLineage(ctx context.Context, ast asset.Asset, upstreams, downstreams []lineage.Node) error {

api/httpapi/handlers/asset_handler_test.go

@@ -690,15 +690,72 @@ func TestAssetHandlerGet(t *testing.T) {
 		},
 		{
 			Description:  `should parse querystring to get config`,
-			Querystring:  "?text=asd&type=table&service=bigquery&size=30&offset=50",
+			Querystring:  "?types=table&services=bigquery&size=30&offset=50&sort=created_at&direction=desc&data[dataset]=booking&data[project]=p-godata-id&q=internal&q_fields=name,urn",
 			ExpectStatus: http.StatusOK,
 			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
 				ar.On("GetAll", ctx, asset.Config{
-					Text:    "asd",
-					Type:    "table",
-					Service: "bigquery",
-					Size:    30,
-					Offset:  50,
+					Types:         []asset.Type{"table"},
+					Services:      []string{"bigquery"},
+					Size:          30,
+					Offset:        50,
+					SortDirection: "desc",
+					SortBy:        "created_at",
+					Data: map[string]string{
+						"dataset": "booking",
+						"project": "p-godata-id",
+					},
+					Query:       "internal",
+					QueryFields: []string{"name", "urn"},
+				}).Return([]asset.Asset{}, nil, nil)
+			},
+		},
+		{
+			Description:  `should parse data and query fields querystring to get config`,
+			Querystring:  "?data[dataset]=booking&data[project]=p-godata-id&q=internal&q_fields=name,urn,description,services",
+			ExpectStatus: http.StatusOK,
+			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
+				ar.On("GetAll", ctx, asset.Config{
+					Data: map[string]string{
+						"dataset": "booking",
+						"project": "p-godata-id",
+					},
+					Query:       "internal",
+					QueryFields: []string{"name", "urn", "description", "services"},
+				}).Return([]asset.Asset{}, nil, nil)
+			},
+		},
+		{
+			Description:  `should parse data fields querystring to get config`,
+			Querystring:  "?data[dataset]=booking&data[project]=p-godata-id",
+			ExpectStatus: http.StatusOK,
+			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
+				ar.On("GetAll", ctx, asset.Config{
+					Data: map[string]string{
+						"dataset": "booking",
+						"project": "p-godata-id",
+					},
+				}).Return([]asset.Asset{}, nil, nil)
+			},
+		},
+		{
+			Description:  `should parse query fields querystring to get config`,
+			Querystring:  "?q=internal&q_fields=name,urn,description,services",
+			ExpectStatus: http.StatusOK,
+			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
+				ar.On("GetAll", ctx, asset.Config{
+					Query:       "internal",
+					QueryFields: []string{"name", "urn", "description", "services"},
+				}).Return([]asset.Asset{}, nil, nil)
+			},
+		},
+		{
+			Description:  "should convert multiple types and services from querystring to config",
+			Querystring:  "?types=table,job&services=bigquery,kafka",
+			ExpectStatus: http.StatusOK,
+			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
+				ar.On("GetAll", ctx, asset.Config{
+					Types:    []asset.Type{"table", "job"},
+					Services: []string{"bigquery", "kafka"},
 				}).Return([]asset.Asset{}, nil, nil)
 			},
 		},
@@ -736,23 +793,23 @@ func TestAssetHandlerGet(t *testing.T) {
 		{
 			Description:  "should return total in the payload if with_total flag is given",
 			ExpectStatus: http.StatusOK,
-			Querystring:  "?with_total=true&text=dsa&type=job&service=kafka&size=10&offset=5",
+			Querystring:  "?with_total=true&types=job&services=kafka&size=10&offset=5",
 			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
 				ar.On("GetAll", ctx, asset.Config{
-					Text:    "dsa",
-					Type:    "job",
-					Service: "kafka",
-					Size:    10,
-					Offset:  5,
+					Types:    []asset.Type{"job"},
+					Services: []string{"kafka"},
+					Size:     10,
+					Offset:   5,
 				}).Return([]asset.Asset{
 					{ID: "testid-1"},
 					{ID: "testid-2"},
 					{ID: "testid-3"},
 				}, nil, nil)
 				ar.On("GetCount", ctx, asset.Config{
-					Text:    "dsa",
-					Type:    "job",
-					Service: "kafka",
+					Size:     10,
+					Offset:   5,
+					Types:    []asset.Type{"job"},
+					Services: []string{"kafka"},
 				}).Return(150, nil, nil)
 			},
 			PostCheck: func(r *http.Response) error {
@@ -959,7 +1016,6 @@ func TestAssetHandlerGetVersionHistory(t *testing.T) {
 				"id": assetID,
 			})
 			rw := httptest.NewRecorder()
-
 			ar := new(mocks.AssetRepository)
 			tc.Setup(rr.Context(), ar)
 

api/v1beta1/asset.go

@@ -15,15 +15,16 @@ import (
 )
 
 func (h *Handler) GetAllAssets(ctx context.Context, req *compassv1beta1.GetAllAssetsRequest) (*compassv1beta1.GetAllAssetsResponse, error) {
-	config := asset.Config{
+	config := asset.GRPCConfig{
 		Text:    req.GetText(),
 		Type:    asset.Type(req.GetType()),
 		Service: req.GetService(),
 		Size:    int(req.GetSize()),
 		Offset:  int(req.GetOffset()),
 	}
+	cfg := config.ToConfig()
 
-	assets, err := h.AssetRepository.GetAll(ctx, config)
+	assets, err := h.AssetRepository.GetAll(ctx, cfg)
 	if err != nil {
 		return nil, internalServerError(h.Logger, err.Error())
 	}
@@ -42,11 +43,14 @@ func (h *Handler) GetAllAssets(ctx context.Context, req *compassv1beta1.GetAllAs
 	}
 
 	if req.GetWithTotal() {
-		total, err := h.AssetRepository.GetCount(ctx, asset.Config{
+		grpcConfig := asset.GRPCConfig{
 			Type:    config.Type,
 			Service: config.Service,
 			Text:    config.Text,
-		})
+		}
+		cfg = grpcConfig.ToConfig()
+
+		total, err := h.AssetRepository.GetCount(ctx, cfg)
 		if err != nil {
 			return nil, internalServerError(h.Logger, err.Error())
 		}

api/v1beta1/asset_test.go

@@ -64,13 +64,15 @@ func TestGetAllAssets(t *testing.T) {
 			},
 			ExpectStatus: codes.OK,
 			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
-				ar.On("GetAll", ctx, asset.Config{
+				cfg := asset.GRPCConfig{
 					Text:    "asd",
 					Type:    "table",
 					Service: "bigquery",
 					Size:    30,
 					Offset:  50,
-				}).Return([]asset.Asset{}, nil, nil)
+				}
+				config := cfg.ToConfig()
+				ar.On("GetAll", ctx, config).Return([]asset.Asset{}, nil, nil)
 			},
 		},
 		{
@@ -108,22 +110,22 @@ func TestGetAllAssets(t *testing.T) {
 				WithTotal: true,
 			},
 			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
-				ar.On("GetAll", ctx, asset.Config{
+				ar.On("GetAll", ctx, asset.GRPCConfig{
 					Text:    "dsa",
 					Type:    "job",
 					Service: "kafka",
 					Size:    10,
 					Offset:  5,
-				}).Return([]asset.Asset{
+				}.ToConfig()).Return([]asset.Asset{
 					{ID: "testid-1"},
 					{ID: "testid-2"},
 					{ID: "testid-3"},
 				}, nil, nil)
-				ar.On("GetCount", ctx, asset.Config{
+				ar.On("GetCount", ctx, asset.GRPCConfig{
 					Text:    "dsa",
 					Type:    "job",
 					Service: "kafka",
-				}).Return(150, nil, nil)
+				}.ToConfig()).Return(150, nil, nil)
 			},
 			PostCheck: func(resp *compassv1beta1.GetAllAssetsResponse) error {
 				expected := &compassv1beta1.GetAllAssetsResponse{

asset/asset.go

@@ -13,14 +13,6 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-type Config struct {
-	Text    string `json:"text"`
-	Type    Type   `json:"type"`
-	Service string `json:"service"`
-	Size    int    `json:"size"`
-	Offset  int    `json:"offset"`
-}
-
 type Repository interface {
 	GetAll(context.Context, Config) ([]Asset, error)
 	GetCount(context.Context, Config) (int, error)