添加查询接口

dpsv1/client.go

@@ -3,11 +3,14 @@ package dpsv1
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"git.sxidc.com/service-supports/dps-sdk/pb/v1"
 	"git.sxidc.com/service-supports/dps-sdk/pb/v1/request"
+	"git.sxidc.com/service-supports/dps-sdk/pb/v1/response"
 	"git.sxidc.com/service-supports/dps-sdk/ports"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"time"
 )
 
 type Client struct {
@@ -49,7 +52,7 @@ func DestroyClient(client *Client) error {
 }
 
 func (c *Client) AutoMigrate(req *ports.AutoMigrateRequest) error {
-	tableModelDescribeJsonByte, err := json.Marshal(req.TableModelDescribe)
+	tableModelDescribeJsonBytes, err := json.Marshal(req.TableModelDescribe)
 	if err != nil {
 		return err
 	}
@@ -58,7 +61,7 @@ func (c *Client) AutoMigrate(req *ports.AutoMigrateRequest) error {
 		DatabaseID:            req.DatabaseID,
 		TablePrefixWithSchema: req.TablePrefixWithSchema,
 		Version:               req.Version,
-		TableModelDescribe:    tableModelDescribeJsonByte,
+		TableModelDescribe:    tableModelDescribeJsonBytes,
 	})
 	if err != nil {
 		return err
@@ -68,12 +71,12 @@ func (c *Client) AutoMigrate(req *ports.AutoMigrateRequest) error {
 }
 
 func (c *Client) Insert(req *ports.InsertRequest) (string, error) {
-	keysJsonByte, err := json.Marshal(req.Keys)
+	keysJsonBytes, err := json.Marshal(req.Keys)
 	if err != nil {
 		return "", err
 	}
 
-	tableRowJsonByte, err := json.Marshal(req.TableRow)
+	tableRowJsonBytes, err := json.Marshal(req.TableRow)
 	if err != nil {
 		return "", err
 	}
@@ -82,8 +85,8 @@ func (c *Client) Insert(req *ports.InsertRequest) (string, error) {
 		DatabaseID:            req.DatabaseID,
 		TablePrefixWithSchema: req.TablePrefixWithSchema,
 		Version:               req.Version,
-		Keys:                  keysJsonByte,
-		TableRow:              tableRowJsonByte,
+		Keys:                  keysJsonBytes,
+		TableRow:              tableRowJsonBytes,
 		UserID:                req.UserID,
 	})
 	if err != nil {
@@ -96,7 +99,7 @@ func (c *Client) Insert(req *ports.InsertRequest) (string, error) {
 // TODO InsertBatch
 
 func (c *Client) Delete(req *ports.DeleteRequest) (string, error) {
-	keysJsonByte, err := json.Marshal(req.Keys)
+	keysJsonBytes, err := json.Marshal(req.Keys)
 	if err != nil {
 		return "", err
 	}
@@ -105,7 +108,7 @@ func (c *Client) Delete(req *ports.DeleteRequest) (string, error) {
 		DatabaseID:            req.DatabaseID,
 		TablePrefixWithSchema: req.TablePrefixWithSchema,
 		Version:               req.Version,
-		Keys:                  keysJsonByte,
+		Keys:                  keysJsonBytes,
 		UserID:                req.UserID,
 	})
 	if err != nil {
@@ -118,7 +121,7 @@ func (c *Client) Delete(req *ports.DeleteRequest) (string, error) {
 // TODO DeleteBatch
 
 func (c *Client) Update(req *ports.UpdateRequest) (string, error) {
-	keysJsonByte, err := json.Marshal(req.Keys)
+	keysJsonBytes, err := json.Marshal(req.Keys)
 	if err != nil {
 		return "", err
 	}
@@ -132,7 +135,7 @@ func (c *Client) Update(req *ports.UpdateRequest) (string, error) {
 		DatabaseID:            req.DatabaseID,
 		TablePrefixWithSchema: req.TablePrefixWithSchema,
 		Version:               req.Version,
-		Keys:                  keysJsonByte,
+		Keys:                  keysJsonBytes,
 		NewTableRow:           newTableRowJsonByte,
 		UserID:                req.UserID,
 	})
@@ -144,7 +147,7 @@ func (c *Client) Update(req *ports.UpdateRequest) (string, error) {
 }
 
 func (c *Client) Replay(req *ports.ReplayRequest) (string, error) {
-	keysJsonByte, err := json.Marshal(req.Keys)
+	keysJsonBytes, err := json.Marshal(req.Keys)
 	if err != nil {
 		return "", err
 	}
@@ -153,7 +156,7 @@ func (c *Client) Replay(req *ports.ReplayRequest) (string, error) {
 		DatabaseID:            req.DatabaseID,
 		TablePrefixWithSchema: req.TablePrefixWithSchema,
 		Version:               req.Version,
-		Keys:                  keysJsonByte,
+		Keys:                  keysJsonBytes,
 		UserID:                req.UserID,
 	})
 	if err != nil {
@@ -162,3 +165,244 @@ func (c *Client) Replay(req *ports.ReplayRequest) (string, error) {
 
 	return reply.Statement, nil
 }
+
+func (c *Client) QueryByWhereAndOrderBy(req *ports.QueryByWhereAndOrderByRequest) (string, []map[string]any, int64, error) {
+	whereJsonBytes, err := json.Marshal(req.Where)
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	reply, err := c.queryServiceClient.QueryByWhereAndOrderBy(context.Background(), &request.QueryByWhereAndOrderByRequest{
+		DatabaseID:            req.DatabaseID,
+		TablePrefixWithSchema: req.TablePrefixWithSchema,
+		Version:               req.Version,
+		Select:                req.Select,
+		Where:                 whereJsonBytes,
+		OrderBy:               req.OrderBy,
+		PageNo:                req.PageNo,
+		PageSize:              req.PageSize,
+	})
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	infosMap, err := c.infoDataToInfoMapBatch(reply.Infos)
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	return reply.Statement, infosMap, reply.TotalCount, nil
+}
+
+func (c *Client) CommonQuery(req *ports.CommonQueryRequest) (string, []map[string]any, int64, error) {
+	whereJsonBytes, err := json.Marshal(req.Where)
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	orJsonBytes, err := json.Marshal(req.Or)
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	havingJsonBytes, err := json.Marshal(req.Having)
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	reply, err := c.queryServiceClient.CommonQuery(context.Background(), &request.CommonQueryRequest{
+		DatabaseID:            req.DatabaseID,
+		TablePrefixWithSchema: req.TablePrefixWithSchema,
+		Version:               req.Version,
+		Select:                req.Select,
+		Where:                 whereJsonBytes,
+		OrderBy:               req.OrderBy,
+		Or:                    orJsonBytes,
+		GroupBy:               req.GroupBy,
+		Join:                  req.Join,
+		Having:                havingJsonBytes,
+		PageNo:                req.PageNo,
+		PageSize:              req.PageSize,
+	})
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	infosMap, err := c.infoDataToInfoMapBatch(reply.Infos)
+	if err != nil {
+		return "", nil, 0, err
+	}
+
+	return reply.Statement, infosMap, reply.TotalCount, nil
+}
+
+func (c *Client) QueryByKeys(req *ports.QueryByKeysRequest) (string, map[string]any, error) {
+	keysJsonBytes, err := json.Marshal(req.Keys)
+	if err != nil {
+		return "", nil, err
+	}
+
+	whereJsonBytes, err := json.Marshal(req.Where)
+	if err != nil {
+		return "", nil, err
+	}
+
+	reply, err := c.queryServiceClient.QueryByKeys(context.Background(), &request.QueryByKeysRequest{
+		DatabaseID:            req.DatabaseID,
+		TablePrefixWithSchema: req.TablePrefixWithSchema,
+		Version:               req.Version,
+		Select:                req.Select,
+		Keys:                  keysJsonBytes,
+		Where:                 whereJsonBytes,
+	})
+	if err != nil {
+		return "", nil, err
+	}
+
+	infoMap, err := c.infoDataToInfoMap(reply.Info)
+	if err != nil {
+		return "", nil, err
+	}
+
+	return reply.Statement, infoMap, nil
+}
+
+func (c *Client) CommonQueryByKeys(req *ports.CommonQueryByKeysRequest) (string, map[string]any, error) {
+	keysJsonBytes, err := json.Marshal(req.Keys)
+	if err != nil {
+		return "", nil, err
+	}
+
+	whereJsonBytes, err := json.Marshal(req.Where)
+	if err != nil {
+		return "", nil, err
+	}
+
+	orJsonBytes, err := json.Marshal(req.Or)
+	if err != nil {
+		return "", nil, err
+	}
+
+	reply, err := c.queryServiceClient.CommonQueryByKeys(context.Background(), &request.CommonQueryByKeysRequest{
+		DatabaseID:            req.DatabaseID,
+		TablePrefixWithSchema: req.TablePrefixWithSchema,
+		Version:               req.Version,
+		Select:                req.Select,
+		Keys:                  keysJsonBytes,
+		Where:                 whereJsonBytes,
+		Or:                    orJsonBytes,
+	})
+	if err != nil {
+		return "", nil, err
+	}
+
+	infoMap, err := c.infoDataToInfoMap(reply.Info)
+	if err != nil {
+		return "", nil, err
+	}
+
+	return reply.Statement, infoMap, nil
+}
+
+func (c *Client) CountWhere(req *ports.CountWhereRequest) (string, int64, error) {
+	whereJsonBytes, err := json.Marshal(req.Where)
+	if err != nil {
+		return "", 0, err
+	}
+
+	reply, err := c.queryServiceClient.CountWhere(context.Background(), &request.CountWhereRequest{
+		DatabaseID:            req.DatabaseID,
+		TablePrefixWithSchema: req.TablePrefixWithSchema,
+		Version:               req.Version,
+		Where:                 whereJsonBytes,
+		PageNo:                req.PageNo,
+		PageSize:              req.PageSize,
+	})
+	if err != nil {
+		return "", 0, err
+	}
+
+	return reply.Statement, reply.Count, nil
+}
+
+func (c *Client) CommonCount(req *ports.CommonCountRequest) (string, int64, error) {
+	whereJsonBytes, err := json.Marshal(req.Where)
+	if err != nil {
+		return "", 0, err
+	}
+
+	orJsonBytes, err := json.Marshal(req.Or)
+	if err != nil {
+		return "", 0, err
+	}
+
+	havingJsonBytes, err := json.Marshal(req.Having)
+	if err != nil {
+		return "", 0, err
+	}
+
+	reply, err := c.queryServiceClient.CommonCount(context.Background(), &request.CommonCountRequest{
+		DatabaseID:            req.DatabaseID,
+		TablePrefixWithSchema: req.TablePrefixWithSchema,
+		Version:               req.Version,
+		Where:                 whereJsonBytes,
+		Or:                    orJsonBytes,
+		GroupBy:               req.GroupBy,
+		Join:                  req.Join,
+		Having:                havingJsonBytes,
+		PageNo:                req.PageNo,
+		PageSize:              req.PageSize,
+	})
+	if err != nil {
+		return "", 0, err
+	}
+
+	return reply.Statement, reply.Count, nil
+}
+
+func (c *Client) infoDataToInfoMap(infoData *response.InfoData) (map[string]any, error) {
+	retInfoMap := make(map[string]any)
+
+	for _, column := range infoData.Columns {
+		switch column.Value.Type {
+		case "Time":
+			timeObj := time.Unix(0, int64(column.Value.GetUint64Value()))
+			retInfoMap[column.Name] = timeObj
+		case "string":
+			retInfoMap[column.Name] = column.Value.GetStringValue()
+		case "bool":
+			retInfoMap[column.Name] = column.Value.GetBoolValue()
+		case "int32":
+			retInfoMap[column.Name] = int(column.Value.GetUint32Value())
+		case "int64":
+			retInfoMap[column.Name] = int64(column.Value.GetUint64Value())
+		case "uint32":
+			retInfoMap[column.Name] = column.Value.GetUint32Value()
+		case "uint64":
+			retInfoMap[column.Name] = column.Value.GetUint64Value()
+		case "float32":
+			retInfoMap[column.Name] = float32(column.Value.GetFloat64Value())
+		case "float64":
+			retInfoMap[column.Name] = column.Value.GetFloat64Value()
+		default:
+			return nil, errors.New("不支持的数据类型" + column.Value.Type)
+		}
+	}
+
+	return retInfoMap, nil
+}
+
+func (c *Client) infoDataToInfoMapBatch(infosData []*response.InfoData) ([]map[string]any, error) {
+	retInfosDataMap := make([]map[string]any, 0)
+
+	for _, infoData := range infosData {
+		retInfoMap, err := c.infoDataToInfoMap(infoData)
+		if err != nil {
+			return nil, err
+		}
+
+		retInfosDataMap = append(retInfosDataMap, retInfoMap)
+	}
+
+	return retInfosDataMap, nil
+}

ports/client.go

@@ -5,4 +5,10 @@ type Client interface {
 	Insert(request *InsertRequest) (string, error)
 	Delete(request *DeleteRequest) (string, error)
 	Update(request *UpdateRequest) (string, error)
+	QueryByWhereAndOrderBy(request *QueryByWhereAndOrderByRequest) (string, []map[string]any, int64, error)
+	CommonQuery(request *CommonQueryRequest) (string, []map[string]any, int64, error)
+	QueryByKeys(request *QueryByKeysRequest) (string, map[string]any, error)
+	CommonQueryByKeys(request *CommonQueryByKeysRequest) (string, map[string]any, error)
+	CountWhere(request *CountWhereRequest) (string, int64, error)
+	CommonCount(request *CommonCountRequest) (string, int64, error)
 }

ports/client_query_response.go

@@ -1,18 +0,0 @@
-package ports
-
-type QueryResponse struct {
-	Statement  string
-	Infos      []map[string]any
-	TotalCount int64
-	PageNo     int32
-}
-
-type QueryByKeysResponse struct {
-	Statement string
-	Info      map[string]any
-}
-
-type CountResponse struct {
-	Statement string
-	Count     int64
-}