feat: check sever version

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## 0.15.0 [unreleased]
 
+### Features
+
+1. [#146](https://github.com/InfluxCommunity/influxdb3-python/pull/146): Add function to get InfluxDB version.
+
 ## 0.14.0 [2025-06-18]
 
 ### Features

influxdb_client_3/__init__.py

@@ -467,6 +467,24 @@ async def query_async(self, query: str, language: str = "sql", mode: str = "all"
         except ArrowException as e:
             raise InfluxDB3ClientQueryError(f"Error while executing query: {e}")
 
+    def get_server_version(self) -> str:
+        version = None
+        (resp_body, _, header) = self._client.api_client.call_api(
+            resource_path="/ping",
+            method="GET",
+            response_type=object
+        )
+
+        for key, value in header.items():
+            if key.lower() == "x-influxdb-version":
+                version = value
+                break
+
+        if version is None and isinstance(resp_body, dict):
+            version = resp_body['version'] if version is None else version
+
+        return version
+
     def close(self):
         """Close the client and clean up resources."""
         self._write_api.close()

tests/test_influxdb_client_3.py

@@ -1,12 +1,22 @@
+import re
 import unittest
 from unittest.mock import patch
 
+from pytest_httpserver import HTTPServer
+
 from influxdb_client_3 import InfluxDBClient3, WritePrecision, DefaultWriteOptions, Point, WriteOptions, WriteType
 from influxdb_client_3.exceptions import InfluxDB3ClientQueryError
+from influxdb_client_3.write_client.rest import ApiException
 from tests.util import asyncio_run
 from tests.util.mocks import ConstantFlightServer, ConstantData, ErrorFlightServer
 
 
+def http_server():
+    httpserver = HTTPServer()
+    httpserver.start()
+    return httpserver
+
+
 class TestInfluxDBClient3(unittest.TestCase):
 
     @patch('influxdb_client_3._InfluxDBClient')
@@ -22,6 +32,11 @@ def setUp(self, mock_query_api, mock_write_api, mock_influx_db_client):
             database="my_db",
             token="my_token"
         )
+        self.http_server = http_server()
+
+    def tearDown(self):
+        if self.http_server is not None:
+            self.http_server.stop()
 
     def test_init(self):
         self.assertEqual(self.client._org, "my_org")
@@ -226,6 +241,49 @@ async def test_async_query_with_arrow_error(self):
                 await c.query_async("SELECT * FROM my_data")
             self.assertIn("Error while executing query", str(err.exception))
 
+    def test_get_version_header_success(self):
+        server = self.http_server
+        server.expect_request(re.compile(".*")).respond_with_json(
+            headers={"X-Influxdb-Version": "1.8.2"},
+            response_json={"version": "3.0"}
+        )
+        version = InfluxDBClient3(
+            host=f'http://{server.host}:{server.port}', org="ORG", database="DB", token="TOKEN"
+        ).get_server_version()
+        assert version == "1.8.2"
+
+    def test_get_version_in_body_success(self):
+        server = self.http_server
+        server.expect_request('/ping').respond_with_json(
+            response_json={"version": "3.0"},
+        )
+        version = InfluxDBClient3(
+            host=f'http://{server.host}:{server.port}', org="ORG", database="DB", token="TOKEN"
+        ).get_server_version()
+        assert version == "3.0"
+
+    def test_get_version_empty(self):
+        server = self.http_server
+        server.expect_request("/ping").respond_with_data(
+            headers={"abc": "1.8.2"},
+        )
+
+        version = InfluxDBClient3(
+            host=f'http://{server.host}:{server.port}', org="ORG", database="DB", token="TOKEN"
+        ).get_server_version()
+        assert version is None
+
+    def test_get_version_fail(self):
+        server = self.http_server
+        server.expect_request("/ping").respond_with_json(
+            response_json={"error": "error"},
+            status=400
+        )
+        with self.assertRaises(ApiException):
+            InfluxDBClient3(
+                host=f'http://{server.host}:{server.port}', org="ORG", database="DB", token="TOKEN"
+            ).get_server_version()
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_influxdb_client_3_integration.py

@@ -274,3 +274,7 @@ async def test_verify_query_async(self):
         result_list = result.to_pylist()
         for item in data:
             assert lp_to_py_object(item) in result_list, f"original lp data \"{item}\" should be in result list"
+
+    def test_get_server_version(self):
+        version = self.client.get_server_version()
+        assert version is not None