「GoCN酷Go推荐」sql 发布管理工具 gopg-migrations-轻识

1. gopg-migrations 是什么？

gopg-migrations 在我们使用数据库时，每次发布都要改一些表结构，或者修改数据之类的操作。

这些操作，手动执行既危险，又容易忘记；当新人看到奇怪的表结构，感觉到很诧异。如果我们能将所有操作管理起来，本地测试好之后再发布上去，那么就不会有上述这些问题。

因此，gopg-migrations 这个库，给我们带来非常大的便利。当然，这里是对 postgres 数据库的支持。

2. 怎么使用

下载扩展库 github.com/go-pg/migrations/v7即可使用。

2.1 假如我们先创建一个数据库 students，然后调用 migrations 去执行代码。

步骤如下：

编写创建表的 sql，使用 migrations 目录管理起来，num_xxx 表示按照 num 顺序执行主逻辑 main.go 主逻辑如下：

package main

import (
    "fmt"
    "time"

    pg "github.com/test_go_pg/pg"

    "go.uber.org/zap"
)

func main() {
    fmt.Println("Starting go-pg-migrations...")

    // Bootstrap check pg
    if err := pg.PGDBWrite.Ping(); err != nil {
        fmt.Println(pg.DBWriteConnectionError, zap.Error(err))
        return
    }
    fmt.Println("PostgreSQL is running",
        zap.String("user", pg.PGDBWrite.Options().User),
        zap.String("addr", pg.PGDBWrite.Options().Addr),
        zap.String("db", pg.PGDBWrite.Options().Database))

    // Migrate to latest pg schema
    if err := pg.PGDBWrite.Migrate(); err != nil {
        fmt.Println(pg.DBMigrationError, zap.Error(err))
        return
    }

    time.Sleep(time.Hour)
    fmt.Println("go-pg-migrations is stopping...")
}

gopg-migrations 相关函数编码如下：

package pg

import (
    "context"
    "fmt"
    "time"

    "github.com/go-pg/migrations/v7"
    "github.com/go-pg/pg/v9"
    "go.uber.org/zap"
)

type TGPGDB struct {
    *pg.DB
}

func NewTGPGDB(db *pg.DB) *TGPGDB {
    return &TGPGDB{db}
}

type TGPGDBOptions struct {
    Url string

    DisableBeforeQueryLog bool
    DisableAfterQueryLog  bool
}

func CreateTGPGDB(url string) *TGPGDB {
    return CreateTGPGDBWithOptions(&TGPGDBOptions{Url: url})
}

func CreateTGPGDBWithOptions(dbOpts *TGPGDBOptions) *TGPGDB {
    opts, err := pg.ParseURL(dbOpts.Url)
    if err != nil {
        fmt.Println(DBURLParseError, zap.String("URL", dbOpts.Url), zap.Error(err))
        return nil
    }
    opts.ReadTimeout = DBReadTimeout
    opts.WriteTimeout = DBWriteTimeout
    opts.TLSConfig = nil // disabled for faster local connection (even in production)
    if DBNumConns > 0 {
        opts.PoolSize = DBNumConns
    }

    db := NewTGPGDB(pg.Connect(opts))
    return db
}

// Ping simulates a "blank query" behavior similar to lib/pq's
// to check if the db connection is alive.
func (db *TGPGDB) Ping() error {
    _, err := db.ExecOne("SELECT 1")
    return err
}

// Migrate check and migrate to lastest db version.
func (db *TGPGDB) Migrate() error {
    // Make sure to only search specified migrations dir
    cl := migrations.NewCollection()
    cl.DisableSQLAutodiscover(true)
    err := cl.DiscoverSQLMigrations(DBMigrationsDir)
    if err != nil {
        return err
    }

    var oldVersion, newVersion int64
    // Run all migrations in a transaction so we rollback if migrations fail anywhere
    err = db.RunInTransaction(func(tx *pg.Tx) error {
        // Intentionally ignore harmless errors on initializing gopg_migrations
        _, _, err = cl.Run(db, "init")
        //if err != nil && !DBMigrationsAlreadyInit(err) {
        if err != nil{
            return err
        }
        oldVersion, newVersion, err = cl.Run(db, "up")
        return err
    })
    if err != nil {
        return err
    }
    if newVersion == oldVersion {
        fmt.Println("db schema up to date")
    } else {
        fmt.Println("db schema migrated successfully", zap.Int64("from", oldVersion), zap.Int64("to", newVersion))
    }
    return nil
}

// WithContextTimeout
func WithContextTimeout(ctx context.Context, f func(ctx context.Context)) {
    WithContextTimeoutValue(ctx, DBStmtTimeout, f)
}

// WithContextTimeoutValue
func WithContextTimeoutValue(ctx context.Context, timeout time.Duration, f func(ctx context.Context)) {
    // check context timeout setting with upper bound read/write limit
    if timeout > DBReadTimeout && timeout > DBWriteTimeout {
        fmt.Println(DBContextTimeoutExceedUpperBound,
            zap.Error(fmt.Errorf("query timeout %s exceed upper bound (%s|%s)", timeout, DBReadTimeout, DBWriteTimeout)))
    }
    newCtx, cancel := context.WithTimeout(ctx, timeout)
    f(newCtx)
    cancel()
}