123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128 |
- package sdk
- import (
- "encoding/json"
- "errors"
- v1 "git.sxidc.com/service-supports/ds-sdk/grpc_client/v1"
- "git.sxidc.com/service-supports/ds-sdk/grpc_client/v1/request"
- )
- type TxFunc func(tx *Transaction) error
- type Transaction struct {
- stream v1.SqlService_TransactionClient
- }
- func (tx *Transaction) ExecuteRawSql(sql string, executeParams map[string]any) ([]SqlResult, error) {
- var retErr error
- defer func() {
- if retErr != nil {
- innerErr := tx.stream.CloseSend()
- if innerErr != nil {
- panic(innerErr)
- }
- }
- }()
- executeParamsJsonBytes, err := json.Marshal(executeParams)
- if err != nil {
- retErr = err
- return nil, retErr
- }
- err = tx.stream.SendMsg(&request.TransactionOperation{
- Request: &request.TransactionOperation_ExecuteRawSqlRequest{
- ExecuteRawSqlRequest: &request.ExecuteRawSqlRequest{
- SQL: sql,
- ExecuteParams: string(executeParamsJsonBytes),
- },
- },
- })
- 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([]SqlResult, len(tableRows))
- for i, row := range tableRows {
- results[i] = row
- }
- return results, nil
- }
- func (tx *Transaction) ExecuteSql(name string, executeParams map[string]any) ([]SqlResult, error) {
- var retErr error
- defer func() {
- if retErr != nil {
- innerErr := tx.stream.CloseSend()
- if innerErr != nil {
- panic(innerErr)
- }
- }
- }()
- executeParamsJsonBytes, err := json.Marshal(executeParams)
- if err != nil {
- retErr = err
- return nil, retErr
- }
- err = tx.stream.Send(&request.TransactionOperation{
- Request: &request.TransactionOperation_ExecuteSqlRequest{
- ExecuteSqlRequest: &request.ExecuteSqlRequest{
- Name: name,
- ExecuteParams: string(executeParamsJsonBytes),
- },
- },
- })
- 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([]SqlResult, len(tableRows))
- for i, row := range tableRows {
- results[i] = row
- }
- return results, nil
- }
|