func (repo *resourceRepository) GetByName(ctx context.Context, jobName string) (models.ResourceSpec, error) {
	if strings.TrimSpace(jobName) == "" {
		return models.ResourceSpec{}, errors.Errorf("resource name cannot be an empty string")
	}
	if repo.cache.dirty {
		if err := repo.refreshCache(); err != nil {
			return models.ResourceSpec{}, err
		}
	}
	blob, ok := repo.cache.data[jobName]
	if !ok {
		return models.ResourceSpec{}, models.ErrNoSuchSpec
	}
	return blob.item.(models.ResourceSpec), nil
}
GetByName returns a job requested by the name
package local
import (
	"context"
	"fmt"
	"io/ioutil"
	"os"
	"path/filepath"
	"strings"
	"github.com/spf13/afero"
	"github.com/odpf/optimus/models"
	"github.com/pkg/errors"
	"gopkg.in/yaml.v2"
)
const (
	ResourceSpecFileName = "resource.yaml"
)
type Resource struct {
	Version int
	Name    string
	Type    string
	Spec    interface{}
	Labels  map[string]string
}
type resourceRepository struct {
	fs    afero.Fs
	cache struct {
		dirty bool
		data map[string]cacheItem
	}
	ds models.Datastorer
}
func (repo *resourceRepository) SaveAt(resourceSpec models.ResourceSpec, rootDir string) error {
	if len(resourceSpec.Name) == 0 || len(resourceSpec.Type) == 0 {
		return fmt.Errorf("resource is missing required fields")
	}
	typeController, _ := resourceSpec.Datastore.Types()[resourceSpec.Type]
	specBytes, err := typeController.Adapter().ToYaml(resourceSpec)
	if err != nil {
		return err
	}
	if rootDir == "" {
		rootDir = resourceSpec.Name
	}
	if err = repo.fs.MkdirAll(repo.assetFolderPath(rootDir), os.FileMode(0765)|os.ModeDir); err != nil {
		return errors.Wrapf(err, "repo.fs.MkdirAll: %s", rootDir)
	}
	for assetName, assetValue := range resourceSpec.Assets {
		if err := afero.WriteFile(repo.fs, repo.assetFilePath(rootDir, assetName), []byte(assetValue), os.FileMode(0755)); err != nil {
			return errors.Wrapf(err, "WriteFile.Asset: %s", repo.assetFilePath(rootDir, assetName))
		}
	}
	if afero.WriteFile(repo.fs, repo.resourceFilePath(rootDir), specBytes, os.FileMode(0755)); err != nil {
		return err
	}
	repo.cache.dirty = true
	return nil
}
func (repo *resourceRepository) Save(ctx context.Context, resourceSpec models.ResourceSpec) error {
	if resourceSpec.Name == "" {
		return errors.New("invalid job name")
	}
	if repo.cache.dirty {
		if err := repo.refreshCache(); err != nil {
			return err
		}
	}
	specDir := ""
	existingJob, ok := repo.cache.data[resourceSpec.Name]
	if ok {
		specDir = existingJob.path
	}
	return repo.SaveAt(resourceSpec, specDir)
}
func (repo *resourceRepository) GetAll(ctx context.Context) ([]models.ResourceSpec, error) {
	var resourceSpecs []models.ResourceSpec
	if repo.cache.dirty {
		if err := repo.refreshCache(); err != nil {
			return resourceSpecs, err
		}
	}
	for _, j := range repo.cache.data {
		resourceSpecs = append(resourceSpecs, j.item.(models.ResourceSpec))
	}
	if len(resourceSpecs) < 1 {
		return nil, models.ErrNoResources
	}
	return resourceSpecs, nil
}