feat: prepare to add warning into API call

Author: bednar

influxdb_client/client/_base.py

@@ -98,13 +98,6 @@ def __init__(self, url, token, debug=None, timeout=10_000, enable_gzip=False, or
         self.profilers = kwargs.get('profilers', None)
         pass
 
-    def _response_header(self, response, header_name='X-Influxdb-Version') -> str:
-        if response is not None and len(response) >= 3:
-            if header_name in response[2]:
-                return response[2][header_name]
-
-        return "unknown"
-
     @classmethod
     def _from_config_file(cls, config_file: str = "config.ini", debug=None, enable_gzip=False, **kwargs):
         config = configparser.ConfigParser()

influxdb_client/client/influxdb_client.py

@@ -407,7 +407,7 @@ def version(self) -> str:
 
         response = ping_service.get_ping_with_http_info(_return_http_data_only=False)
 
-        return self._response_header(response)
+        return ping_service.response_header(response)
 
     def build(self) -> str:
         """
@@ -417,9 +417,7 @@ def build(self) -> str:
         """
         ping_service = PingService(self.api_client)
 
-        response = ping_service.get_ping_with_http_info(_return_http_data_only=False)
-
-        return self._response_header(response, header_name='X-Influxdb-Build')
+        return ping_service.build_type()
 
     def ready(self) -> Ready:
         """

influxdb_client/client/influxdb_client_async.py

@@ -251,7 +251,7 @@ async def version(self) -> str:
         ping_service = PingService(self.api_client)
 
         response = await ping_service.get_ping_async(_return_http_data_only=False)
-        return self._response_header(response)
+        return ping_service.response_header(response)
 
     async def build(self) -> str:
         """
@@ -261,8 +261,7 @@ async def build(self) -> str:
         """
         ping_service = PingService(self.api_client)
 
-        response = await ping_service.get_ping_async(_return_http_data_only=False)
-        return self._response_header(response, header_name='X-Influxdb-Build')
+        return await ping_service.build_type_async()
 
     def query_api(self, query_options: QueryOptions = QueryOptions()) -> QueryApiAsync:
         """

influxdb_client/service/_base_service.py

@@ -2,6 +2,13 @@
 
 # noinspection PyMethodMayBeStatic
 class _BaseService(object):
+
+    def __init__(self, api_client=None):
+        """Init common services operation."""
+        if api_client is None:
+            raise ValueError("Invalid value for `api_client`, must be defined.")
+        self.api_client = api_client
+
     def _check_operation_params(self, operation_id, supported_params, local_params):
         supported_params.append('async_req')
         supported_params.append('_return_http_data_only')
@@ -16,3 +23,40 @@ def _check_operation_params(self, operation_id, supported_params, local_params):
                 )
             local_params[key] = val
         del local_params['kwargs']
+
+    def _is_cloud_instance(self) -> bool:
+        return False
+
+    async def _is_cloud_instance_async(self) -> bool:
+        return False
+
+    def build_type(self) -> str:
+        """
+        Return the build type of the connected InfluxDB Server.
+
+        :return: The type of InfluxDB build.
+        """
+        from influxdb_client import PingService
+        ping_service = PingService(self.api_client)
+
+        response = ping_service.get_ping_with_http_info(_return_http_data_only=False)
+        return self.response_header(response, header_name='X-Influxdb-Build')
+
+    async def build_type_async(self) -> str:
+        """
+        Return the build type of the connected InfluxDB Server.
+
+        :return: The type of InfluxDB build.
+        """
+        from influxdb_client import PingService
+        ping_service = PingService(self.api_client)
+
+        response = await ping_service.get_ping_async(_return_http_data_only=False)
+        return self.response_header(response, header_name='X-Influxdb-Build')
+
+    def response_header(self, response, header_name='X-Influxdb-Version') -> str:
+        if response is not None and len(response) >= 3:
+            if header_name in response[2]:
+                return response[2][header_name]
+
+        return "unknown"

influxdb_client/service/bucket_schemas_service.py

@@ -27,13 +27,7 @@ class BucketSchemasService(_BaseService):
 
     def __init__(self, api_client=None):  # noqa: E501,D401,D403
         """BucketSchemasService - a operation defined in OpenAPI."""
-        if api_client is None:
-            raise ValueError("Invalid value for `api_client`, must be defined.")
-        self.api_client = api_client
-
-        from influxdb_client.client.warnings import CloudOnlyWarning
-        CloudOnlyWarning.print_warning('BucketSchemasService',
-                                       'https://docs.influxdata.com/influxdb/cloud/organizations/buckets/bucket-schema/')  # noqa: E501
+        super().__init__(api_client)
 
     def create_measurement_schema(self, bucket_id, measurement_schema_create_request, **kwargs):  # noqa: E501,D401,D403
         """Create a measurement schema for a bucket.

influxdb_client/service/invokable_scripts_service.py

@@ -33,7 +33,8 @@ def __init__(self, api_client=None):  # noqa: E501,D401,D403
 
         from influxdb_client.client.warnings import CloudOnlyWarning
         CloudOnlyWarning.print_warning('InvokableScriptsService',
-                                       'https://docs.influxdata.com/influxdb/cloud/api-guide/api-invokable-scripts/')  # noqa: E501
+                                       'https://docs.influxdata.com/influxdb/cloud/api-guide/api-invokable-scripts/',
+                                       api_client)  # noqa: E501
 
     def delete_scripts_id(self, script_id, **kwargs):  # noqa: E501,D401,D403
         """Delete a script.