DirectoryHelper with Glob support; GCS NewWriter should truncate

files/factory.go

@@ -39,3 +39,30 @@ func (f *Factory) Get(uri string) (FileHelper, error) {
 		return nil, errors.Errorf("Scheme %v is not supported", u.Scheme)
 	}
 }
+
+// GetDirHelper returns the correct DirectoryHelper based on a files scheme
+func (f *Factory) GetDirHelper(uri string) (DirectoryHelper, error) {
+	u, err := url.Parse(uri)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Failed to parse URI %v", uri)
+	}
+
+	switch u.Scheme {
+	case "":
+		return &LocalFileHelper{}, nil
+	case GCSScheme:
+		ctx := context.Background()
+		client, err := storage.NewClient(ctx)
+		if err != nil {
+			return nil, errors.Wrapf(err, "Failed to create GCS storage client")
+		}
+		return &gcs.GcsHelper{
+			Ctx:    ctx,
+			Client: client,
+		}, nil
+	case FileScheme:
+		return &LocalFileHelper{}, nil
+	default:
+		return nil, errors.Errorf("Scheme %v is not supported", u.Scheme)
+	}
+}

files/interface.go

@@ -11,5 +11,11 @@ import (
 type FileHelper interface {
 	Exists(path string) (bool, error)
 	NewReader(path string) (io.Reader, error)
+	// TODO(jlewi): Should the return type be io.WriteCloser?
 	NewWriter(path string) (io.Writer, error)
 }
+
+type DirectoryHelper interface {
+	FileHelper
+	Glob(pattern string) ([]string, error)
+}

files/local.go

@@ -3,6 +3,7 @@ package files
 import (
 	"io"
 	"os"
+	"path/filepath"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -50,3 +51,8 @@ func (h *LocalFileHelper) Exists(uri string) (bool, error) {
 	}
 	return true, nil
 }
+
+// Glob returns the list of files that match the pattern.
+func (h *LocalFileHelper) Glob(uri string) ([]string, error) {
+	return filepath.Glob(uri)
+}

gcp/credentials.go

@@ -10,6 +10,8 @@ import (
 	"path/filepath"
 	"strings"
 
+	"github.com/jlewi/monogo/files"
+
 	"cloud.google.com/go/storage"
 	"github.com/go-logr/logr"
 	"github.com/go-logr/zapr"
@@ -43,7 +45,7 @@ type WebFlowHelper struct {
 // NewWebFlowHelper constructs a new web flow helper. oAuthClientFile should be the path to a credentials.json
 // downloaded from the API console.
 func NewWebFlowHelper(oAuthClientFile string, scopes []string) (*WebFlowHelper, error) {
-	var fHelper gcs.FileHelper
+	var fHelper files.FileHelper
 
 	if strings.HasPrefix(oAuthClientFile, "gs://") {
 		ctx := context.Background()
@@ -58,7 +60,7 @@ func NewWebFlowHelper(oAuthClientFile string, scopes []string) (*WebFlowHelper,
 			Client: client,
 		}
 	} else {
-		fHelper = &gcs.LocalFileHelper{}
+		fHelper = &files.LocalFileHelper{}
 	}
 
 	reader, err := fHelper.NewReader(oAuthClientFile)

gcp/gcs/util.go → gcp/gcs/helper.go

@@ -92,11 +92,6 @@ func (h *GcsHelper) NewWriter(uri string) (io.Writer, error) {
 
 	o := b.Object(p.Path)
 
-	// Make sure object doesn't already exist
-	if ObjectExists(h.Ctx, o) {
-		return nil, errors.WithStack(errors.Errorf("Can't write %v; It already exists", uri))
-	}
-
 	return o.NewWriter(h.Ctx), nil
 }
 
@@ -131,6 +126,43 @@ func (h *GcsHelper) Exists(uri string) (bool, error) {
 	return ObjectExists(h.Ctx, o), nil
 }
 
+// Glob lists all objects matching some glob expression.
+
+func (h *GcsHelper) Glob(uri string) ([]string, error) {
+	paths := []string{}
+	p, err := Parse(uri)
+	if err != nil {
+		return paths, errors.WithStack(errors.Wrapf(err, "Could not glob objects matching %v", uri))
+	}
+
+	b := h.Client.Bucket(p.Bucket)
+
+	q := &storage.Query{
+		MatchGlob: p.Path,
+	}
+
+	objs := b.Objects(h.Ctx, q)
+
+	for {
+		i, err := objs.Next()
+
+		if err == iterator.Done {
+			return paths, nil
+		}
+
+		if err != nil {
+			return paths, errors.WithStack(errors.Wrapf(err, "Error getting next object matching %v", uri))
+		}
+
+		iPath := GcsPath{
+			Bucket: i.Bucket,
+			Path:   i.Name,
+		}
+
+		paths = append(paths, iPath.ToURI())
+	}
+}
+
 // BuildInputOutputList builds a map from input files to the files that they
 // should be mapped to.
 //

gcp/gcs/util_test.go → gcp/gcs/helper_test.go

@@ -1,6 +1,9 @@
 package gcs
 
 import (
+	"context"
+	"io"
+	"os"
 	"regexp"
 	"testing"
 
@@ -146,3 +149,52 @@ func TestFindMatches(t *testing.T) {
 		}
 	}
 }
+
+func Test_Glob(t *testing.T) {
+	if os.Getenv("GITHUB_ACTIONS") != "" {
+		t.Skip("Skipping test in GitHub Actions")
+	}
+
+	files := []string{
+		"gs://foyle-dev-mongo-testing/somefile/file-1.txt",
+		"gs://foyle-dev-mongo-testing/somefile/file-2.txt",
+		"gs://foyle-dev-mongo-testing/somefile/file-22.txt",
+		"gs://foyle-dev-mongo-testing/somefile/otherfile.txt",
+	}
+
+	ctx := context.Background()
+	client, err := storage.NewClient(ctx)
+	if err != nil {
+		t.Fatalf("Failed to create GCS storage client; error: %+v", err)
+	}
+
+	h := &GcsHelper{
+		Ctx:    ctx,
+		Client: client,
+	}
+
+	// Create the files
+	for _, f := range files {
+		w, err := h.NewWriter(f)
+		if err != nil {
+			t.Fatalf("Could not create writer for %v; error %v", f, err)
+		}
+		closer := w.(io.WriteCloser)
+		if err := closer.Close(); err != nil {
+			t.Fatalf("Could not close writer for %v; error %v", f, err)
+		}
+	}
+
+	expected := []string{
+		"gs://foyle-dev-mongo-testing/somefile/file-1.txt",
+		"gs://foyle-dev-mongo-testing/somefile/file-2.txt",
+	}
+
+	actual, err := h.Glob("gs://foyle-dev-mongo-testing/somefile/file-?.txt")
+	if err != nil {
+		t.Fatalf("Glob returned error %v", err)
+	}
+	if d := cmp.Diff(expected, actual); d != "" {
+		t.Errorf("Glob() mismatch (-want +got):\n%s", d)
+	}
+}