diff --git a/cmd/apply.go b/cmd/apply.go
index 42ed0e1..d28f773 100644
--- a/cmd/apply.go
+++ b/cmd/apply.go
@@ -1,31 +1,28 @@
 package cmd
 
 import (
-	"fmt"
 	"log"
 	"os"
 	"sync"
 
-	"github.com/liweiyi88/onedump/dumpjob"
+	"github.com/liweiyi88/onedump/dump"
 	"github.com/spf13/cobra"
 	"gopkg.in/yaml.v3"
 )
 
-var (
-	file string
-)
+var file string
 
 var applyCmd = &cobra.Command{
 	Use:   "apply -f /path/to/jobs.yaml",
 	Args:  cobra.ExactArgs(0),
-	Short: "Dump db content from different sources to diferent destinations",
+	Short: "Dump database content from different sources to diferent destinations with a yaml config file.",
 	Run: func(cmd *cobra.Command, args []string) {
 		content, err := os.ReadFile(file)
 		if err != nil {
 			log.Fatalf("failed to read job file from %s, error: %v", file, err)
 		}
 
-		var oneDump dumpjob.OneDump
+		var oneDump dump.Dump
 		err = yaml.Unmarshal(content, &oneDump)
 		if err != nil {
 			log.Fatalf("failed to read job content from %s, error: %v", file, err)
@@ -42,23 +39,19 @@ var applyCmd = &cobra.Command{
 			return
 		}
 
-		resultCh := make(chan *dumpjob.JobResult)
+		resultCh := make(chan *dump.JobResult)
 
 		for _, job := range oneDump.Jobs {
-			go func(job dumpjob.Job, resultCh chan *dumpjob.JobResult) {
+			go func(job *dump.Job, resultCh chan *dump.JobResult) {
 				resultCh <- job.Run()
 			}(job, resultCh)
 		}
 
 		var wg sync.WaitGroup
 		wg.Add(numberOfJobs)
-		go func(resultCh chan *dumpjob.JobResult) {
+		go func(resultCh chan *dump.JobResult) {
 			for result := range resultCh {
-				if result.Error != nil {
-					fmt.Printf("Job %s failed, it took %s with error: %v \n", result.JobName, result.Elapsed, result.Error)
-				} else {
-					fmt.Printf("Job %s succeeded and it took %v \n", result.JobName, result.Elapsed)
-				}
+				result.Print()
 				wg.Done()
 			}
 		}(resultCh)
@@ -70,6 +63,6 @@ var applyCmd = &cobra.Command{
 
 func init() {
 	rootCmd.AddCommand(applyCmd)
-	applyCmd.Flags().StringVarP(&file, "file", "f", "", "jobs yaml file path (required)")
+	applyCmd.Flags().StringVarP(&file, "file", "f", "", "jobs yaml file path.")
 	applyCmd.MarkFlagRequired("file")
 }
diff --git a/cmd/mysql.go b/cmd/mysql.go
deleted file mode 100644
index 1a001b1..0000000
--- a/cmd/mysql.go
+++ /dev/null
@@ -1,45 +0,0 @@
-package cmd
-
-import (
-	"log"
-	"strings"
-
-	"github.com/liweiyi88/onedump/dump"
-	"github.com/spf13/cobra"
-)
-
-var (
-	dsn       string
-	options   []string
-	mysqlGzip bool
-)
-
-var mysqlDumpCmd = &cobra.Command{
-	Use:   "mysql /path/to/dump-file.sql",
-	Args:  cobra.ExactArgs(1),
-	Short: "Dump mysql database to a file",
-	Run: func(cmd *cobra.Command, args []string) {
-		dumpFile := strings.TrimSpace(args[0])
-		if dumpFile == "" {
-			log.Fatal("you must specify the dump file path. e.g. /download/dump.sql")
-		}
-
-		dumper, err := dump.NewMysqlDumper(dsn, options, false)
-		if err != nil {
-			log.Fatal("failed to crete mysql dumper", err)
-		}
-
-		err = dumper.Dump(dumpFile, mysqlGzip)
-		if err != nil {
-			log.Fatal("failed to dump mysql datbase", err)
-		}
-	},
-}
-
-func init() {
-	rootCmd.AddCommand(mysqlDumpCmd)
-	mysqlDumpCmd.Flags().StringVarP(&dsn, "dsn", "d", "", "database dsn (required) ")
-	mysqlDumpCmd.MarkFlagRequired("dsn")
-	mysqlDumpCmd.Flags().StringArrayVarP(&options, "options", "o", nil, "use options to overwrite the default or add new mysqldump options e.g. --dump-options \"--no-create-info\" --dump-options \"--skip-comments\"")
-	mysqlDumpCmd.Flags().BoolVarP(&mysqlGzip, "gzip", "", true, "if need to gzip the file")
-}
diff --git a/cmd/root.go b/cmd/root.go
index bd18bf0..60a3fda 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -2,13 +2,53 @@ package cmd
 
 import (
 	"fmt"
+	"log"
 	"os"
+	"strings"
 
+	"github.com/liweiyi88/onedump/dump"
 	"github.com/spf13/cobra"
 )
 
+var (
+	sshHost, sshUser, sshPrivateKeyFile string
+	dbDsn                               string
+	jobName                             string
+	dumpOptions                         []string
+	gzip                                bool
+)
+
 var rootCmd = &cobra.Command{
-	Short: "onedump is a database dump, backup and load tool",
+	Use:   "<dbdriver> </path/to/dump-file.sql>",
+	Short: "Dump database content from a source to a destination via cli command.",
+	Args:  cobra.ExactArgs(2),
+	Run: func(cmd *cobra.Command, args []string) {
+		driver := strings.TrimSpace(args[0])
+
+		dumpFile := strings.TrimSpace(args[1])
+		if dumpFile == "" {
+			log.Fatal("you must specify the dump file path. e.g. /download/dump.sql")
+		}
+
+		name := "dump via cli"
+		if strings.TrimSpace(jobName) != "" {
+			name = jobName
+		}
+
+		job := dump.NewJob(
+			name,
+			driver,
+			dumpFile,
+			dbDsn,
+			dump.WithGzip(gzip),
+			dump.WithSshHost(sshHost),
+			dump.WithSshUser(sshUser),
+			dump.WithPrivateKeyFile(sshPrivateKeyFile),
+			dump.WithDumpOptions(dumpOptions),
+		)
+
+		job.Run().Print()
+	},
 }
 
 func Execute() {
@@ -17,3 +57,14 @@ func Execute() {
 		os.Exit(1)
 	}
 }
+
+func init() {
+	rootCmd.Flags().StringVarP(&sshHost, "ssh-host", "", "", "SSH host e.g. yourdomain.com (you can omit port as it uses 22 by default) or 56.09.139.09:33. (required) ")
+	rootCmd.Flags().StringVarP(&sshUser, "ssh-user", "", "root", "SSH username")
+	rootCmd.Flags().StringVarP(&sshPrivateKeyFile, "privatekey", "f", "", "private key file path for SSH connection")
+	rootCmd.Flags().StringArrayVarP(&dumpOptions, "dump-options", "", nil, "use options to overwrite or add new dump command options. e.g. for mysql: --dump-options \"--no-create-info\" --dump-options \"--skip-comments\"")
+	rootCmd.Flags().StringVarP(&dbDsn, "db-dsn", "d", "", "the database dsn for connection. e.g. <dbUser>:<dbPass>@tcp(<dbHost>:<dbPort>)/<dbName>")
+	rootCmd.MarkFlagRequired("db-dsn")
+	rootCmd.Flags().BoolVarP(&gzip, "gzip", "g", true, "if need to gzip the file")
+	rootCmd.Flags().StringVarP(&jobName, "job-name", "", "", "The dump job name")
+}
diff --git a/cmd/ssh.go b/cmd/ssh.go
deleted file mode 100644
index 3c55541..0000000
--- a/cmd/ssh.go
+++ /dev/null
@@ -1,53 +0,0 @@
-package cmd
-
-import (
-	"log"
-	"strings"
-
-	"github.com/liweiyi88/onedump/dump"
-	"github.com/spf13/cobra"
-)
-
-var (
-	sshHost, sshUser, sshPrivateKeyFile, databaseDsn string
-	dumpOptions                                      []string
-	gzip                                             bool
-)
-
-var sshDumpCmd = &cobra.Command{
-	Use:   "ssh mysql </path/to/dump-file.sql>",
-	Args:  cobra.ExactArgs(2),
-	Short: "Dump remote database to a file",
-	Run: func(cmd *cobra.Command, args []string) {
-		dumpFile := strings.TrimSpace(args[1])
-		if dumpFile == "" {
-			log.Fatal("you must specify the dump file path. e.g. /download/dump.sql")
-		}
-
-		dbDriver := strings.TrimSpace(args[0])
-
-		command, err := dump.GetSshDumpCommand(dbDriver, databaseDsn, dumpFile, dumpOptions)
-		if err != nil {
-			log.Fatal("failed to get database dump command", err)
-		}
-
-		sshDumper := dump.NewSshDumper(sshHost, sshUser, sshPrivateKeyFile)
-		err = sshDumper.Dump(dumpFile, command, gzip)
-
-		if err != nil {
-			log.Fatal("failed to run dump command via ssh", err)
-		}
-	},
-}
-
-func init() {
-	rootCmd.AddCommand(sshDumpCmd)
-	sshDumpCmd.Flags().StringVarP(&sshHost, "sshHost", "", "", "SSH host e.g. yourdomain.com (you can omit port as it uses 22 by default) or 56.09.139.09:33. (required) ")
-	sshDumpCmd.MarkFlagRequired("sshHost")
-	sshDumpCmd.Flags().StringVarP(&sshUser, "sshUser", "", "root", "SSH username")
-	sshDumpCmd.Flags().StringVarP(&sshPrivateKeyFile, "privateKeyFile", "f", "", "private key file path for SSH connection")
-	sshDumpCmd.Flags().StringArrayVarP(&dumpOptions, "dump-options", "", nil, "use options to overwrite or add new dump command options. e.g. for mysql: --dump-options \"--no-create-info\" --dump-options \"--skip-comments\"")
-	sshDumpCmd.Flags().StringVarP(&databaseDsn, "dbDsn", "", "", "the database dsn for connection. e.g. <dbUser>:<dbPass>@tcp(<dbHost>:<dbPort>)/<dbName>")
-	sshDumpCmd.MarkFlagRequired("dbDsn")
-	sshDumpCmd.Flags().BoolVarP(&gzip, "gzip", "g", true, "if need to gzip the file")
-}
diff --git a/dump/dbconfig.go b/driver/driver.go
similarity index 71%
rename from dump/dbconfig.go
rename to driver/driver.go
index 3ba9841..6eebcb3 100644
--- a/dump/dbconfig.go
+++ b/driver/driver.go
@@ -1,4 +1,9 @@
-package dump
+package driver
+
+type Driver interface {
+	GetDumpCommand() (string, []string, error)
+	GetSshDumpCommand() (string, error)
+}
 
 type DBConfig struct {
 	DBName   string
diff --git a/dump/mysqldumper.go b/driver/mysql.go
similarity index 77%
rename from dump/mysqldumper.go
rename to driver/mysql.go
index bc68e6c..ca43a85 100644
--- a/dump/mysqldumper.go
+++ b/driver/mysql.go
@@ -1,4 +1,4 @@
-package dump
+package driver
 
 import (
 	"fmt"
@@ -13,14 +13,14 @@ import (
 
 const CredentialFilePrefix = "mysqldumpcred-"
 
-type Mysql struct {
+type MysqlDriver struct {
 	MysqlDumpBinaryPath string
 	Options             []string
 	ViaSsh              bool
 	*DBConfig
 }
 
-func NewMysqlDumper(dsn string, options []string, viaSsh bool) (*Mysql, error) {
+func NewMysqlDriver(dsn string, options []string, viaSsh bool) (*MysqlDriver, error) {
 	config, err := mysql.ParseDSN(dsn)
 	if err != nil {
 		return nil, err
@@ -42,7 +42,7 @@ func NewMysqlDumper(dsn string, options []string, viaSsh bool) (*Mysql, error) {
 		commandOptions = options
 	}
 
-	return &Mysql{
+	return &MysqlDriver{
 		MysqlDumpBinaryPath: "mysqldump",
 		Options:             commandOptions,
 		ViaSsh:              viaSsh,
@@ -51,7 +51,7 @@ func NewMysqlDumper(dsn string, options []string, viaSsh bool) (*Mysql, error) {
 }
 
 // Get dump command used by ssh dumper.
-func (mysql *Mysql) GetSshDumpCommand() (string, error) {
+func (mysql *MysqlDriver) GetSshDumpCommand() (string, error) {
 	args, err := mysql.getDumpCommandArgs()
 	if err != nil {
 		return "", err
@@ -60,10 +60,26 @@ func (mysql *Mysql) GetSshDumpCommand() (string, error) {
 	return fmt.Sprintf("mysqldump %s", strings.Join(args, " ")), nil
 }
 
+func (mysql *MysqlDriver) GetDumpCommand() (string, []string, error) {
+	args, err := mysql.getDumpCommandArgs()
+
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to get dump command args %w", err)
+	}
+
+	// check and get the binary path.
+	mysqldumpBinaryPath, err := exec.LookPath(mysql.MysqlDumpBinaryPath)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to find mysqldump executable %s %w", mysql.MysqlDumpBinaryPath, err)
+	}
+
+	return mysqldumpBinaryPath, args, nil
+}
+
 // Store the username password in a temp file, and use it with the mysqldump command.
 // It avoids to expoes credentials when you run the mysqldump command as user can view the whole command via ps aux.
 // Inspired by https://github.com/spatie/db-dumper
-func (mysql *Mysql) getDumpCommandArgs() ([]string, error) {
+func (mysql *MysqlDriver) getDumpCommandArgs() ([]string, error) {
 
 	args := []string{}
 
@@ -83,7 +99,7 @@ func (mysql *Mysql) getDumpCommandArgs() ([]string, error) {
 	return args, nil
 }
 
-func (mysql *Mysql) createCredentialFile() (string, error) {
+func (mysql *MysqlDriver) createCredentialFile() (string, error) {
 	var fileName string
 
 	contents := `[client]
@@ -108,26 +124,3 @@ host = %s`
 
 	return file.Name(), nil
 }
-
-func (mysql *Mysql) Dump(dumpFile string, shouldGzip bool) error {
-	args, err := mysql.getDumpCommandArgs()
-
-	if err != nil {
-		return fmt.Errorf("failed to get dump command args %w", err)
-	}
-
-	// check and get the binary path.
-	mysqldumpBinaryPath, err := exec.LookPath(mysql.MysqlDumpBinaryPath)
-	if err != nil {
-		return fmt.Errorf("failed to find mysqldump executable %s %w", mysql.MysqlDumpBinaryPath, err)
-	}
-
-	cmd := exec.Command(mysqldumpBinaryPath, args...)
-
-	dump(cmd, dumpFile, shouldGzip, "")
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
diff --git a/dump/mysqldumper_test.go b/driver/mysql_test.go
similarity index 72%
rename from dump/mysqldumper_test.go
rename to driver/mysql_test.go
index 392bfb2..161a944 100644
--- a/dump/mysqldumper_test.go
+++ b/driver/mysql_test.go
@@ -1,4 +1,4 @@
-package dump
+package driver
 
 import (
 	"os"
@@ -10,7 +10,7 @@ import (
 var testDBDsn = "admin:my_password@tcp(127.0.0.1:3306)/dump_test"
 
 func TestDefaultGetDumpCommand(t *testing.T) {
-	mysql, err := NewMysqlDumper(testDBDsn, nil, false)
+	mysql, err := NewMysqlDriver(testDBDsn, nil, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -34,7 +34,7 @@ func TestDefaultGetDumpCommand(t *testing.T) {
 }
 
 func TestGetDumpCommandWithOptions(t *testing.T) {
-	mysql, err := NewMysqlDumper(testDBDsn, []string{"--skip-comments", "--extended-insert", "--no-create-info", "--default-character-set=utf-8", "--single-transaction", "--skip-lock-tables", "--quick", "--set-gtid-purged=ON"}, false)
+	mysql, err := NewMysqlDriver(testDBDsn, []string{"--skip-comments", "--extended-insert", "--no-create-info", "--default-character-set=utf-8", "--single-transaction", "--skip-lock-tables", "--quick", "--set-gtid-purged=ON"}, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -78,7 +78,7 @@ func TestGetDumpCommandWithOptions(t *testing.T) {
 }
 
 func TestCreateCredentialFile(t *testing.T) {
-	mysql, err := NewMysqlDumper(testDBDsn, nil, false)
+	mysql, err := NewMysqlDriver(testDBDsn, nil, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -113,37 +113,37 @@ host = 127.0.0.1`
 	t.Log("removed temp credential file", fileName)
 }
 
-func TestDump(t *testing.T) {
-	dsn := "root@tcp(127.0.0.1:3306)/test_local"
-	mysql, err := NewMysqlDumper(dsn, nil, false)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	dumpfile, err := os.CreateTemp("", "dbdump")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer dumpfile.Close()
-
-	err = mysql.Dump(dumpfile.Name(), false)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	out, err := os.ReadFile(dumpfile.Name())
-	if err != nil {
-		t.Fatal("failed to read the test dump file")
-	}
-
-	if len(out) == 0 {
-		t.Fatal("test dump file is empty")
-	}
-
-	t.Log("test dump file content size", len(out))
-
-	err = os.Remove(dumpfile.Name())
-	if err != nil {
-		t.Fatal("can not cleanup the test dump file", err)
-	}
-}
+// func TestDump(t *testing.T) {
+// 	dsn := "root@tcp(127.0.0.1:3306)/test_local"
+// 	mysql, err := NewMysqlDumper(dsn, nil, false)
+// 	if err != nil {
+// 		t.Fatal(err)
+// 	}
+
+// 	dumpfile, err := os.CreateTemp("", "dbdump")
+// 	if err != nil {
+// 		t.Fatal(err)
+// 	}
+// 	defer dumpfile.Close()
+
+// 	err = mysql.Dump(dumpfile.Name(), false)
+// 	if err != nil {
+// 		t.Fatal(err)
+// 	}
+
+// 	out, err := os.ReadFile(dumpfile.Name())
+// 	if err != nil {
+// 		t.Fatal("failed to read the test dump file")
+// 	}
+
+// 	if len(out) == 0 {
+// 		t.Fatal("test dump file is empty")
+// 	}
+
+// 	t.Log("test dump file content size", len(out))
+
+// 	err = os.Remove(dumpfile.Name())
+// 	if err != nil {
+// 		t.Fatal("can not cleanup the test dump file", err)
+// 	}
+// }
diff --git a/dump/dump.go b/dump/dump.go
index 75660c8..712c255 100644
--- a/dump/dump.go
+++ b/dump/dump.go
@@ -5,24 +5,250 @@ import (
 	"compress/gzip"
 	"errors"
 	"fmt"
-	"log"
+	"net"
 	"os"
 	"os/exec"
 	"strings"
 	"time"
 
+	"github.com/liweiyi88/onedump/driver"
 	"github.com/liweiyi88/onedump/storage"
 	"golang.org/x/crypto/ssh"
 )
 
-func dumpToFile(runner any, dumpFile string, shouldGzip bool, command string, store storage.Storage) error {
+type Dump struct {
+	Jobs []*Job `yaml:"jobs"`
+}
+
+func (dump *Dump) Validate() error {
+	errorCollection := make([]string, 0)
+
+	for _, job := range dump.Jobs {
+		err := job.validate()
+		if err != nil {
+			errorCollection = append(errorCollection, err.Error())
+		}
+	}
+
+	if len(errorCollection) == 0 {
+		return nil
+	}
+
+	return errors.New(strings.Join(errorCollection, ","))
+}
+
+type JobResult struct {
+	Error   error
+	JobName string
+	Elapsed time.Duration
+}
+
+func (result *JobResult) Print() {
+	if result.Error != nil {
+		fmt.Printf("Job: %s failed, it took %s with error: %v \n", result.JobName, result.Elapsed, result.Error)
+	} else {
+		fmt.Printf("Job: %s succeeded, it took %v \n", result.JobName, result.Elapsed)
+	}
+}
+
+type Job struct {
+	DumpFile       string                  `yaml:"dumpfile"`
+	Name           string                  `yaml:"name"`
+	DBDriver       string                  `yaml:"dbdriver"`
+	DBDsn          string                  `yaml:"dbdsn"`
+	Gzip           bool                    `yaml:"gzip"`
+	SshHost        string                  `yaml:"sshhost"`
+	SshUser        string                  `yaml:"sshuser"`
+	PrivateKeyFile string                  `yaml:"privatekeyfile"`
+	DumpOptions    []string                `yaml:"options"`
+	S3             *storage.AWSCredentials `yaml:"s3"`
+}
+
+type Option func(job *Job)
+
+func WithSshHost(sshHost string) Option {
+	return func(job *Job) {
+		job.SshHost = sshHost
+	}
+}
+
+func WithSshUser(sshUser string) Option {
+	return func(job *Job) {
+		job.SshUser = sshUser
+	}
+}
+
+func WithGzip(gzip bool) Option {
+	return func(job *Job) {
+		job.Gzip = gzip
+	}
+}
+
+func WithDumpOptions(dumpOptions []string) Option {
+	return func(job *Job) {
+		job.DumpOptions = dumpOptions
+	}
+}
+
+func WithPrivateKeyFile(privateKeyFile string) Option {
+	return func(job *Job) {
+		job.PrivateKeyFile = privateKeyFile
+	}
+}
+
+func NewJob(name, driver, dumpFile, dbDsn string, opts ...Option) *Job {
+	job := &Job{
+		Name:     name,
+		DBDriver: driver,
+		DumpFile: dumpFile,
+		DBDsn:    dbDsn,
+	}
+
+	for _, opt := range opts {
+		opt(job)
+	}
+
+	return job
+}
+
+func (job Job) validate() error {
+	if strings.TrimSpace(job.Name) == "" {
+		return errors.New("job name is required")
+	}
+
+	if strings.TrimSpace(job.DumpFile) == "" {
+		return errors.New("dump file path is required")
+	}
+
+	if strings.TrimSpace(job.DBDsn) == "" {
+		return errors.New("databse dsn is required")
+	}
+
+	if strings.TrimSpace(job.DBDriver) == "" {
+		return errors.New("databse driver is required")
+	}
+
+	return nil
+}
+
+func (job *Job) viaSsh() bool {
+	if strings.TrimSpace(job.SshHost) != "" && strings.TrimSpace(job.SshUser) != "" && strings.TrimSpace(job.PrivateKeyFile) != "" {
+		return true
+	}
+
+	return false
+}
+
+func (job *Job) getDBDriver() (driver.Driver, error) {
+	switch job.DBDriver {
+	case "mysql":
+		driver, err := driver.NewMysqlDriver(job.DBDsn, job.DumpOptions, job.viaSsh())
+		if err != nil {
+			return nil, err
+		}
+
+		return driver, nil
+	default:
+		return nil, fmt.Errorf("%s is not a supported database driver", job.DBDriver)
+	}
+}
+
+func ensureHaveSSHPort(addr string) string {
+	if _, _, err := net.SplitHostPort(addr); err != nil {
+		return net.JoinHostPort(addr, "22")
+	}
+	return addr
+}
+
+func (job *Job) sshDump() error {
+	host := ensureHaveSSHPort(job.SshHost)
+
+	pKey, err := os.ReadFile(job.PrivateKeyFile)
+	if err != nil {
+		return fmt.Errorf("can not read the private key file :%w", err)
+	}
+
+	signer, err := ssh.ParsePrivateKey(pKey)
+	if err != nil {
+		return fmt.Errorf("failed to create singer :%w", err)
+	}
+
+	conf := &ssh.ClientConfig{
+		User:            job.SshUser,
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		Auth: []ssh.AuthMethod{
+			ssh.PublicKeys(signer),
+		},
+	}
+
+	client, err := ssh.Dial("tcp", host, conf)
+	if err != nil {
+		return fmt.Errorf("failed to dial remote server via ssh: %w", err)
+	}
+
+	defer client.Close()
+
+	session, err := client.NewSession()
+	if err != nil {
+		return fmt.Errorf("failed to start ssh session: %w", err)
+	}
+
+	defer session.Close()
+
+	err = job.dump(session)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (job *Job) execDump() error {
+	err := job.dump(nil)
+	if err != nil {
+		return fmt.Errorf("failed to exec command dump: %w", err)
+	}
+
+	return nil
+}
+
+func (job *Job) Run() *JobResult {
+	start := time.Now()
+	var result JobResult
+
+	defer func() {
+		elapsed := time.Since(start)
+		result.Elapsed = elapsed
+	}()
+
+	result.JobName = job.Name
+
+	if job.viaSsh() {
+		err := job.sshDump()
+		if err != nil {
+			result.Error = fmt.Errorf("job %s, failed to run ssh dump command: %v", job.Name, err)
+		}
+
+		return &result
+	}
+
+	err := job.execDump()
+	if err != nil {
+		result.Error = fmt.Errorf("job %s, failed to run dump command: %v", job.Name, err)
+
+	}
+
+	return &result
+}
+
+func (job *Job) dumpToFile(sshSession *ssh.Session, store storage.Storage) error {
 	file, err := store.CreateDumpFile()
 	if err != nil {
 		return fmt.Errorf("failed to create storage dump file: %w", err)
 	}
 
 	var gzipWriter *gzip.Writer
-	if shouldGzip {
+	if job.Gzip {
 		gzipWriter = gzip.NewWriter(file)
 	}
 
@@ -34,32 +260,48 @@ func dumpToFile(runner any, dumpFile string, shouldGzip bool, command string, st
 		file.Close()
 	}()
 
-	switch runner := runner.(type) {
-	case *exec.Cmd:
-		runner.Stderr = os.Stderr
-		if gzipWriter != nil {
-			runner.Stdout = gzipWriter
-		} else {
-			runner.Stdout = file
-		}
+	driver, err := job.getDBDriver()
+	if err != nil {
+		return fmt.Errorf("failed to get db driver: %w", err)
+	}
 
-		if err := runner.Run(); err != nil {
-			return fmt.Errorf("remote command error: %v", err)
-		}
-	case *ssh.Session:
+	if sshSession != nil {
 		var remoteErr bytes.Buffer
-		runner.Stderr = &remoteErr
+		sshSession.Stderr = &remoteErr
 		if gzipWriter != nil {
-			runner.Stdout = gzipWriter
+			sshSession.Stdout = gzipWriter
 		} else {
-			runner.Stdout = file
+			sshSession.Stdout = file
+		}
+
+		sshCommand, err := driver.GetSshDumpCommand()
+		if err != nil {
+			return fmt.Errorf("failed to get ssh dump command %w", err)
 		}
 
-		if err := runner.Run(command); err != nil {
+		if err := sshSession.Run(sshCommand); err != nil {
 			return fmt.Errorf("remote command error: %s, %v", remoteErr.String(), err)
 		}
-	default:
-		return errors.New("unsupport runner type")
+
+		return nil
+	}
+
+	command, args, err := driver.GetDumpCommand()
+	if err != nil {
+		return fmt.Errorf("job %s failed to get dump command: %v", job.Name, err)
+	}
+
+	cmd := exec.Command(command, args...)
+
+	cmd.Stderr = os.Stderr
+	if gzipWriter != nil {
+		cmd.Stdout = gzipWriter
+	} else {
+		cmd.Stdout = file
+	}
+
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("remote command error: %v", err)
 	}
 
 	return nil
@@ -69,14 +311,13 @@ func dumpToFile(runner any, dumpFile string, shouldGzip bool, command string, st
 // It checks the filename to determine if we need to upload the file to remote storage or keep it locally.
 // For uploading file to S3 bucket, the filename shold follow the pattern: s3://<bucket_name>/<key> .
 // For any remote upload, we try to cache it in a local dir then upload it to the remote storage.
-func dump(runner any, dumpFile string, shouldGzip bool, command string) error {
-	dumpFilename := ensureFileSuffix(dumpFile, shouldGzip)
-	store, err := storage.CreateStorage(dumpFilename)
+func (job *Job) dump(sshSession *ssh.Session) error {
+	store, err := job.createStorage()
 	if err != nil {
 		return fmt.Errorf("failed to create storage: %w", err)
 	}
 
-	err = dumpToFile(runner, dumpFile, shouldGzip, command, store)
+	err = job.dumpToFile(sshSession, store)
 	if err != nil {
 		return err
 	}
@@ -93,6 +334,24 @@ func dump(runner any, dumpFile string, shouldGzip bool, command string) error {
 	return nil
 }
 
+// Factory method to create the storage struct based on filename.
+func (job *Job) createStorage() (storage.Storage, error) {
+	filename := ensureFileSuffix(job.DumpFile, job.Gzip)
+	s3Storage, ok, err := storage.CreateS3Storage(filename, job.S3)
+
+	if err != nil {
+		return nil, err
+	}
+
+	if ok {
+		return s3Storage, nil
+	}
+
+	return &storage.LocalStorage{
+		Filename: filename,
+	}, nil
+}
+
 // Ensure a file has proper file extension.
 func ensureFileSuffix(filename string, shouldGzip bool) string {
 	if !shouldGzip {
@@ -105,13 +364,3 @@ func ensureFileSuffix(filename string, shouldGzip bool) string {
 
 	return filename + ".gz"
 }
-
-// Performanece debug function.
-func trace(name string) func() {
-	start := time.Now()
-
-	return func() {
-		elapsed := time.Since(start)
-		log.Printf("%s took %s", name, elapsed)
-	}
-}
diff --git a/dump/dump_test.go b/dump/dump_test.go
new file mode 100644
index 0000000..5b360eb
--- /dev/null
+++ b/dump/dump_test.go
@@ -0,0 +1,165 @@
+package dump
+
+import (
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"encoding/pem"
+	"fmt"
+	"os"
+	"testing"
+)
+
+func generateTestRSAPrivatePEMFile() (string, error) {
+	tempDir := os.TempDir()
+
+	key, err := rsa.GenerateKey(rand.Reader, 4096)
+	if err != nil {
+		return "", fmt.Errorf("could not genereate rsa key pair %w", err)
+	}
+
+	keyPEM := pem.EncodeToMemory(
+		&pem.Block{
+			Type:  "RSA PRIVATE KEY",
+			Bytes: x509.MarshalPKCS1PrivateKey(key),
+		},
+	)
+
+	privatePEMFile := fmt.Sprintf("%s/%s", tempDir, "sshdump_test.rsa")
+
+	if err := os.WriteFile(privatePEMFile, keyPEM, 0700); err != nil {
+		return "", fmt.Errorf("failed to write private key to file %w", err)
+	}
+
+	return privatePEMFile, nil
+}
+
+func TestEnsureSSHHostHavePort(t *testing.T) {
+	sshHost := "127.0.0.1"
+
+	if ensureHaveSSHPort(sshHost) != sshHost+":22" {
+		t.Error("ssh host port is not ensured")
+	}
+}
+
+// func TestNewSshDumper(t *testing.T) {
+// 	sshDumper := NewSshDumper("127.0.0.1", "root", "~/.ssh/test.pem")
+
+// 	if sshDumper.Host != "127.0.0.1" {
+// 		t.Errorf("ssh host is unexpected, exepct: %s, actual: %s", "127.0.0.1", sshDumper.Host)
+// 	}
+
+// 	if sshDumper.User != "root" {
+// 		t.Errorf("ssh user is unexpected, exepct: %s, actual: %s", "root", sshDumper.User)
+// 	}
+
+// 	if sshDumper.PrivateKeyFile != "~/.ssh/test.pem" {
+// 		t.Errorf("ssh private key file path is unexpected, exepct: %s, actual: %s", "~/.ssh/test.pem", sshDumper.PrivateKeyFile)
+// 	}
+// }
+
+// func TestSSHDump(t *testing.T) {
+// 	privateKeyFile, err := generateTestRSAPrivatePEMFile()
+
+// 	dumpFile := os.TempDir() + "sshdump.sql.gz"
+
+// 	if err != nil {
+// 		t.Error("failed to generate test rsa key pairs", err)
+// 	}
+
+// 	defer func() {
+// 		err := os.Remove(privateKeyFile)
+// 		if err != nil {
+// 			t.Logf("failed to remove private key file %s", privateKeyFile)
+// 		}
+// 	}()
+
+// 	go func() {
+// 		sshDumper := NewSshDumper("127.0.0.1:2022", "root", privateKeyFile)
+// 		err := sshDumper.Dump(dumpFile, "echo hello", true)
+// 		if err != nil {
+// 			t.Error("failed to dump file", err)
+// 		}
+// 	}()
+
+// 	// An SSH server is represented by a ServerConfig, which holds
+// 	// certificate details and handles authentication of ServerConns.
+// 	config := &ssh.ServerConfig{
+// 		PublicKeyCallback: func(c ssh.ConnMetadata, pubKey ssh.PublicKey) (*ssh.Permissions, error) {
+// 			return &ssh.Permissions{
+// 				// Record the public key used for authentication.
+// 				Extensions: map[string]string{
+// 					"pubkey-fp": ssh.FingerprintSHA256(pubKey),
+// 				},
+// 			}, nil
+// 		},
+// 	}
+
+// 	privateBytes, err := os.ReadFile(privateKeyFile)
+// 	if err != nil {
+// 		t.Fatal("Failed to load private key: ", err)
+// 	}
+
+// 	private, err := ssh.ParsePrivateKey(privateBytes)
+// 	if err != nil {
+// 		t.Fatal("Failed to parse private key: ", err)
+// 	}
+
+// 	config.AddHostKey(private)
+
+// 	// Once a ServerConfig has been configured, connections can be
+// 	// accepted.
+// 	listener, err := net.Listen("tcp", "0.0.0.0:2022")
+// 	if err != nil {
+// 		t.Fatal("failed to listen for connection: ", err)
+// 	}
+
+// 	nConn, err := listener.Accept()
+// 	if err != nil {
+// 		t.Fatal("failed to accept incoming connection: ", err)
+// 	}
+
+// 	// Before use, a handshake must be performed on the incoming
+// 	// net.Conn.
+// 	conn, chans, reqs, err := ssh.NewServerConn(nConn, config)
+// 	if err != nil {
+// 		log.Fatal("failed to handshake: ", err)
+// 	}
+// 	t.Logf("logged in with key %s", conn.Permissions.Extensions["pubkey-fp"])
+
+// 	// The incoming Request channel must be serviced.
+// 	go ssh.DiscardRequests(reqs)
+
+// 	// Service the incoming Channel channel.
+// 	newChannel := <-chans
+// 	// Channels have a type, depending on the application level
+// 	// protocol intended. In the case of a shell, the type is
+// 	// "session" and ServerShell may be used to present a simple
+// 	// terminal interface.
+// 	if newChannel.ChannelType() != "session" {
+// 		newChannel.Reject(ssh.UnknownChannelType, "unknown channel type")
+// 		t.Fatal("unknown channel type")
+// 	}
+
+// 	channel, requests, err := newChannel.Accept()
+
+// 	if err != nil {
+// 		log.Fatalf("Could not accept channel: %v", err)
+// 	}
+
+// 	req := <-requests
+// 	req.Reply(true, []byte("ssh dump"))
+// 	channel.SendRequest("exit-status", false, []byte{0, 0, 0, 0})
+
+// 	channel.Close()
+// 	conn.Close()
+
+// 	if _, err := os.Stat(dumpFile); errors.Is(err, os.ErrNotExist) {
+// 		t.Error("dump file does not existed")
+// 	} else {
+// 		err := os.Remove(dumpFile)
+// 		if err != nil {
+// 			t.Fatal("failed to remove the test dump file", err)
+// 		}
+// 	}
+// }
diff --git a/dump/sshdumper.go b/dump/sshdumper.go
deleted file mode 100644
index 4185092..0000000
--- a/dump/sshdumper.go
+++ /dev/null
@@ -1,93 +0,0 @@
-package dump
-
-import (
-	"fmt"
-	"net"
-	"os"
-
-	"golang.org/x/crypto/ssh"
-)
-
-type SshDumper struct {
-	User, Host, PrivateKeyFile string
-}
-
-func NewSshDumper(host, user, privateKeyFile string) *SshDumper {
-	return &SshDumper{
-		Host:           host,
-		User:           user,
-		PrivateKeyFile: privateKeyFile,
-	}
-}
-
-func (sshDumper *SshDumper) Dump(dumpFile, command string, shouldGzip bool) error {
-	defer trace("ssh dump")()
-
-	host := ensureHavePort(sshDumper.Host)
-
-	pKey, err := os.ReadFile(sshDumper.PrivateKeyFile)
-	if err != nil {
-		return fmt.Errorf("can not read the private key file :%w", err)
-	}
-
-	signer, err := ssh.ParsePrivateKey(pKey)
-	if err != nil {
-		return fmt.Errorf("failed to create singer :%w", err)
-	}
-
-	conf := &ssh.ClientConfig{
-		User:            sshDumper.User,
-		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
-		Auth: []ssh.AuthMethod{
-			ssh.PublicKeys(signer),
-		},
-	}
-
-	client, err := ssh.Dial("tcp", host, conf)
-	if err != nil {
-		return fmt.Errorf("failed to dial remote server via ssh: %w", err)
-	}
-
-	defer client.Close()
-
-	session, err := client.NewSession()
-	if err != nil {
-		return fmt.Errorf("failed to start ssh session: %w", err)
-	}
-
-	defer session.Close()
-
-	err = dump(session, dumpFile, shouldGzip, command)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func GetSshDumpCommand(dbDriver, dsn, dumpFile string, dumpOptions []string) (string, error) {
-	switch dbDriver {
-	case "mysql":
-		mysqlDumper, err := NewMysqlDumper(dsn, dumpOptions, true)
-		if err != nil {
-			return "", err
-		}
-
-		command, err := mysqlDumper.GetSshDumpCommand()
-
-		if err != nil {
-			return "", err
-		}
-
-		return command, nil
-	default:
-		return "", fmt.Errorf("%s is not a supported database driver", dbDriver)
-	}
-}
-
-func ensureHavePort(addr string) string {
-	if _, _, err := net.SplitHostPort(addr); err != nil {
-		return net.JoinHostPort(addr, "22")
-	}
-	return addr
-}
diff --git a/dump/sshdumper_test.go b/dump/sshdumper_test.go
deleted file mode 100644
index 868a74f..0000000
--- a/dump/sshdumper_test.go
+++ /dev/null
@@ -1,170 +0,0 @@
-package dump
-
-import (
-	"crypto/rand"
-	"crypto/rsa"
-	"crypto/x509"
-	"encoding/pem"
-	"errors"
-	"fmt"
-	"log"
-	"net"
-	"os"
-	"testing"
-
-	"golang.org/x/crypto/ssh"
-)
-
-func generateTestRSAPrivatePEMFile() (string, error) {
-	tempDir := os.TempDir()
-
-	key, err := rsa.GenerateKey(rand.Reader, 4096)
-	if err != nil {
-		return "", fmt.Errorf("could not genereate rsa key pair %w", err)
-	}
-
-	keyPEM := pem.EncodeToMemory(
-		&pem.Block{
-			Type:  "RSA PRIVATE KEY",
-			Bytes: x509.MarshalPKCS1PrivateKey(key),
-		},
-	)
-
-	privatePEMFile := fmt.Sprintf("%s/%s", tempDir, "sshdump_test.rsa")
-
-	if err := os.WriteFile(privatePEMFile, keyPEM, 0700); err != nil {
-		return "", fmt.Errorf("failed to write private key to file %w", err)
-	}
-
-	return privatePEMFile, nil
-}
-
-func TestEnsureSSHHostHavePort(t *testing.T) {
-	sshHost := "127.0.0.1"
-
-	if ensureHavePort(sshHost) != sshHost+":22" {
-		t.Error("ssh host port is not ensured")
-	}
-}
-
-func TestNewSshDumper(t *testing.T) {
-	sshDumper := NewSshDumper("127.0.0.1", "root", "~/.ssh/test.pem")
-
-	if sshDumper.Host != "127.0.0.1" {
-		t.Errorf("ssh host is unexpected, exepct: %s, actual: %s", "127.0.0.1", sshDumper.Host)
-	}
-
-	if sshDumper.User != "root" {
-		t.Errorf("ssh user is unexpected, exepct: %s, actual: %s", "root", sshDumper.User)
-	}
-
-	if sshDumper.PrivateKeyFile != "~/.ssh/test.pem" {
-		t.Errorf("ssh private key file path is unexpected, exepct: %s, actual: %s", "~/.ssh/test.pem", sshDumper.PrivateKeyFile)
-	}
-}
-
-func TestSSHDump(t *testing.T) {
-	privateKeyFile, err := generateTestRSAPrivatePEMFile()
-
-	dumpFile := os.TempDir() + "sshdump.sql.gz"
-
-	if err != nil {
-		t.Error("failed to generate test rsa key pairs", err)
-	}
-
-	defer func() {
-		err := os.Remove(privateKeyFile)
-		if err != nil {
-			t.Logf("failed to remove private key file %s", privateKeyFile)
-		}
-	}()
-
-	go func() {
-		sshDumper := NewSshDumper("127.0.0.1:2022", "root", privateKeyFile)
-		err := sshDumper.Dump(dumpFile, "echo hello", true)
-		if err != nil {
-			t.Error("failed to dump file", err)
-		}
-	}()
-
-	// An SSH server is represented by a ServerConfig, which holds
-	// certificate details and handles authentication of ServerConns.
-	config := &ssh.ServerConfig{
-		PublicKeyCallback: func(c ssh.ConnMetadata, pubKey ssh.PublicKey) (*ssh.Permissions, error) {
-			return &ssh.Permissions{
-				// Record the public key used for authentication.
-				Extensions: map[string]string{
-					"pubkey-fp": ssh.FingerprintSHA256(pubKey),
-				},
-			}, nil
-		},
-	}
-
-	privateBytes, err := os.ReadFile(privateKeyFile)
-	if err != nil {
-		t.Fatal("Failed to load private key: ", err)
-	}
-
-	private, err := ssh.ParsePrivateKey(privateBytes)
-	if err != nil {
-		t.Fatal("Failed to parse private key: ", err)
-	}
-
-	config.AddHostKey(private)
-
-	// Once a ServerConfig has been configured, connections can be
-	// accepted.
-	listener, err := net.Listen("tcp", "0.0.0.0:2022")
-	if err != nil {
-		t.Fatal("failed to listen for connection: ", err)
-	}
-
-	nConn, err := listener.Accept()
-	if err != nil {
-		t.Fatal("failed to accept incoming connection: ", err)
-	}
-
-	// Before use, a handshake must be performed on the incoming
-	// net.Conn.
-	conn, chans, reqs, err := ssh.NewServerConn(nConn, config)
-	if err != nil {
-		log.Fatal("failed to handshake: ", err)
-	}
-	t.Logf("logged in with key %s", conn.Permissions.Extensions["pubkey-fp"])
-
-	// The incoming Request channel must be serviced.
-	go ssh.DiscardRequests(reqs)
-
-	// Service the incoming Channel channel.
-	newChannel := <-chans
-	// Channels have a type, depending on the application level
-	// protocol intended. In the case of a shell, the type is
-	// "session" and ServerShell may be used to present a simple
-	// terminal interface.
-	if newChannel.ChannelType() != "session" {
-		newChannel.Reject(ssh.UnknownChannelType, "unknown channel type")
-		t.Fatal("unknown channel type")
-	}
-
-	channel, requests, err := newChannel.Accept()
-
-	if err != nil {
-		log.Fatalf("Could not accept channel: %v", err)
-	}
-
-	req := <-requests
-	req.Reply(true, []byte("ssh dump"))
-	channel.SendRequest("exit-status", false, []byte{0, 0, 0, 0})
-
-	channel.Close()
-	conn.Close()
-
-	if _, err := os.Stat(dumpFile); errors.Is(err, os.ErrNotExist) {
-		t.Error("dump file does not existed")
-	} else {
-		err := os.Remove(dumpFile)
-		if err != nil {
-			t.Fatal("failed to remove the test dump file", err)
-		}
-	}
-}
diff --git a/dumpjob/dumpjob.go b/dumpjob/dumpjob.go
deleted file mode 100644
index 4e7e659..0000000
--- a/dumpjob/dumpjob.go
+++ /dev/null
@@ -1,117 +0,0 @@
-package dumpjob
-
-import (
-	"errors"
-	"fmt"
-	"strings"
-	"time"
-
-	"github.com/liweiyi88/onedump/dump"
-)
-
-type OneDump struct {
-	Jobs []Job `yaml:"jobs"`
-}
-
-func (oneDump *OneDump) Validate() error {
-	errorCollection := make([]string, 0)
-
-	for _, job := range oneDump.Jobs {
-		err := job.validate()
-		if err != nil {
-			errorCollection = append(errorCollection, err.Error())
-		}
-	}
-
-	if len(errorCollection) == 0 {
-		return nil
-	}
-
-	return errors.New(strings.Join(errorCollection, ","))
-}
-
-type JobResult struct {
-	Error   error
-	JobName string
-	Elapsed time.Duration
-}
-
-type Job struct {
-	DumpFile       string   `yaml:"dumpfile"`
-	Name           string   `yaml:"name"`
-	DBDriver       string   `yaml:"dbdriver"`
-	DBDsn          string   `yaml:"dbdsn"`
-	Gzip           bool     `yaml:"gzip"`
-	SshHost        string   `yaml:"sshhost"`
-	SshUser        string   `yaml:"sshuser"`
-	PrivateKeyFile string   `yaml:"privatekeyfile"`
-	Options        []string `yaml:"options"`
-}
-
-func (job Job) validate() error {
-	if job.Name == "" {
-		return errors.New("job name is required")
-	}
-
-	if job.DumpFile == "" {
-		return errors.New("dump file path is required")
-	}
-
-	if job.DBDsn == "" {
-		return errors.New("databse dsn is required")
-	}
-
-	if job.DBDriver == "" {
-		return errors.New("databse driver is required")
-	}
-
-	return nil
-}
-
-func (job Job) viaSsh() bool {
-	if job.SshHost != "" && job.SshUser != "" && job.PrivateKeyFile != "" {
-		return true
-	}
-
-	return false
-}
-
-func (job Job) Run() *JobResult {
-	start := time.Now()
-	var result JobResult
-
-	defer func() {
-		elapsed := time.Since(start)
-		result.Elapsed = elapsed
-	}()
-
-	result.JobName = job.Name
-
-	if job.viaSsh() {
-		command, err := dump.GetSshDumpCommand(job.DBDriver, job.DBDsn, job.DumpFile, job.Options)
-		if err != nil {
-			result.Error = fmt.Errorf("job %s, failed to get dump command: %v", job.Name, err)
-			return &result
-		}
-
-		sshDumper := dump.NewSshDumper(job.SshHost, job.SshUser, job.PrivateKeyFile)
-		err = sshDumper.Dump(job.DumpFile, command, job.Gzip)
-		if err != nil {
-			result.Error = fmt.Errorf("job %s, failed to run dump command: %v", job.Name, err)
-		}
-	} else {
-		dumper, err := dump.NewMysqlDumper(job.DBDsn, job.Options, false)
-		if err != nil {
-			result.Error = fmt.Errorf("job %s, failed to crete  mysql dumper: %v", job.Name, err)
-			return &result
-		}
-
-		err = dumper.Dump(job.DumpFile, job.Gzip)
-		if err != nil {
-			result.Error = fmt.Errorf("job %s, failed to dump mysql dumper: %v", job.Name, err)
-			return &result
-		}
-	}
-
-	return &result
-}
diff --git a/storage/s3.go b/storage/s3.go
index 6daaa00..32034ca 100644
--- a/storage/s3.go
+++ b/storage/s3.go
@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 )
@@ -15,7 +16,7 @@ const s3Prefix = "s3://"
 
 var ErrInvalidS3Path = fmt.Errorf("invalid s3 filename, it should follow the format %s<bucket>/<path|filename>", s3Prefix)
 
-func createS3Storage(filename string) (*S3Storage, bool, error) {
+func CreateS3Storage(filename string, credentials *AWSCredentials) (*S3Storage, bool, error) {
 	name := strings.TrimSpace(filename)
 
 	if !strings.HasPrefix(name, s3Prefix) {
@@ -40,17 +41,25 @@ func createS3Storage(filename string) (*S3Storage, bool, error) {
 		CacheDir:      cacheDir,
 		CacheFile:     s3Filename,
 		CacheFilePath: fmt.Sprintf("%s/%s", cacheDir, s3Filename),
+		Credentials:   credentials,
 		Bucket:        bucket,
 		Key:           key,
 	}, true, nil
 }
 
+type AWSCredentials struct {
+	Region          string `yaml:"region"`
+	AccessKeyId     string `yaml:"access-key-id"`
+	SecretAccessKey string `yaml:"secret-access-key"`
+}
+
 type S3Storage struct {
 	Bucket        string
 	Key           string
 	CacheFile     string
 	CacheDir      string
 	CacheFilePath string
+	Credentials   *AWSCredentials
 }
 
 func (s3 *S3Storage) CreateDumpFile() (*os.File, error) {
@@ -84,7 +93,18 @@ func (s3 *S3Storage) Upload() error {
 		}
 	}()
 
-	session := session.Must(session.NewSession())
+	var awsConfig aws.Config
+	if s3.Credentials != nil {
+		if s3.Credentials.Region != "" {
+			awsConfig.Region = aws.String(s3.Credentials.Region)
+		}
+
+		if s3.Credentials.AccessKeyId != "" && s3.Credentials.SecretAccessKey != "" {
+			awsConfig.Credentials = credentials.NewStaticCredentials(s3.Credentials.AccessKeyId, s3.Credentials.SecretAccessKey, "")
+		}
+	}
+
+	session := session.Must(session.NewSession(&awsConfig))
 	uploader := s3manager.NewUploader(session)
 
 	log.Printf("uploading file %s to s3...", uploadFile.Name())
diff --git a/storage/s3_test.go b/storage/s3_test.go
deleted file mode 100644
index 53f91d2..0000000
--- a/storage/s3_test.go
+++ /dev/null
@@ -1,28 +0,0 @@
-package storage
-
-import (
-	"errors"
-	"testing"
-)
-
-func TestCreateS3Storage(t *testing.T) {
-	store, ok, err := createS3Storage("://")
-	if ok || store != nil || err != nil {
-		t.Error("it should create no store, returns not ok without err")
-	}
-
-	store, ok, err = createS3Storage("s3://fdfdf")
-	if ok || store != nil || !errors.Is(err, ErrInvalidS3Path) {
-		t.Error("it is an invalid s3 filename, it should create no store, returns not ok with ErrInvalidS3Path")
-	}
-
-	store, ok, err = createS3Storage("s3://bucket/path/to/file.jpg")
-
-	if !ok || err != nil {
-		t.Error("expected it should create a s3 storage", err)
-	}
-
-	if store.Bucket != "bucket" || store.Key != "path/to/file.jpg" || store.CacheFile != "file.jpg" || store.CacheDir != uploadCacheDir() {
-		t.Errorf("store has unexpected fields: %+v", store)
-	}
-}
diff --git a/storage/storage.go b/storage/storage.go
index a90929b..78834d1 100644
--- a/storage/storage.go
+++ b/storage/storage.go
@@ -33,20 +33,3 @@ func uploadCacheDir() string {
 
 	return fmt.Sprintf("%s/%s", dir, uploadDumpCacheDir)
 }
-
-// Factory method to create the storage struct based on filename.
-func CreateStorage(filename string) (Storage, error) {
-	s3Storage, ok, err := createS3Storage(filename)
-
-	if err != nil {
-		return nil, err
-	}
-
-	if ok {
-		return s3Storage, nil
-	}
-
-	return &LocalStorage{
-		Filename: filename,
-	}, nil
-}