feat: update types filter

Author: scortier

api/httpapi/handlers/asset_handler.go

@@ -71,9 +71,9 @@ 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:    cfg.Type,
-			Service: cfg.Service,
-			Text:    cfg.Text,
+			Types:    cfg.Types,
+			Services: cfg.Services,
+			Text:     cfg.Text,
 		})
 		if err != nil {
 			internalServerError(w, h.logger, err.Error())
@@ -321,25 +321,21 @@ func (h *AssetHandler) validateAsset(ast asset.Asset) error {
 	if ast.URN == "" {
 		return fmt.Errorf("urn is required")
 	}
-	if ast.Type == "" {
+	if ast.Types != nil {
 		return fmt.Errorf("type is required")
 	}
-	if !ast.Type.IsValid() {
-		return fmt.Errorf("type is invalid")
-	}
+	//if !ast.Types.IsValid() {
+	//	return fmt.Errorf("type is invalid")
+	//}
 	if ast.Name == "" {
 		return fmt.Errorf("name is required")
 	}
 	if ast.Data == nil {
 		return fmt.Errorf("data is required")
 	}
-
-	if ast.Service == "" {
+	if ast.Services != nil {
 		return fmt.Errorf("service is required")
 	}
-	//if !ast.Service.IsValid() {
-	//	return fmt.Errorf("service is invalid")
-	//}
 
 	return nil
 }
@@ -400,11 +396,11 @@ func (h *AssetHandler) buildAssetConfig(query url.Values) (cfg asset.Config, err
 
 	types := query.Get("type")
 	if types != "" {
-		cfg.Type = strings.Split(types, ",")
+		cfg.Types = strings.Split(types, ",")
 	}
 	services := query.Get("service")
 	if services != "" {
-		cfg.Service = strings.Split(services, ",")
+		cfg.Services = strings.Split(services, ",")
 	}
 
 	cfg.Name = query.Get("name")
@@ -466,8 +462,8 @@ func filterConfigFromAssetValues(querystring url.Values) map[string][]string {
 func (h *AssetHandler) saveLineage(ctx context.Context, ast asset.Asset, upstreams, downstreams []lineage.Node) error {
 	node := lineage.Node{
 		URN:     ast.URN,
-		Type:    ast.Type,
-		Service: ast.Service,
+		Type:    ast.Types,
+		Service: ast.Services,
 	}
 
 	return h.lineageRepo.Upsert(ctx, node, upstreams, downstreams)

api/httpapi/handlers/asset_handler_test.go

@@ -192,9 +192,9 @@ func TestAssetHandlerUpsert(t *testing.T) {
 	t.Run("should return HTTP 200 and asset's ID if the asset is successfully created/updated", func(t *testing.T) {
 		ast := asset.Asset{
 			URN:       "test dagger",
-			Type:      asset.TypeTable,
+			Types:     asset.TypeTable,
 			Name:      "de-dagger-test",
-			Service:   "kafka",
+			Services:  []string{"kafka"},
 			UpdatedBy: user.User{ID: userID},
 			Data:      map[string]interface{}{},
 		}
@@ -229,8 +229,8 @@ func TestAssetHandlerUpsert(t *testing.T) {
 		lr.On("Upsert", rr.Context(),
 			lineage.Node{
 				URN:     ast.URN,
-				Type:    ast.Type,
-				Service: ast.Service,
+				Type:    ast.Types,
+				Service: ast.Services,
 			},
 			upstreams,
 			downstreams,
@@ -695,8 +695,8 @@ func TestAssetHandlerGet(t *testing.T) {
 			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
 				ar.On("GetAll", ctx, asset.Config{
 					Text:          "asd",
-					Type:          []string{"table", "job"},
-					Service:       []string{"bigquery", "presto"},
+					Types:         []string{"table", "job"},
+					Services:      []string{"bigquery", "presto"},
 					Size:          30,
 					Offset:        50,
 					SortDirection: "desc",
@@ -741,20 +741,20 @@ func TestAssetHandlerGet(t *testing.T) {
 			Querystring:  "?with_total=true&text=dsa&type=job&service=kafka&size=10&offset=5",
 			Setup: func(ctx context.Context, ar *mocks.AssetRepository) {
 				ar.On("GetAll", ctx, asset.Config{
-					Text:    "dsa",
-					Type:    []string{"job"},
-					Service: []string{"kafka"},
-					Size:    10,
-					Offset:  5,
+					Text:     "dsa",
+					Types:    []string{"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:    []string{"job"},
-					Service: []string{"kafka"},
+					Text:     "dsa",
+					Types:    []string{"job"},
+					Services: []string{"kafka"},
 				}).Return(150, nil, nil)
 			},
 			PostCheck: func(r *http.Response) error {

api/httpapi/handlers/record_handler.go

@@ -233,7 +233,7 @@ func (h *RecordHandler) validateRecord(ast asset.Asset) error {
 	if ast.Data == nil {
 		return fmt.Errorf("data is required")
 	}
-	if ast.Service == "" {
+	if ast.Services != nil {
 		return fmt.Errorf("service is required")
 	}
 

asset/asset.go

@@ -27,8 +27,8 @@ type Repository interface {
 type Asset struct {
 	ID          string                 `json:"id" diff:"-"`
 	URN         string                 `json:"urn" diff:"-"`
-	Type        Type                   `json:"type" diff:"-"`
-	Service     string                 `json:"service" diff:"-"`
+	Types       []Type                 `json:"type" diff:"-"`
+	Services    []string               `json:"service" diff:"-"`
 	Name        string                 `json:"name" diff:"name"`
 	Description string                 `json:"description" diff:"description"`
 	Data        map[string]interface{} `json:"data" diff:"data"`

asset/config.go

@@ -22,19 +22,19 @@ func init() {
 }
 
 type RecordFilter map[string][]string
+type ConfigType []Type
 
 type Config struct {
-	Text          string   `json:"text"`
-	Type          []string `json:"type"`
-	Service       []string `json:"service"`
-	Size          int      `json:"size"`
-	Offset        int      `json:"offset"`
-	SortBy        string   `json:"sort" validate:"omitempty,oneof=recent created_at updated_at"`
-	SortDirection string   `json:"direction" validate:"omitempty,oneof=asc desc"`
+	Text          string
+	Types         ConfigType
+	Services      []string
+	Size          int
+	Offset        int
+	SortBy        string `validate:"omitempty,oneof=name types services created_at updated_at"`
+	SortDirection string `validate:"omitempty,oneof=asc desc"`
 	Filters       RecordFilter
-	TypeWhiteList []string
-	URN           string `json:"urn"`
-	Name          string `json:"name"`
+	URN           string
+	Name          string
 }
 
 // Validate will check whether fields in the filter fulfills the constraint

asset/version.go

@@ -15,8 +15,8 @@ const BaseVersion = "0.1"
 type AssetVersion struct {
 	ID        string         `json:"id" db:"id"`
 	URN       string         `json:"urn" db:"urn"`
-	Type      string         `json:"type" db:"type"`
-	Service   string         `json:"service" db:"service"`
+	Types     []string       `json:"type" db:"type"`
+	Services  []string       `json:"service" db:"service"`
 	Version   string         `json:"version" db:"version"`
 	Changelog diff.Changelog `json:"changelog" db:"changelog"`
 	UpdatedBy user.User      `json:"updated_by" db:"updated_by"`

store/postgres/asset_model.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/lib/pq"
 	"time"
 
 	"github.com/jmoiron/sqlx/types"
@@ -14,18 +15,18 @@ import (
 )
 
 type AssetModel struct {
-	ID          string    `db:"id"`
-	URN         string    `db:"urn"`
-	Type        string    `db:"type"`
-	Name        string    `db:"name"`
-	Service     string    `db:"service"`
-	Description string    `db:"description"`
-	Data        JSONMap   `db:"data"`
-	Labels      JSONMap   `db:"labels"`
-	Version     string    `db:"version"`
-	UpdatedBy   UserModel `db:"updated_by"`
-	CreatedAt   time.Time `db:"created_at"`
-	UpdatedAt   time.Time `db:"updated_at"`
+	ID          string         `db:"id"`
+	URN         string         `db:"urn"`
+	Types       []string       `db:"type"`
+	Name        string         `db:"name"`
+	Services    pq.StringArray `db:"service"`
+	Description string         `db:"description"`
+	Data        JSONMap        `db:"data"`
+	Labels      JSONMap        `db:"labels"`
+	Version     string         `db:"version"`
+	UpdatedBy   UserModel      `db:"updated_by"`
+	CreatedAt   time.Time      `db:"created_at"`
+	UpdatedAt   time.Time      `db:"updated_at"`
 	// version specific information
 	Changelog types.JSONText `db:"changelog"`
 	Owners    types.JSONText `db:"owners"`
@@ -36,9 +37,9 @@ func (a *AssetModel) toAsset(owners []user.User) asset.Asset {
 	return asset.Asset{
 		ID:          a.ID,
 		URN:         a.URN,
-		Type:        asset.Type(a.Type),
+		Types:       asset.Type(a.Types),
 		Name:        a.Name,
-		Service:     a.Service,
+		Services:    a.Services,
 		Description: a.Description,
 		Data:        a.Data,
 		Labels:      a.buildLabels(),
@@ -61,8 +62,8 @@ func (a *AssetModel) toAssetVersion() (asset.AssetVersion, error) {
 	return asset.AssetVersion{
 		ID:        a.ID,
 		URN:       a.URN,
-		Type:      a.Type,
-		Service:   a.Service,
+		Types:     a.Types,
+		Services:  a.Services,
 		Version:   a.Version,
 		Changelog: clog,
 		UpdatedBy: a.UpdatedBy.toUser(),
@@ -86,9 +87,9 @@ func (a *AssetModel) toVersionedAsset(latestAssetVersion asset.Asset) (asset.Ass
 	return asset.Asset{
 		ID:          latestAssetVersion.ID,
 		URN:         latestAssetVersion.URN,
-		Type:        asset.Type(latestAssetVersion.Type),
+		Types:       latestAssetVersion.Types,
 		Name:        a.Name,
-		Service:     latestAssetVersion.Service,
+		Services:    latestAssetVersion.Services,
 		Description: a.Description,
 		Data:        a.Data,
 		Labels:      a.buildLabels(),

store/postgres/asset_repository.go

@@ -300,7 +300,7 @@ func (r *AssetRepository) insert(ctx context.Context, ast *asset.Asset) (id stri
 	err = r.client.RunWithinTx(ctx, func(tx *sqlx.Tx) error {
 		query, args, err := sq.Insert("assets").
 			Columns("urn", "type", "service", "name", "description", "data", "labels", "updated_by", "version").
-			Values(ast.URN, ast.Type, ast.Service, ast.Name, ast.Description, ast.Data, ast.Labels, ast.UpdatedBy.ID, asset.BaseVersion).
+			Values(ast.URN, ast.Types, ast.Services, ast.Name, ast.Description, ast.Data, ast.Labels, ast.UpdatedBy.ID, asset.BaseVersion).
 			Suffix("RETURNING \"id\"").
 			PlaceholderFormat(sq.Dollar).
 			ToSql()
@@ -360,7 +360,7 @@ func (r *AssetRepository) update(ctx context.Context, assetID string, newAsset *
 				version = $10
 			WHERE id = $11;
 			`,
-			newAsset.URN, newAsset.Type, newAsset.Service, newAsset.Name, newAsset.Description, newAsset.Data, newAsset.Labels, time.Now(), newAsset.UpdatedBy.ID, newVersion, assetID)
+			newAsset.URN, newAsset.Types, newAsset.Services, newAsset.Name, newAsset.Description, newAsset.Data, newAsset.Labels, time.Now(), newAsset.UpdatedBy.ID, newVersion, assetID)
 		if err != nil {
 			return fmt.Errorf("error running update asset query: %w", err)
 		}
@@ -404,7 +404,7 @@ func (r *AssetRepository) insertAssetVersion(ctx context.Context, execer sqlx.Ex
 	}
 	query, args, err := sq.Insert("assets_versions").
 		Columns("asset_id", "urn", "type", "service", "name", "description", "data", "labels", "created_at", "updated_at", "updated_by", "version", "owners", "changelog").
-		Values(oldAsset.ID, oldAsset.URN, oldAsset.Type, oldAsset.Service, oldAsset.Name, oldAsset.Description, oldAsset.Data, oldAsset.Labels,
+		Values(oldAsset.ID, oldAsset.URN, oldAsset.Types, oldAsset.Services, oldAsset.Name, oldAsset.Description, oldAsset.Data, oldAsset.Labels,
 			oldAsset.CreatedAt, oldAsset.UpdatedAt, oldAsset.UpdatedBy.ID, oldAsset.Version, oldAsset.Owners, jsonChangelog).
 		PlaceholderFormat(sq.Dollar).
 		ToSql()
@@ -636,14 +636,14 @@ func (r *AssetRepository) getAssetVersionSQL() sq.SelectBuilder {
 
 func (r *AssetRepository) buildConfigQuery(builder sq.SelectBuilder, cfg asset.Config) sq.SelectBuilder {
 	clause := sq.Eq{}
-	if len(cfg.Type) > 0 {
-		builder = builder.Where("type @> ?", cfg.Type)
+	if len(cfg.Types) > 0 {
+		builder = builder.Where("type @> ?", cfg.Types)
 	}
 	//if cfg.Type != "" {
 	//	clause["type"] = cfg.Type
 	//}
-	if len(cfg.Service) > 0 {
-		builder = builder.Where("service @> ?", cfg.Service)
+	if len(cfg.Services) > 0 {
+		builder = builder.Where("service @> ?", cfg.Services)
 	}
 	//if cfg.Service != "" {
 	//	clause["service"] = cfg.Service