add query_post, get_test_default_client method

Signed-off-by: ZhangJian He <[email protected]>
Co-authored-by: heiliuchao <[email protected]>

opengemini_client/client_impl.py

@@ -146,7 +146,16 @@ def query(self, query: Query) -> QueryResult:
         resp = self.request(method='GET', server_url=server_url, url_path=UrlConst.QUERY, params=params)
         if resp.status_code == HTTPStatus.OK:
             return resolve_query_body(resp)
-        raise HTTPError(f"Query error: {resp.status_code}, Response: {resp.text}")
+        raise HTTPError(f"Query error_code: {resp.status_code}, error_msg: {resp.text}")
+
+    def _query_post(self, query: Query) -> QueryResult:
+        server_url = self.get_server_url()
+        params = {'db': query.database, 'q': query.command, 'rp': query.retention_policy}
+
+        resp = self.request(method='POST', server_url=server_url, url_path=UrlConst.QUERY, params=params)
+        if resp.status_code == HTTPStatus.OK:
+            return resolve_query_body(resp)
+        raise HTTPError(f"Query error_code: {resp.status_code}, error_msg: {resp.text}")
 
     def write_batch_points(self, database: str, batch_points: BatchPoints):
         return

opengemini_client/models.py

@@ -42,6 +42,18 @@ class Config:
     tls_config: ssl.SSLContext = None
 
 
+@dataclass
+class RetentionPolicy:
+    name: str
+    duration: str
+    shard_group_duration: str
+    hot_duration: str
+    warm_duration: str
+    index_duration: str
+    replica_num: int
+    is_default: bool
+
+
 @dataclass
 class RpConfig:
     name: str

opengemini_client/ping_test.py

@@ -3,21 +3,15 @@
 
 import requests
 
-from opengemini_client import client_impl
+from opengemini_client import client_impl, test_utils
 from opengemini_client import models
 
 
 class PingTest(unittest.TestCase):
 
     # noinspection PyMethodMayBeStatic
     def test_ping_success(self):
-        cfg = models.Config(address=[models.Address(host='127.0.0.1', port=8086)],
-                            auth_config=models.AuthConfig(auth_type=models.AuthType(0)),
-                            batch_config=models.BatchConfig(batch_size=10, batch_interval=10),
-                            timeout=datetime.timedelta(seconds=10), connection_timeout=datetime.timedelta(seconds=10),
-                            gzip_enabled=False, tls_enabled=False
-                            )
-        with client_impl.OpenGeminiDBClient(config=cfg) as cli:
+        with test_utils.get_test_default_client() as cli:
             cli.ping(0)
 
     def test_ping_error(self):

opengemini_client/test_utils.py

@@ -0,0 +1,16 @@
+import datetime
+
+from opengemini_client import client_impl
+from opengemini_client import models
+
+
+def get_test_default_client():
+    cfg = models.Config(address=[models.Address(host='127.0.0.1', port=8086)],
+                        auth_config=models.AuthConfig(auth_type=models.AuthType(0)),
+                        batch_config=models.BatchConfig(batch_size=10, batch_interval=10),
+                        timeout=datetime.timedelta(seconds=10),
+                        connection_timeout=datetime.timedelta(seconds=10),
+                        gzip_enabled=False, tls_enabled=False
+                        )
+    cli = client_impl.OpenGeminiDBClient(cfg)
+    return cli