diff --git a/client.go b/client.go
index 10188eec25..f68f9844e0 100644
--- a/client.go
+++ b/client.go
@@ -424,6 +424,12 @@ func (c *Client) New(ctx context.Context, cfg Function) (err error) {
 		return
 	}
 
+	// Push the produced function image
+	c.progressListener.Increment("Pushing container image to registry")
+	if err = c.Push(ctx, f.Root); err != nil {
+		return
+	}
+
 	// Deploy the initialized Function, returning its publicly
 	// addressible name for possible registration.
 	c.progressListener.Increment("Deploying Function to cluster")
@@ -602,10 +608,6 @@ func (c *Client) Deploy(ctx context.Context, path string) (err error) {
 		return ErrNotBuilt
 	}
 
-	if err = c.Push(ctx, &f); err != nil {
-		return
-	}
-
 	// Deploy a new or Update the previously-deployed Function
 	c.progressListener.Increment("Deploying function to the cluster")
 	result, err := c.deployer.Deploy(ctx, f)
@@ -717,8 +719,17 @@ func (c *Client) Emit(ctx context.Context, endpoint string) error {
 }
 
 // Push the image for the named service to the configured registry
-func (c *Client) Push(ctx context.Context, f *Function) (err error) {
-	imageDigest, err := c.pusher.Push(ctx, *f)
+func (c *Client) Push(ctx context.Context, path string) (err error) {
+	f, err := NewFunction(path)
+	if err != nil {
+		return
+	}
+
+	if !f.Built() {
+		return ErrNotBuilt
+	}
+
+	imageDigest, err := c.pusher.Push(ctx, f)
 	if err != nil {
 		return
 	}
diff --git a/cmd/build.go b/cmd/build.go
index c61a7a755c..82bce0e32b 100644
--- a/cmd/build.go
+++ b/cmd/build.go
@@ -29,30 +29,31 @@ func newBuildClient(cfg buildConfig) (*fn.Client, error) {
 	builder := buildpacks.NewBuilder()
 	listener := progress.New()
 
-	pusherOption := fn.WithPusher(nil)
+	var (
+		pusher *docker.Pusher
+		err    error
+	)
 	if cfg.Push {
 		credentialsProvider := creds.NewCredentialsProvider(
 			creds.WithPromptForCredentials(newPromptForCredentials()),
 			creds.WithPromptForCredentialStore(newPromptForCredentialStore()),
 			creds.WithTransport(cfg.Transport))
-		pusher, err := docker.NewPusher(
+		pusher, err = docker.NewPusher(
 			docker.WithCredentialsProvider(credentialsProvider),
 			docker.WithProgressListener(listener),
 			docker.WithTransport(cfg.Transport))
-
 		if err != nil {
 			return nil, err
 		}
 		pusher.Verbose = cfg.Verbose
-		pusherOption = fn.WithPusher(pusher)
 	}
+
 	builder.Verbose = cfg.Verbose
 	listener.Verbose = cfg.Verbose
 
 	return fn.New(
 		fn.WithBuilder(builder),
-		// fn.WithPusher(pusher),
-		pusherOption,
+		fn.WithPusher(pusher),
 		fn.WithProgressListener(listener),
 		fn.WithRegistry(cfg.Registry), // for image name when --image not provided
 		fn.WithVerbose(cfg.Verbose)), nil
@@ -201,7 +202,7 @@ func runBuild(cmd *cobra.Command, _ []string, clientFn buildClientFn) (err error
 
 	err = client.Build(cmd.Context(), config.Path)
 	if err == nil && config.Push {
-		err = client.Push(cmd.Context(), &function)
+		err = client.Push(cmd.Context(), config.Path)
 	}
 	return
 }
diff --git a/cmd/build_test.go b/cmd/build_test.go
index 56b733dc4e..7d918e635b 100644
--- a/cmd/build_test.go
+++ b/cmd/build_test.go
@@ -144,31 +144,57 @@ created: 2009-11-10 23:00:00`,
 
 func Test_newBuildClient(t *testing.T) {
 	tests := []struct {
-		name    string
-		cfg     buildConfig
-		succeed bool
+		name         string
+		cfg          buildConfig
+		fileContents string
+		succeed      bool
 	}{
 		{
 			name: "push flag set to false avoids pusher instanciation",
 			cfg: buildConfig{
 				Push: false,
 			},
+			fileContents: `name: test-func
+runtime: go
+image: registry.io/foo/bar:latest
+imageDigest: sha256:1111111111111111111111
+created: 2009-11-10 23:00:00`,
 			succeed: false,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+
+			tempDir, err := os.MkdirTemp("", "func-tests")
+			if err != nil {
+				t.Fatalf("temp dir couldn't be created %v", err)
+			}
+			t.Cleanup(func() {
+				os.RemoveAll(tempDir)
+			})
+
+			fullPath := tempDir + "/func.yaml"
+			tempFile, err := os.Create(fullPath)
+			if err != nil {
+				t.Fatalf("temp file couldn't be created %v", err)
+			}
+			_, err = tempFile.WriteString(tt.fileContents)
+			if err != nil {
+				t.Fatalf("file content was not written %v", err)
+			}
+
 			client, err := newBuildClient(tt.cfg)
 			if err != nil {
 				t.Error(err)
 			}
+
 			defer func() {
 				if r := recover(); r != nil && tt.succeed {
 					t.Errorf("expected function call to succeed %v, got actually %v", tt.succeed, r)
 				}
 			}()
-			err = client.Push(context.TODO(), &fn.Function{})
+			err = client.Push(context.TODO(), tempDir)
 			if err != nil {
 				t.Error(err)
 			}
diff --git a/cmd/deploy.go b/cmd/deploy.go
index 6fbb71c2d3..d5f16a346f 100644
--- a/cmd/deploy.go
+++ b/cmd/deploy.go
@@ -27,19 +27,25 @@ func init() {
 
 func newDeployClient(cfg deployConfig) (*fn.Client, error) {
 	listener := progress.New()
-
 	builder := buildpacks.NewBuilder()
 
-	credentialsProvider := creds.NewCredentialsProvider(
-		creds.WithPromptForCredentials(newPromptForCredentials()),
-		creds.WithPromptForCredentialStore(newPromptForCredentialStore()),
-		creds.WithTransport(cfg.Transport))
-	pusher, err := docker.NewPusher(
-		docker.WithCredentialsProvider(credentialsProvider),
-		docker.WithProgressListener(listener),
-		docker.WithTransport(cfg.Transport))
-	if err != nil {
-		return nil, err
+	var (
+		pusher *docker.Pusher
+		err    error
+	)
+	if cfg.Push {
+		credentialsProvider := creds.NewCredentialsProvider(
+			creds.WithPromptForCredentials(newPromptForCredentials()),
+			creds.WithPromptForCredentialStore(newPromptForCredentialStore()),
+			creds.WithTransport(cfg.Transport))
+		pusher, err = docker.NewPusher(
+			docker.WithCredentialsProvider(credentialsProvider),
+			docker.WithProgressListener(listener),
+			docker.WithTransport(cfg.Transport))
+		if err != nil {
+			return nil, err
+		}
+		pusher.Verbose = cfg.Verbose
 	}
 
 	deployer, err := knative.NewDeployer(cfg.Namespace)
@@ -49,7 +55,6 @@ func newDeployClient(cfg deployConfig) (*fn.Client, error) {
 
 	listener.Verbose = cfg.Verbose
 	builder.Verbose = cfg.Verbose
-	pusher.Verbose = cfg.Verbose
 	deployer.Verbose = cfg.Verbose
 
 	return fn.New(
@@ -90,7 +95,7 @@ kn func deploy --registry quay.io/myuser
 kn func deploy --image quay.io/myuser/myfunc -n myns
 `,
 		SuggestFor: []string{"delpoy", "deplyo"},
-		PreRunE:    bindEnv("image", "namespace", "path", "registry", "confirm", "build"),
+		PreRunE:    bindEnv("image", "namespace", "path", "registry", "confirm", "build", "push"),
 	}
 
 	cmd.Flags().BoolP("confirm", "c", false, "Prompt to confirm all configuration options (Env: $FUNC_CONFIRM)")
@@ -100,6 +105,7 @@ kn func deploy --image quay.io/myuser/myfunc -n myns
 	cmd.Flags().StringP("image", "i", "", "Full image name in the form [registry]/[namespace]/[name]:[tag] (optional). This option takes precedence over --registry (Env: $FUNC_IMAGE)")
 	cmd.Flags().StringP("registry", "r", "", "Registry + namespace part of the image to build, ex 'quay.io/myuser'.  The full image name is automatically determined based on the local directory name. If not provided the registry will be taken from func.yaml (Env: $FUNC_REGISTRY)")
 	cmd.Flags().BoolP("build", "b", true, "Build the image before deploying (Env: $FUNC_BUILD)")
+	cmd.Flags().BoolP("push", "u", true, "Attempt to push the function image to registry before deploying (Env: $FUNC_PUSH)")
 	setPathFlag(cmd)
 	setNamespaceFlag(cmd)
 
@@ -198,6 +204,12 @@ func runDeploy(cmd *cobra.Command, _ []string, clientFn deployClientFn) (err err
 		}
 	}
 
+	if config.Push {
+		if err := client.Push(cmd.Context(), config.Path); err != nil {
+			return err
+		}
+	}
+
 	return client.Deploy(cmd.Context(), config.Path)
 
 	// NOTE: Namespace is optional, default is that used by k8s client
@@ -289,6 +301,9 @@ type deployConfig struct {
 	// Build the associated Function before deploying.
 	Build bool
 
+	// Push function image to the registry before deploying.
+	Push bool
+
 	// Envs passed via cmd to be added/updated
 	EnvToUpdate *util.OrderedMap
 
@@ -313,6 +328,7 @@ func newDeployConfig(cmd *cobra.Command) (deployConfig, error) {
 		Verbose:     viper.GetBool("verbose"), // defined on root
 		Confirm:     viper.GetBool("confirm"),
 		Build:       viper.GetBool("build"),
+		Push:        viper.GetBool("push"),
 		EnvToUpdate: envToUpdate,
 		EnvToRemove: envToRemove,
 	}, nil