client.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392
  1. package dpsv1
  2. import (
  3. "context"
  4. "encoding/json"
  5. "errors"
  6. "git.sxidc.com/service-supports/dps-sdk/pb/v1"
  7. "git.sxidc.com/service-supports/dps-sdk/pb/v1/request"
  8. "git.sxidc.com/service-supports/dps-sdk/pb/v1/response"
  9. "git.sxidc.com/service-supports/dps-sdk/ports"
  10. "google.golang.org/grpc"
  11. "google.golang.org/grpc/credentials/insecure"
  12. "reflect"
  13. "time"
  14. )
  15. type Client struct {
  16. conn *grpc.ClientConn
  17. commandServiceClient v1.CommandServiceClient
  18. queryServiceClient v1.QueryServiceClient
  19. }
  20. func NewClient(address string) (*Client, error) {
  21. conn, err := grpc.DialContext(context.Background(), address,
  22. grpc.WithTransportCredentials(insecure.NewCredentials()))
  23. if err != nil {
  24. return nil, err
  25. }
  26. return &Client{
  27. conn: conn,
  28. commandServiceClient: v1.NewCommandServiceClient(conn),
  29. queryServiceClient: v1.NewQueryServiceClient(conn),
  30. }, nil
  31. }
  32. func DestroyClient(client *Client) error {
  33. if client == nil {
  34. return nil
  35. }
  36. err := client.conn.Close()
  37. if err != nil {
  38. return err
  39. }
  40. client.conn = nil
  41. client.commandServiceClient = nil
  42. client.queryServiceClient = nil
  43. client = nil
  44. return nil
  45. }
  46. func (c *Client) AutoMigrate(req *ports.AutoMigrateRequest) error {
  47. tableModelDescribeJsonBytes, err := json.Marshal(req.TableModelDescribe)
  48. if err != nil {
  49. return err
  50. }
  51. _, err = c.commandServiceClient.AutoMigrate(context.Background(), &request.AutoMigrateRequest{
  52. DatabaseID: req.DatabaseID,
  53. TablePrefixWithSchema: req.TablePrefixWithSchema,
  54. Version: req.Version,
  55. TableModelDescribe: tableModelDescribeJsonBytes,
  56. })
  57. if err != nil {
  58. return err
  59. }
  60. return nil
  61. }
  62. func (c *Client) Insert(req *ports.InsertRequest) (string, error) {
  63. tableRowJsonBytes, err := c.formTableRow(req.TableRow)
  64. if err != nil {
  65. return "", err
  66. }
  67. reply, err := c.commandServiceClient.Insert(context.Background(), &request.InsertRequest{
  68. DatabaseID: req.DatabaseID,
  69. TablePrefixWithSchema: req.TablePrefixWithSchema,
  70. Version: req.Version,
  71. KeyColumns: req.KeyColumns,
  72. TableRow: tableRowJsonBytes,
  73. UserID: req.UserID,
  74. })
  75. if err != nil {
  76. return "", err
  77. }
  78. return reply.Statement, nil
  79. }
  80. // TODO InsertBatch
  81. func (c *Client) Delete(req *ports.DeleteRequest) (string, error) {
  82. reply, err := c.commandServiceClient.Delete(context.Background(), &request.DeleteRequest{
  83. DatabaseID: req.DatabaseID,
  84. TablePrefixWithSchema: req.TablePrefixWithSchema,
  85. Version: req.Version,
  86. KeyValues: req.KeyValues,
  87. UserID: req.UserID,
  88. })
  89. if err != nil {
  90. return "", err
  91. }
  92. return reply.Statement, nil
  93. }
  94. // TODO DeleteBatch
  95. func (c *Client) Update(req *ports.UpdateRequest) (string, error) {
  96. newTableRowJsonByte, err := c.formTableRow(req.NewTableRow)
  97. if err != nil {
  98. return "", err
  99. }
  100. reply, err := c.commandServiceClient.Update(context.Background(), &request.UpdateRequest{
  101. DatabaseID: req.DatabaseID,
  102. TablePrefixWithSchema: req.TablePrefixWithSchema,
  103. Version: req.Version,
  104. KeyValues: req.KeyValues,
  105. NewTableRow: newTableRowJsonByte,
  106. UserID: req.UserID,
  107. })
  108. if err != nil {
  109. return "", err
  110. }
  111. return reply.Statement, nil
  112. }
  113. func (c *Client) Replay(req *ports.ReplayRequest) (string, error) {
  114. reply, err := c.commandServiceClient.Replay(context.Background(), &request.ReplayRequest{
  115. DatabaseID: req.DatabaseID,
  116. TablePrefixWithSchema: req.TablePrefixWithSchema,
  117. Version: req.Version,
  118. KeyValues: req.KeyValues,
  119. UserID: req.UserID,
  120. })
  121. if err != nil {
  122. return "", err
  123. }
  124. return reply.Statement, nil
  125. }
  126. func (c *Client) QueryByWhereAndOrderBy(req *ports.QueryByWhereAndOrderByRequest) (string, []map[string]any, int64, error) {
  127. whereJsonBytes, err := json.Marshal(req.Where)
  128. if err != nil {
  129. return "", nil, 0, err
  130. }
  131. reply, err := c.queryServiceClient.QueryByWhereAndOrderBy(context.Background(), &request.QueryByWhereAndOrderByRequest{
  132. DatabaseID: req.DatabaseID,
  133. TablePrefixWithSchema: req.TablePrefixWithSchema,
  134. Version: req.Version,
  135. Select: req.Select,
  136. Where: whereJsonBytes,
  137. OrderBy: req.OrderBy,
  138. PageNo: req.PageNo,
  139. PageSize: req.PageSize,
  140. })
  141. if err != nil {
  142. return "", nil, 0, err
  143. }
  144. infosMap, err := c.infoDataToInfoMapBatch(reply.Infos)
  145. if err != nil {
  146. return "", nil, 0, err
  147. }
  148. return reply.Statement, infosMap, reply.TotalCount, nil
  149. }
  150. func (c *Client) CommonQuery(req *ports.CommonQueryRequest) (string, []map[string]any, int64, error) {
  151. whereJsonBytes, err := json.Marshal(req.Where)
  152. if err != nil {
  153. return "", nil, 0, err
  154. }
  155. orJsonBytes, err := json.Marshal(req.Or)
  156. if err != nil {
  157. return "", nil, 0, err
  158. }
  159. havingJsonBytes, err := json.Marshal(req.Having)
  160. if err != nil {
  161. return "", nil, 0, err
  162. }
  163. reply, err := c.queryServiceClient.CommonQuery(context.Background(), &request.CommonQueryRequest{
  164. DatabaseID: req.DatabaseID,
  165. TablePrefixWithSchema: req.TablePrefixWithSchema,
  166. Version: req.Version,
  167. Select: req.Select,
  168. Where: whereJsonBytes,
  169. OrderBy: req.OrderBy,
  170. Or: orJsonBytes,
  171. GroupBy: req.GroupBy,
  172. Join: req.Join,
  173. Having: havingJsonBytes,
  174. PageNo: req.PageNo,
  175. PageSize: req.PageSize,
  176. })
  177. if err != nil {
  178. return "", nil, 0, err
  179. }
  180. infosMap, err := c.infoDataToInfoMapBatch(reply.Infos)
  181. if err != nil {
  182. return "", nil, 0, err
  183. }
  184. return reply.Statement, infosMap, reply.TotalCount, nil
  185. }
  186. func (c *Client) QueryByKeys(req *ports.QueryByKeysRequest) (string, map[string]any, error) {
  187. reply, err := c.queryServiceClient.QueryByKeys(context.Background(), &request.QueryByKeysRequest{
  188. DatabaseID: req.DatabaseID,
  189. TablePrefixWithSchema: req.TablePrefixWithSchema,
  190. Version: req.Version,
  191. Select: req.Select,
  192. KeyValues: req.KeyValues,
  193. })
  194. if err != nil {
  195. return "", nil, err
  196. }
  197. infoMap, err := c.infoDataToInfoMap(reply.Info)
  198. if err != nil {
  199. return "", nil, err
  200. }
  201. return reply.Statement, infoMap, nil
  202. }
  203. func (c *Client) CountWhere(req *ports.CountWhereRequest) (string, int64, error) {
  204. whereJsonBytes, err := json.Marshal(req.Where)
  205. if err != nil {
  206. return "", 0, err
  207. }
  208. reply, err := c.queryServiceClient.CountWhere(context.Background(), &request.CountWhereRequest{
  209. DatabaseID: req.DatabaseID,
  210. TablePrefixWithSchema: req.TablePrefixWithSchema,
  211. Version: req.Version,
  212. Where: whereJsonBytes,
  213. })
  214. if err != nil {
  215. return "", 0, err
  216. }
  217. return reply.Statement, reply.Count, nil
  218. }
  219. func (c *Client) CommonCount(req *ports.CommonCountRequest) (string, int64, error) {
  220. whereJsonBytes, err := json.Marshal(req.Where)
  221. if err != nil {
  222. return "", 0, err
  223. }
  224. orJsonBytes, err := json.Marshal(req.Or)
  225. if err != nil {
  226. return "", 0, err
  227. }
  228. havingJsonBytes, err := json.Marshal(req.Having)
  229. if err != nil {
  230. return "", 0, err
  231. }
  232. reply, err := c.queryServiceClient.CommonCount(context.Background(), &request.CommonCountRequest{
  233. DatabaseID: req.DatabaseID,
  234. TablePrefixWithSchema: req.TablePrefixWithSchema,
  235. Version: req.Version,
  236. Where: whereJsonBytes,
  237. Or: orJsonBytes,
  238. GroupBy: req.GroupBy,
  239. Join: req.Join,
  240. Having: havingJsonBytes,
  241. })
  242. if err != nil {
  243. return "", 0, err
  244. }
  245. return reply.Statement, reply.Count, nil
  246. }
  247. func (c *Client) formTableRow(tableRow map[string]any) (*request.TableRow, error) {
  248. if tableRow == nil || len(tableRow) == 0 {
  249. return &request.TableRow{Columns: make([]*request.Column, 0)}, nil
  250. }
  251. columns := make([]*request.Column, 0)
  252. for columnName, value := range tableRow {
  253. valueType := reflect.TypeOf(value)
  254. if valueType.Kind() == reflect.Ptr {
  255. reflectValue := reflect.ValueOf(value).Elem()
  256. valueType = reflectValue.Type()
  257. value = reflectValue.Interface()
  258. }
  259. typedValue := new(request.ColumnValue)
  260. typedValue.Kind = int32(valueType.Kind())
  261. typedValue.Type = valueType.Name()
  262. switch valueType.Name() {
  263. case "Time":
  264. timeObj := value.(time.Time)
  265. typedValue.TypedValue = &request.ColumnValue_Uint64Value{Uint64Value: uint64(timeObj.UnixNano())}
  266. case "string":
  267. typedValue.TypedValue = &request.ColumnValue_StringValue{StringValue: value.(string)}
  268. case "bool":
  269. typedValue.TypedValue = &request.ColumnValue_BoolValue{BoolValue: value.(bool)}
  270. case "int":
  271. typedValue.TypedValue = &request.ColumnValue_Uint32Value{Uint32Value: uint32(value.(int))}
  272. case "int32":
  273. typedValue.TypedValue = &request.ColumnValue_Uint32Value{Uint32Value: uint32(value.(int32))}
  274. case "int64":
  275. typedValue.TypedValue = &request.ColumnValue_Uint64Value{Uint64Value: uint64(value.(int64))}
  276. case "uint32":
  277. typedValue.TypedValue = &request.ColumnValue_Uint32Value{Uint32Value: value.(uint32)}
  278. case "uint64":
  279. typedValue.TypedValue = &request.ColumnValue_Uint64Value{Uint64Value: value.(uint64)}
  280. case "float32":
  281. typedValue.TypedValue = &request.ColumnValue_Float64Value{Float64Value: float64(value.(float32))}
  282. case "float64":
  283. typedValue.TypedValue = &request.ColumnValue_Float64Value{Float64Value: value.(float64)}
  284. default:
  285. return nil, errors.New("不支持的数据类型" + valueType.Name())
  286. }
  287. columns = append(columns, &request.Column{
  288. Name: columnName,
  289. Value: typedValue,
  290. })
  291. }
  292. return &request.TableRow{Columns: columns}, nil
  293. }
  294. func (c *Client) infoDataToInfoMap(infoData *response.InfoData) (map[string]any, error) {
  295. retInfoMap := make(map[string]any)
  296. for _, column := range infoData.Columns {
  297. switch column.Value.Type {
  298. case "Time":
  299. timeObj := time.Unix(0, int64(column.Value.GetUint64Value()))
  300. retInfoMap[column.Name] = timeObj
  301. case "string":
  302. retInfoMap[column.Name] = column.Value.GetStringValue()
  303. case "bool":
  304. retInfoMap[column.Name] = column.Value.GetBoolValue()
  305. case "int":
  306. retInfoMap[column.Name] = int(column.Value.GetUint32Value())
  307. case "int32":
  308. retInfoMap[column.Name] = int(column.Value.GetUint32Value())
  309. case "int64":
  310. retInfoMap[column.Name] = int64(column.Value.GetUint64Value())
  311. case "uint32":
  312. retInfoMap[column.Name] = column.Value.GetUint32Value()
  313. case "uint64":
  314. retInfoMap[column.Name] = column.Value.GetUint64Value()
  315. case "float32":
  316. retInfoMap[column.Name] = float32(column.Value.GetFloat64Value())
  317. case "float64":
  318. retInfoMap[column.Name] = column.Value.GetFloat64Value()
  319. default:
  320. return nil, errors.New("不支持的数据类型" + column.Value.Type)
  321. }
  322. }
  323. return retInfoMap, nil
  324. }
  325. func (c *Client) infoDataToInfoMapBatch(infosData []*response.InfoData) ([]map[string]any, error) {
  326. retInfosDataMap := make([]map[string]any, 0)
  327. for _, infoData := range infosData {
  328. retInfoMap, err := c.infoDataToInfoMap(infoData)
  329. if err != nil {
  330. return nil, err
  331. }
  332. retInfosDataMap = append(retInfosDataMap, retInfoMap)
  333. }
  334. return retInfosDataMap, nil
  335. }