ds-sdk/sdk/sdk.go

package sdk

import (
	"errors"
	"git.sxidc.com/go-tools/utils/strutils"
	"git.sxidc.com/service-supports/ds-sdk/client"
	"git.sxidc.com/service-supports/ds-sdk/grpc_client"
	"git.sxidc.com/service-supports/ds-sdk/grpc_client/v1/request"
	"io"
)

type SDK struct {
	options    *Options
	client     *client.Client
	grpcClient *grpc_client.Client
}

func (s *SDK) ExecuteRawSql(sql string, executeParams map[string]any) ([]SqlResult, error) {
	if strutils.IsStringEmpty(sql) {
		return make([]SqlResult, 0), nil
	}

	options := s.options

	tableRows, err := s.client.ExecuteRawSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, sql, executeParams)
	if err != nil {
		return nil, err
	}

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

	return results, nil
}

func (s *SDK) CreateSQL(name string, spec map[string]any) error {
	if strutils.IsStringEmpty(name) {
		return errors.New("没有传递SQL资源名称")
	}

	options := s.options

	err := s.client.CreateSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, name, spec)
	if err != nil {
		return err
	}

	return nil
}

func (s *SDK) DeleteSQL(name string) error {
	if strutils.IsStringEmpty(name) {
		return errors.New("没有传递SQL资源名称")
	}

	options := s.options

	err := s.client.DeleteSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, name)
	if err != nil {
		return err
	}

	return nil
}

func (s *SDK) ExecuteSql(name string, executeParams map[string]any) ([]SqlResult, error) {
	if strutils.IsStringEmpty(name) {
		return nil, errors.New("没有传递SQL资源名称")
	}

	options := s.options

	tableRows, err := s.client.ExecuteSql(options.token, options.baseUrl,
		options.namespace, options.dataSource, name, executeParams)
	if err != nil {
		return nil, err
	}

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

	return results, nil
}

func (s *SDK) Transaction(txFunc TxFunc) error {
	stream, err := s.grpcClient.Transaction()
	if err != nil {
		return err
	}

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

	err = stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_TransactionBeginRequest{
			TransactionBeginRequest: &request.TransactionBeginRequest{
				Token:      s.options.token,
				Namespace:  s.options.namespace,
				DataSource: s.options.dataSource,
			},
		}})
	if err != nil {
		return err
	}

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

	if !resp.Success {
		return errors.New(resp.Msg)
	}

	err = txFunc(&Transaction{
		stream: stream,
	})
	if err != nil {
		return err
	}

	err = stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_TransactionEndRequest{
			TransactionEndRequest: &request.TransactionEndRequest{},
		}})
	if err != nil {
		return err
	}

	_, err = stream.Recv()
	if err != nil && err != io.EOF {
		return err
	}

	return nil
}