dbquery/transaction_chain.go

package dbquery

/**
 * 事务操作
 */
import (
	"database/sql"
	"errors"
	"log"
	"strconv"
	"strings"

	"git.tetele.net/tgo/helper"
)

type TxQuery struct {
	dbname    string
	table     string
	alias     string
	title     string
	where     []string
	where_or  []string
	join      [][]string //[["tablea as a","a.id=b.id","left"]]
	data      []string
	value     []interface{}
	save_data []map[string]interface{} //批量操作的数据[["title":"a","num":1,],["title":"a","num":1,]]
	upd_field []string                 // 批量更新时需要更新的字段，为空时按除id外的字段进行更新
	orderby   string
	groupby   string
	having    string
	page      int
	page_size int
	stmt      *sql.Stmt
	conn      *sql.DB
	tx        *sql.Tx
	debug     bool
}

func NewTxQuery(t ...string) *TxQuery {

	var conn_type *sql.DB = DB

	if len(t) > 0 {
		switch t[0] {
		case "mysql":
			conn_type = DB

		case "mssql": //sql server
			conn_type = MSDB_CONN

		}
	}

	tx, err := conn_type.Begin()
	if err != nil {
		log.Println("start tx begin error", err)
	}

	return &TxQuery{
		conn: conn_type,
		tx:   tx,
	}
}

func (this *TxQuery) Conn(conn *sql.DB) *TxQuery {
	this.conn = conn
	return this
}
func (this *TxQuery) Db(dbname string) *TxQuery {
	this.dbname = dbname
	return this
}

func (this *TxQuery) Table(tablename string) *TxQuery {
	this.table = tablename
	return this
}
func (this *TxQuery) Alias(tablename string) *TxQuery {
	this.alias = tablename
	return this
}

func (this *TxQuery) Title(title string) *TxQuery {
	this.title = title
	return this
}
func (this *TxQuery) Page(page int) *TxQuery {
	this.page = page
	return this
}
func (this *TxQuery) PageSize(page_num int) *TxQuery {
	this.page_size = page_num
	return this
}

func (this *TxQuery) Orderby(orderby string) *TxQuery {
	this.orderby = orderby
	return this
}
func (this *TxQuery) Groupby(groupby string) *TxQuery {
	this.groupby = groupby
	return this
}
func (this *TxQuery) Having(having string) *TxQuery {
	this.having = having
	return this
}
func (this *TxQuery) Where(where string) *TxQuery {
	this.where = append(this.where, where)
	return this
}
func (this *TxQuery) Wheres(wheres []string) *TxQuery {
	if len(wheres) > 0 {
		this.where = append(this.where, wheres...)
	}
	return this
}
func (this *TxQuery) WhereOr(where string) *TxQuery {
	this.where_or = append(this.where_or, where)
	return this
}
func (this *TxQuery) Value(value interface{}) *TxQuery {
	this.value = append(this.value, value)
	return this
}
func (this *TxQuery) SaveData(value map[string]interface{}) *TxQuery {
	this.save_data = append(this.save_data, value)
	return this
}
func (this *TxQuery) SaveDatas(value []map[string]interface{}) *TxQuery {
	this.save_data = append(this.save_data, value...)
	return this
}
func (this *TxQuery) UpdField(value string) *TxQuery {
	this.upd_field = append(this.upd_field, value)
	return this
}
func (this *TxQuery) UpdFields(value []string) *TxQuery {
	this.upd_field = append(this.upd_field, value...)
	return this
}
func (this *TxQuery) Values(values []interface{}) *TxQuery {
	this.value = append(this.value, values...)
	return this
}
func (this *TxQuery) Join(join []string) *TxQuery {
	this.join = append(this.join, join)
	return this
}
func (this *TxQuery) Data(data string) *TxQuery {
	this.data = append(this.data, data)
	return this
}
func (this *TxQuery) Datas(datas []string) *TxQuery {
	this.data = append(this.data, datas...)
	return this
}
func (this *TxQuery) Debug(debug bool) *TxQuery {
	this.debug = debug
	return this
}

/*
 * 清理上次查询
 */
func (this *TxQuery) Clean() *TxQuery {
	this.title = ""
	this.where = this.where[0:0]
	this.where_or = this.where_or[0:0]
	this.join = this.join[0:0]
	this.data = this.data[0:0]
	this.value = this.value[0:0]
	this.orderby = ""
	this.groupby = ""
	this.page = 0
	this.page_size = 0
	this.save_data = this.save_data[0:0]
	this.upd_field = this.upd_field[0:0]
	this.having = ""
	return this
}

//构造子查询
func (this *TxQuery) BuildSelectSql() (map[string]interface{}, error) {
	if this.dbname == "" && this.table == "" {
		return nil, errors.New("参数错误，没有数据表")
	}
	var table = ""
	if strings.Contains(this.table, "select ") {
		table = this.table
	} else {
		table = getTableName(this.dbname, this.table)
	}

	var sql, title string

	if this.title != "" {
		title = this.title
	} else {
		title = "*"
	}
	sql = helper.StringJoin("select ", title)

	if this.alias != "" {
		table = helper.StringJoin(table, " as ", this.alias)
	}

	sql = helper.StringJoin(sql, " from ", table)

	if len(this.join) > 0 {
		for _, joinitem := range this.join {
			if len(joinitem) < 2 {
				continue
			}
			if len(joinitem) == 3 {
				sql = helper.StringJoin(sql, " ", joinitem[2], " join ", getTableName(this.dbname, joinitem[0]), " on ", joinitem[1])
			} else { //默认左连接
				sql = helper.StringJoin(sql, " left join ", getTableName(this.dbname, joinitem[0]), " on ", joinitem[1])
			}
		}
	}
	if len(this.where) > 0 || len(this.where_or) > 0 {
		sql = helper.StringJoin(sql, " where ")
	}
	if len(this.where) > 0 {
		sql = helper.StringJoin(sql, " (", strings.Join(this.where, " and "), " ) ")
	}
	if len(this.where_or) > 0 {
		if len(this.where) > 0 {
			sql = helper.StringJoin(sql, " or ", strings.Join(this.where_or, " or "))
		} else {
			sql = helper.StringJoin(sql, strings.Join(this.where_or, " or "))
		}
	}
	if this.groupby != "" {
		sql = helper.StringJoin(sql, " group by ", this.groupby)

	}
	if this.having != "" {
		sql = helper.StringJoin(sql, " having ", this.having)

	}
	if this.orderby != "" {
		sql = helper.StringJoin(sql, " order by ", this.orderby)
	}

	if this.page > 0 || this.page_size > 0 {

		if this.page < 1 {
			this.page = 1
		}
		if this.page_size < 1 {
			this.page_size = 10
		}
		from := strconv.Itoa((this.page - 1) * this.page_size)
		offset := strconv.Itoa(this.page_size)
		if from != "" && offset != "" {
			sql = helper.StringJoin(sql, " limit ", from, " , ", offset)
		}
	}

	if this.debug {
		log.Println("query sql:", sql, this.value)
	}

	condition_len := 0 //所有条件数
	for _, ch2 := range sql {
		if string(ch2) == "?" {
			condition_len++
		}
	}
	if condition_len != len(this.value) {
		return nil, errors.New("参数错误，条件值错误")
	}
	return map[string]interface{}{
		"sql":   sql,
		"value": this.value,
	}, nil
}

// 拼查询sql
func (this *TxQuery) QueryStmt() error {

	res := map[string]interface{}{}
	res, err = this.BuildSelectSql()
	if err != nil {
		return err
	}
	sql := helper.ToStr(res["sql"])

	stmt, err = this.tx.Prepare(sql + " FOR UPDATE")

	if err != nil {
		return err
	}

	this.stmt = stmt

	return nil
}

// 拼更新sql
func (this *TxQuery) UpdateStmt() error {

	if this.dbname == "" && this.table == "" {
		return errors.New("参数错误，没有数据表")
	}
	if len(this.where) < 1 {
		return errors.New("参数错误，缺少条件")
	}

	dbName := getTableName(this.dbname, this.table)

	var sql string

	sql = helper.StringJoin("update ", dbName, " set ", strings.Join(this.data, " , "))

	sql = helper.StringJoin(sql, " where ", strings.Join(this.where, " and "))

	if this.debug {
		log.Println("update sql:", sql, this.value)
	}

	condition_len := 0 //所有条件数

	for _, ch2 := range sql {
		if string(ch2) == "?" {
			condition_len++
		}
	}
	if condition_len != len(this.value) {
		return errors.New("参数错误，条件值错误")
	}

	stmt, err = this.tx.Prepare(sql)

	if err != nil {
		return err
	}

	this.stmt = stmt

	return nil
}

// 拼批量存在更新不存在插入sql
func (this *TxQuery) UpdateAllStmt() error {

	if this.dbname == "" && this.table == "" {
		return errors.New("参数错误，没有数据表")
	}

	dbName := getTableName(this.dbname, this.table)

	var sql string
	var dataSql = []string{}
	var valSql = []string{}
	var updSql = []string{}
	var updFieldLen = len(this.upd_field)
	if len(this.save_data) > 0 {
		this.data = []string{}
		this.value = []interface{}{}
		for i, datum := range this.save_data {
			if i == 0 {
				for k, _ := range datum {
					this.data = append(this.data, k)
					dataSql = append(dataSql, "?")
					if updFieldLen == 0 && k != "id" {
						updSql = append(updSql, k+"=values("+k+")")
					}
				}
				if updFieldLen > 0 {
					for _, k := range this.upd_field {
						updSql = append(updSql, k+"=values("+k+")")
					}
				}
			}
			for _, k := range this.data {
				this.value = append(this.value, datum[k])
			}
			valSql = append(valSql, "("+strings.Join(dataSql, " , ")+")")
		}
	} else {
		for _, datum := range this.data {
			dataSql = append(dataSql, "?")
			if updFieldLen == 0 && datum != "id" {
				updSql = append(updSql, datum+"=values("+datum+")")
			}
		}
		if updFieldLen > 0 {
			for _, k := range this.upd_field {
				updSql = append(updSql, k+"=values("+k+")")
			}
		}
		valSql = append(valSql, "("+strings.Join(dataSql, " , ")+")")
	}

	if len(this.data) == 0 {
		return errors.New("参数错误，没有数据表")
	}
	if len(this.value) == 0 {
		return errors.New("参数错误，条件值错误")
	}
	setText := " values "
	if len(valSql) > 1 {
		setText = " value "
	}
	sql = helper.StringJoin("insert into ", dbName, " (", strings.Join(this.data, " , "), ")", setText, strings.Join(valSql, ","))
	if len(this.value) == 0 {
		return errors.New("参数错误，条件值错误")
	}

	if this.debug {
		log.Println("insert on duplicate key update sql:", sql, this.value)
	}

	stmt, err = this.tx.Prepare(sql)

	if err != nil {
		return err
	}

	this.stmt = stmt

	return nil
}

// 拼插入sql
func (this *TxQuery) CreateStmt() error {

	if this.dbname == "" && this.table == "" {
		return errors.New("参数错误，没有数据表")
	}

	dbName := getTableName(this.dbname, this.table)

	var sql string

	sql = helper.StringJoin("insert into ", dbName, " set ", strings.Join(this.data, " , "))

	if this.debug {
		log.Println("insert sql:", sql, this.value)
	}

	condition_len := 0 //所有条件数

	for _, ch2 := range sql {
		if string(ch2) == "?" {
			condition_len++
		}
	}
	if condition_len != len(this.value) {
		return errors.New("参数错误，条件值错误")
	}

	stmt, err = this.tx.Prepare(sql)

	if err != nil {
		return err
	}

	this.stmt = stmt

	return nil
}

// 拼批量插入sql
func (this *TxQuery) CreateAllStmt() error {

	if this.dbname == "" && this.table == "" {
		return errors.New("参数错误，没有数据表")
	}

	dbName := getTableName(this.dbname, this.table)

	var sql string
	var dataSql = []string{}
	var valSql = []string{}
	if len(this.save_data) > 0 {
		this.data = []string{}
		this.value = []interface{}{}
		for i, datum := range this.save_data {
			if i == 0 {
				for k, _ := range datum {
					this.data = append(this.data, k)
					dataSql = append(dataSql, "?")
				}
			}
			for _, k := range this.data {
				this.value = append(this.value, datum[k])
			}
			valSql = append(valSql, "("+strings.Join(dataSql, " , ")+")")
		}
	} else {
		for i := 0; i < len(this.data); i++ {
			dataSql = append(dataSql, "?")
		}
		valSql = append(valSql, "("+strings.Join(dataSql, " , ")+")")
	}

	if len(this.data) == 0 {
		return errors.New("参数错误，没有数据表")
	}
	if len(this.value) == 0 {
		return errors.New("参数错误，条件值错误")
	}
	setText := " values "
	if len(valSql) > 1 {
		setText = " value "
	}
	sql = helper.StringJoin("insert into ", dbName, " (", strings.Join(this.data, " , "), ")", setText, strings.Join(valSql, ","))
	if len(this.value) == 0 {
		return errors.New("参数错误，条件值错误")
	}

	if this.debug {
		log.Println("insert sql:", sql, this.value)
	}

	stmt, err = this.tx.Prepare(sql)

	if err != nil {
		return err
	}

	this.stmt = stmt

	return nil
}

// 拼删除sql
func (this *TxQuery) DeleteStmt() error {

	if this.dbname == "" && this.table == "" {
		return errors.New("参数错误，没有数据表")
	}
	if len(this.where) < 1 {
		return errors.New("参数错误，缺少条件")
	}

	dbName := getTableName(this.dbname, this.table)

	var sql string

	sql = helper.StringJoin("delete from ", dbName, " where ", strings.Join(this.where, " and "))

	if this.page_size > 0 {
		sql = helper.StringJoin(sql, " limit ", strconv.Itoa(this.page_size))
	}

	if this.debug {
		log.Println("delete sql:", sql, this.value)
	}

	condition_len := 0 //所有条件数

	for _, ch2 := range sql {
		if string(ch2) == "?" {
			condition_len++
		}
	}
	if condition_len != len(this.value) {
		return errors.New("参数错误，条件值错误")
	}

	stmt, err = this.tx.Prepare(sql)

	if err != nil {
		return err
	}

	this.stmt = stmt

	return nil
}

/**
 * 执行查询列表
 * return list error
 */
func (this *TxQuery) Select() ([]map[string]string, error) {

	err := this.QueryStmt()
	if err != nil {
		return []map[string]string{}, err
	}

	if this.stmt == nil {
		return []map[string]string{}, errors.New("缺少必要参数")
	}

	return StmtForQueryList(this.stmt, this.value)
}

/**
 * 执行查询一条数据
 * return row error
 */
func (this *TxQuery) Find() (map[string]string, error) {
	this.page = 1
	this.page_size = 1
	err := this.QueryStmt()
	if err != nil {
		return map[string]string{}, err
	}

	if this.stmt == nil {
		return nil, errors.New("缺少必要参数")
	}
	return StmtForQueryRow(this.stmt, this.value)
}

/**
 * 执行更新
 * return is_updated error
 */
func (this *TxQuery) Update() (int64, error) {

	err := this.UpdateStmt()
	if err != nil {
		return 0, err
	}

	return StmtForUpdateExec(this.stmt, this.value)
}

//批量更新
func (this *TxQuery) UpdateAll() (int64, error) {

	err := this.UpdateAllStmt()
	if err != nil {
		return 0, err
	}

	return StmtForUpdateExec(this.stmt, this.value)
}

/**
 * 执行删除
 * return is_delete error
 */
func (this *TxQuery) Delete() (int64, error) {

	err := this.DeleteStmt()
	if err != nil {
		return 0, err
	}

	return StmtForUpdateExec(this.stmt, this.value)
}

/**
 * 执行写入
 * return is_insert error
 */
func (this *TxQuery) Create() (int64, error) {

	err := this.CreateStmt()
	if err != nil {
		return 0, err
	}

	return StmtForInsertExec(this.stmt, this.value)
}

/**
 * 执行批量写入
 * return is_insert error
 */
func (this *TxQuery) CreateAll() (int64, error) {

	err := this.CreateAllStmt()
	if err != nil {
		return 0, err
	}

	return StmtForInsertExec(this.stmt, this.value)
}

/**
 * 提交
 */
func (this *TxQuery) Commit() error {
	return this.tx.Commit()
}

/**
 * 回滚
 */
func (this *TxQuery) Rollback() error {
	return this.tx.Rollback()
}