ds-sdk/sdk/transaction.go

package sdk

import (
	"encoding/json"
	"git.sxidc.com/go-tools/utils/reflectutils"
	"git.sxidc.com/service-supports/ds-sdk/grpc_client/v1"
	"git.sxidc.com/service-supports/ds-sdk/grpc_client/v1/request"
	"github.com/pkg/errors"
	"reflect"
)

type TxFunc func(tx *Transaction) error

type Transaction struct {
	stream v1.SqlService_TransactionClient
}

func (tx *Transaction) ExecuteRawSql(sql string, values ...any) ([]map[string]any, error) {
	var retErr error

	defer func() {
		if retErr != nil {
			innerErr := tx.stream.CloseSend()
			if innerErr != nil {
				panic(innerErr)
			}
		}
	}()

	rawSqlValues := make([]map[string]any, 0)
	for _, value := range values {
		typedValueReflectValue := reflect.ValueOf(value)
		if !typedValueReflectValue.IsValid() {
			return nil, errors.New("无效值")
		}

		typedValueReflectValueElem := reflectutils.PointerValueElem(typedValueReflectValue)
		values = append(values, map[string]any{
			"kind":  typedValueReflectValueElem.Kind(),
			"value": typedValueReflectValueElem.Interface(),
		})
	}

	requestMap := map[string]any{
		"sql":    sql,
		"values": rawSqlValues,
	}

	requestJsonBytes, err := json.Marshal(requestMap)
	if err != nil {
		retErr = err
		return nil, retErr
	}

	err = tx.stream.SendMsg(&request.TransactionOperation{
		Request: &request.TransactionOperation_ExecuteRawSqlJsonRequest{
			ExecuteRawSqlJsonRequest: string(requestJsonBytes),
		},
	})
	if err != nil {
		retErr = err
		return nil, retErr
	}

	resp, err := tx.stream.Recv()
	if err != nil {
		retErr = err
		return nil, retErr
	}

	if !resp.Success {
		retErr = errors.New(resp.Msg)
		return nil, retErr
	}

	tableRows := make([]map[string]any, 0)
	err = json.Unmarshal([]byte(resp.Results), &tableRows)
	if err != nil {
		retErr = err
		return nil, retErr
	}

	results := make([]map[string]any, len(tableRows))
	for i, row := range tableRows {
		results[i] = row
	}

	return results, nil
}