Queries encapsulating Sqlc to implement more convenient transaction operations

What is SQLC

SQLC is a powerful development tool whose core function is to convert SQL queries into type-safe Go code. By parsing SQL statements and analyzing database structures, sqlc can automatically generate corresponding Go structures and functions, greatly simplifying the code writing process for database operations.

Using sqlc, developers can focus on writing SQL queries and leave the tedious Go code generation work to the tool, thereby accelerating the development process and improving code quality.

SQLC transaction implementation

The code generated by Sqlc usually contains a Queries structure, which encapsulates all database operations. This structure implements a general Querier interface, which defines all database query methods.

The key is that the New function generated by sqlc can accept any object that implements the DBTX interface, including *sql.DB and *sql.Tx.

The core of transaction implementation is to utilize Go's interface polymorphism. When you need to perform operations within a transaction, you create a *sql.Tx object and then pass it to the New function to create a new Queries instance. This instance will perform all operations within the context of a transaction.

Suppose we connect to the Postgres database through pgx and initialize Queries with the following code.

var Pool *pgxpool.Pool
var Queries *sqlc.Queries

func init() {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
    defer cancel()

    connConfig, err := pgxpool.ParseConfig("postgres://user:password@127.0.0.1:5432/db?sslmode=disable")
    if err != nil {
        panic(err)
    }

    pool, err := pgxpool.NewWithConfig(ctx, connConfig)
    if err != nil {
        panic(err)
    }
    if err := pool.Ping(ctx); err != nil {
        panic(err)
    }

    Pool = pool
    Queries = sqlc.New(pool)
}

Encapsulation of transactions

The following code is a clever sqlc transaction encapsulation, which simplifies the process of using database transactions in Go. The function accepts a context and a callback function as parameters. This callback function is the specific operation the user wants to perform in the transaction.

func WithTransaction(ctx context.Context, callback func(qtx *sqlc.Queries) (err error)) (err error) {
    tx, err := Pool.Begin(ctx)
    if err != nil {
        return err
    }
    defer func() {
        if e := tx.Rollback(ctx); e != nil && !errors.Is(e, pgx.ErrTxClosed) {
            err = e
        }
    }()

    if err := callback(Queries.WithTx(tx)); err != nil {
        return err
    }

    return tx.Commit(ctx)
}

The function first starts a new transaction and then delays execution to ensure that the transaction will eventually be rolled back unless it is explicitly committed. This is a safety mechanism to prevent unfinished transactions from occupying resources. Next, the function calls the user-provided callback, passing in a query object with a transaction context, allowing the user to perform the required database operations within the transaction.

If the callback executes successfully without errors, the function commits the transaction. Any errors that occur during the process will cause the transaction to be rolled back. This method not only ensures data consistency, but also greatly simplifies error handling.

The elegance of this encapsulation is that it hides complex transaction management logic behind a simple function call. Users can focus on writing business logic without worrying about starting, committing, or rolling back transactions.

The usage of this code is quite intuitive. You can call the db.WithTransaction function where you need to perform a transaction, passing in a function as a parameter that defines all the database operations you want to perform within the transaction.

err := db.WithTransaction(ctx, func(qtx *sqlc.Queries) error {
    // 在这里执行你的数据库操作
    // 例如：
    _, err := qtx.CreateUser(ctx, sqlc.CreateUserParams{
        Name: "Alice",
        Email: "alice@example.com",
    })
    if err != nil {
        return err
    }

    _, err = qtx.CreatePost(ctx, sqlc.CreatePostParams{
        Title: "First Post",
        Content: "Hello, World!",
        AuthorID: newUserID,
    })
    if err != nil {
        return err
    }

    // 如果所有操作都成功，返回 nil
    return nil
})

if err != nil {
    // 处理错误
    log.Printf("transaction failed: %v", err)
} else {
    log.Println("transaction completed successfully")
}

In this example, we create a user and a post in a transaction. If any operation fails, the entire transaction is rolled back. If all operations are successful, the transaction is committed.

The benefit of this approach is that you don’t need to manually manage the start, commit or rollback of the transaction, all of this is handled by the db.WithTransaction function. You only need to focus on the actual database operations performed within the transaction. This greatly simplifies the code and reduces the possibility of errors.

Further packaging

The packaging method mentioned above is not without its shortcomings.

This simple transaction encapsulation has limitations when dealing with nested transactions. This is because it creates a new transaction every time instead of checking if you are already in one.

In order to implement nested transaction processing, we must obtain the current transaction object, but the current transaction object is hidden inside sqlc.Queries, so we must extend sqlc.Queries.

The structure that extends sqlc.Queries is created by us as Repositories. It extends *sqlc.Queries and adds a new attribute pool, which is a pointer of type pgxpool.Pool.

type Repositories struct {
    *sqlc.Queries
    pool *pgxpool.Pool
}

func NewRepositories(pool *pgxpool.Pool) *Repositories {
    return &Repositories{
        pool:    pool,
        Queries: sqlc.New(pool),
    }
}

But when we start writing code, we will find that *pgxpool.Pool does not satisfy the pgx.Tx interface. This is because *pgxpool.Pool lacks the Rollback and Commit methods. It only contains Begin for starting a transaction. Method, in order to solve this problem, we continue to extend Repositories, add a new attribute tx to it, and add a new NewRepositoriesTx method to it.

type Repositories struct {
    *sqlc.Queries
    tx   pgx.Tx
    pool *pgxpool.Pool
}

func NewRepositoriesTx(tx pgx.Tx) *Repositories {
    return &Repositories{
        tx:      tx,
        Queries: sqlc.New(tx),
    }
}

Now, there are both pool and tx attributes in our Repositories structure. This may not look very elegant. Why can't we abstract a unified TX type? In fact, it is the reason mentioned above, that is, *pgxpool.Pool There is only a method to start a transaction, but not a method to end the transaction. One way to solve this problem is to create another RepositoriesTX structure and store pgx.Tx in it instead of *pgxpool.Pool, but this may cause There are new questions. One of them is that we may have to implement the WithTransaction method for both of them. As for the other question, we will talk about it later. Now let us first implement the WithTransaction method of Repositories.

func (r *Repositories) WithTransaction(ctx context.Context, fn func(qtx *Repositories) (err error)) (err error) {
    var tx pgx.Tx
    if r.tx != nil {
        tx, err = r.tx.Begin(ctx)
    } else {
        tx, err = r.pool.Begin(ctx)
    }
    if err != nil {
        return err
    }
    defer func() {
        if e := tx.Rollback(ctx); e != nil && !errors.Is(e, pgx.ErrTxClosed) {
            err = e
        }
    }()

    if err := fn(NewRepositoriesTx(tx)); err != nil {
        return err
    }

    return tx.Commit(ctx)
}

这个方法和上一章节实现的 WithTransaction 主要不同是，他是实现在 *Repositories 上面而不是全局的，这样我们就可以通过 (r *Repositories) 中的 pgx.Tx 来开始嵌套事务了。

在没有开始事务的时候，我们可以调用 repositories.WithTransaction 来开启一个新的事务。

err := db.repositories.WithTransaction(ctx, func(tx *db.Repositories) error {

    return nil
})

多级事务也是没有问题的，非常容易实现。

err := db.repositories.WithTransaction(ctx, func(tx *db.Repositories) error {
    // 假设此处进行了一些数据操作
    // 然后，开启一个嵌套事务
    return tx.WithTransaction(ctx, func(tx *db.Repositories) error {
        // 这里可以在嵌套事务中进行一些操作
        return nil
    })
})

这个封装方案有效地确保了操作的原子性，即使其中任何一个操作失败，整个事务也会被回滚，从而保障了数据的一致性。

结束语

本文介绍了一个使用 Go 和 pgx 库封装 SQLC 数据库事务的方案。

核心是 Repositories 结构体，它封装了 SQLC 查询接口和事务处理逻辑。通过 WithTransaction 方法，我们可以在现有事务上开始新的子事务或在连接池中开始新的事务，并确保在函数返回时回滚事务。

构造函数 NewRepositories 和 NewRepositoriesTx 分别用于创建普通和事务内的 Repositories 实例。

这样可以将多个数据库操作封装在一个事务中，如果任何一个操作失败，事务将被回滚，提高了代码的可维护性和可读性。

The above is the detailed content of Queries encapsulating Sqlc to implement more convenient transaction operations. For more information, please follow other related articles on the PHP Chinese website!