feat(task_repo): set default version to the latest

Signed-off-by: featherchen <[email protected]>

flyteadmin/pkg/manager/impl/validation/validation.go

@@ -95,6 +95,23 @@ func ValidateIdentifierFieldsSet(id *core.Identifier) error {
 	return nil
 }
 
+// ValidateTaskIdentifierFieldsSet Validates that all required fields, except version, for a task identifier are present.
+func ValidateTaskIdentifierFieldsSet(id *core.Identifier) error {
+	if id == nil {
+		return shared.GetMissingArgumentError(shared.ID)
+	}
+	if err := ValidateEmptyStringField(id.Project, shared.Project); err != nil {
+		return err
+	}
+	if err := ValidateEmptyStringField(id.Domain, shared.Domain); err != nil {
+		return err
+	}
+	if err := ValidateEmptyStringField(id.Name, shared.Name); err != nil {
+		return err
+	}
+	return nil
+}
+
 // ValidateIdentifier Validates that all required fields for an identifier are present.
 func ValidateIdentifier(id *core.Identifier, expectedType common.Entity) error {
 	if id == nil {
@@ -105,6 +122,9 @@ func ValidateIdentifier(id *core.Identifier, expectedType common.Entity) error {
 			"unexpected resource type %s for identifier [%+v], expected %s instead",
 			strings.ToLower(id.ResourceType.String()), id, strings.ToLower(entityToResourceType[expectedType].String()))
 	}
+	if id.ResourceType == core.ResourceType_TASK {
+		return ValidateTaskIdentifierFieldsSet(id)
+	}
 	return ValidateIdentifierFieldsSet(id)
 }
 

flyteadmin/pkg/repositories/gormimpl/task_repo.go

@@ -49,14 +49,20 @@ func (r *TaskRepo) Create(ctx context.Context, input models.Task, descriptionEnt
 func (r *TaskRepo) Get(ctx context.Context, input interfaces.Identifier) (models.Task, error) {
 	var task models.Task
 	timer := r.metrics.GetDuration.Start()
-	tx := r.db.WithContext(ctx).Where(&models.Task{
+	query := r.db.WithContext(ctx).Where(&models.Task{
 		TaskKey: models.TaskKey{
 			Project: input.Project,
 			Domain:  input.Domain,
 			Name:    input.Name,
-			Version: input.Version,
 		},
-	}).Take(&task)
+	})
+
+	if input.Version == "" {
+		query = query.Order("version DESC").Limit(1)
+	} else {
+		query = query.Where("version = ?", input.Version)
+	}
+	tx := query.Take(&task)
 	timer.Stop()
 	if errors.Is(tx.Error, gorm.ErrRecordNotFound) {
 		return models.Task{}, flyteAdminDbErrors.GetMissingEntityError(core.ResourceType_TASK.String(), &core.Identifier{