diff --git a/CHANGELOG.md b/CHANGELOG.md index 7431a1a..1517d6d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,7 @@ ### Breaking Changes -1. [#217](https://github.com/InfluxCommunity/influxdb3-python/pull/217): Makes the writing API simpler and more consistent with other v3 clients. +1. [#217](https://github.com/InfluxCommunity/influxdb3-python/pull/217), [#237](https://github.com/InfluxCommunity/influxdb3-python/pull/237): Makes the writing API simpler and more consistent with other v3 clients. - Removes unused `InfluxLoggingHandler` class. - `WriteApi` now handles all functions to write to InfluxDB. - The following internal classes have been removed: @@ -21,6 +21,7 @@ - The following internal classes have been refactored: - `write_client._sync.RestClient` - this class is now responsible for low-level handling of transport requests. End users should not need to use it directly. - `write_client.client.WriteApi` - all internal settings needed for writing are now encapsulated within this Api. + - Further simplifies the `WriteApi` request path by constructing v2/v3 requests directly through `RestClient`, while preserving existing write behavior. - Refactors Multiprocessing helper class. - __Migration Guidance__ - `InfluxDBClient3` constructor now has additional parameters for configuring the internal `WriteApi` and `RestClient`. diff --git a/influxdb_client_3/write_client/client/write_api.py b/influxdb_client_3/write_client/client/write_api.py index b794a32..09ceec0 100644 --- a/influxdb_client_3/write_client/client/write_api.py +++ b/influxdb_client_3/write_client/client/write_api.py @@ -63,6 +63,24 @@ 'tag_order', } +REQUEST_BUILD_KWARGS = { + 'accept', + 'content_type', + 'content_length', + 'content_encoding', + 'org_id', +} + +POST_WRITE_KWARGS = REQUEST_BUILD_KWARGS.union({ + 'precision', + 'no_sync', + 'accept_partial', + 'use_v2_api', + 'async_req', + '_request_timeout', + 'urlopen_kw', +}) + logger = logging.getLogger('influxdb_client_3.write_client.client.write_api') try: @@ -74,7 +92,7 @@ class WriteType(Enum): - """Configuration which type of writes will client use.""" + """Defines which type of writes the client will use.""" batching = 1 asynchronous = 2 @@ -309,7 +327,7 @@ def __init__(self, :param pool_threads: Number of threads used for connection pools. :param default_header: Default HTTP headers to include in every request. :param rest_client: An instance of a RestClient for internal HTTP communication. - :param write_options: Configuration options for writing data (e.g., synchronous + :param write_options: Options for writing data (e.g., synchronous or batching modes). :param point_settings: Default settings to apply to all points being written. :param kwargs: Additional keyword arguments that may include: @@ -436,90 +454,41 @@ def write_payload(payload): async def post_write_async(self, org, bucket, body, **kwargs): # noqa: E501,D401,D403 """ - Writes data to a bucket. Use this endpoint to send data in [line protocol](https://docs.influxdata.com/influxdb/latest/reference/syntax/line-protocol/) format to InfluxDB. InfluxDB Cloud - Does the following when you send a writing request: - 1. Validates the request and queues the writing. - 2. If queued, responds with _success_ (HTTP `2xx` status code); _error_ otherwise. - 3. Handles to delete it asynchronously and reaches eventual consistency. To ensure that InfluxDB Cloud handles writes and deletes in the order you request them, - wait for a success response (HTTP `2xx` status code) before you send the next request. - Because writes and deletes are asynchronous, your change might not yet be readable when you receive the response. - #### InfluxDB OSS - Validates the request and handles the writing synchronously. - If all points were written successfully, responds with HTTP `2xx` status code; - otherwise, returns the first line that failed.\n - #### Required permissions - `write-buckets` or `write-bucket BUCKET_ID`. *`BUCKET_ID`* is the ID of the destination bucket.\n - #### Rate limits (with InfluxDB Cloud) `write` rate limits apply. - For more information, see [limits and adjustable quotas](https://docs.influxdata.com/influxdb/cloud/account-management/limits/).\n - #### Related guides - [Write data with the InfluxDB API](https://docs.influxdata.com/influxdb/latest/write-data/developer-tools/api) - [Optimize writes to InfluxDB](https://docs.influxdata.com/influxdb/latest/write-data/best-practices/optimize-writes/) - [Troubleshoot issues writing data](https://docs.influxdata.com/influxdb/latest/write-data/troubleshoot/)\n - This method makes an asynchronous HTTP request. - - :param str org: An organization name. (required) - :param str bucket: A bucket name or ID. InfluxDB writes all points in the batch to the specified bucket. (required) - :param str body: In the request body, provide data in [line protocol format](https://docs.influxdata.com/influxdb/latest/reference/syntax/line-protocol/). - To send compressed data, do the following: - 1. Use [GZIP](https://www.gzip.org/) to compress the line protocol data. - 2. In your request, send the compressed data and the `Content-Encoding: gzip` header. - #### Related guides - [Best practices for optimizing writes](https://docs.influxdata.com/influxdb/latest/write-data/best-practices/optimize-writes/) (required) - :param str org_id: An organization ID.#### InfluxDB Cloud - Doesn't use the `org` parameter or `orgID` parameter. - Writes data to the bucket in the organization associated with the authorization (API token). - #### InfluxDB OSS - Requires either the `org` parameter or the `orgID` parameter. - If you pass both `orgID` and `org`, they must both be valid. - Writes data to the bucket in the specified organization. - :return: None - If the method is called asynchronously, - returns the request thread. + Write line protocol asynchronously using the configured write endpoint. + + :param str org: Organization name. + :param str bucket: Bucket or database name. + :param str body: Line protocol payload. + :return: The HTTP response. """ # noqa: E501 - local_var_params, path, path_params, query_params, header_params, body_params = \ - self._post_write_prepare(org, bucket, body, self.default_header, **kwargs) # noqa: E501 - use_v2_api = local_var_params['use_v2_api'] + if body is None: + raise ValueError("Missing the required parameter 'body' when calling `post_write_async`") + http_kwargs = {k: v for k, v in kwargs.items() if k not in SERIALIZER_KWARGS} + self._validate_post_write_kwargs(http_kwargs) + use_v2_api = http_kwargs.get('use_v2_api', DEFAULT_WRITE_USE_V2_API) + no_sync = http_kwargs.get('no_sync', DEFAULT_WRITE_NO_SYNC) + accept_partial = http_kwargs.get('accept_partial', DEFAULT_WRITE_ACCEPT_PARTIAL) + precision = http_kwargs.get('precision') + request_kwargs = { + k: v for k, v in http_kwargs.items() + if k in REQUEST_BUILD_KWARGS + } + request = self._build_write_request( + org, bucket, precision, no_sync, accept_partial, use_v2_api, **request_kwargs) try: loop = asyncio.get_running_loop() return await loop.run_in_executor( None, - self._call_api, - path, - 'POST', - query_params, - header_params, + self._request, + request, body, - local_var_params.get('_request_timeout'), - kwargs.get('urlopen_kw', None), + http_kwargs.get('_request_timeout'), + http_kwargs.get('urlopen_kw', None), ) except ApiException as e: raise self._translate_write_exception(e, use_v2_api) - def call_api(self, resource_path, method, - query_params=None, header_params=None, - body=None, async_req=None, _request_timeout=None, urlopen_kw=None): - """Make the HTTP request (synchronous) and Return deserialized data. - - To make an async_req request, set the async_req parameter. - - :param resource_path: Path to method endpoint. - :param method: Method to call. - :param query_params: Query parameters in the url. - :param header_params: Header parameters to be placed in the request header. - :param body: Request body. - :param async_req bool: execute request asynchronously - :param _request_timeout: timeout setting for this request. If one - number provided, it will be total request - timeout. It can also be a pair (tuple) of - (connection, read) timeouts. - :param urlopen_kw: Additional parameters are passed to - :meth:`urllib3.request.RequestMethods.request` - :return: - If async_req parameter is True, - the request will be called asynchronously. - The method will return the request thread. - If parameter async_req is False or missing, - then the method will return the response directly. - """ - if not async_req: - return self._call_api(resource_path, method, - query_params, header_params, - body, _request_timeout, urlopen_kw) - - else: - thread = self.pool.apply_async(self._call_api, (resource_path, - method, query_params, - header_params, body, _request_timeout, urlopen_kw)) - return thread - def flush(self): """ Flush any buffered writes to InfluxDB without closing the client. @@ -690,27 +659,23 @@ def _retry_callback_delegate(exception): return _BatchResponse(data=batch_item) def _post_write(self, _async_req, bucket, org, body, precision, no_sync, accept_partial, use_v2_api, **kwargs): - # Filter out serializer-specific kwargs before passing to _post_write + if body is None: + raise ValueError("Missing the required parameter 'body' when calling `_post_write`") + # Filter out serializer-specific kwargs before building the HTTP request. http_kwargs = {k: v for k, v in kwargs.items() if k not in SERIALIZER_KWARGS} - http_kwargs['precision'] = precision - http_kwargs['no_sync'] = no_sync - http_kwargs['accept_partial'] = accept_partial - http_kwargs['use_v2_api'] = use_v2_api - - local_var_params, path, path_params, query_params, header_params, body_params = \ - self._post_write_prepare(org, bucket, body, self.default_header, **http_kwargs) # noqa: E501 - - use_v2_api = local_var_params['use_v2_api'] + self._validate_post_write_kwargs(http_kwargs) + request_kwargs = {k: v for k, v in http_kwargs.items() if k in REQUEST_BUILD_KWARGS} + request = self._build_write_request( + org, bucket, precision, no_sync, accept_partial, use_v2_api, **request_kwargs) try: - result = self.call_api( - path, 'POST', - query_params, - header_params, - body=body_params, - async_req=_async_req, - _request_timeout=local_var_params.get('_request_timeout'), - urlopen_kw=http_kwargs.get('urlopen_kw', None)) if _async_req: + result = self.pool.apply_async( + self._request, + (request, body), + { + '_request_timeout': http_kwargs.get('_request_timeout'), + 'urlopen_kw': http_kwargs.get('urlopen_kw', None), + }) original_get = result.get def translated_get(timeout=None): @@ -720,31 +685,76 @@ def translated_get(timeout=None): raise self._translate_write_exception(e, use_v2_api) result.get = translated_get - return result + return result + + return self._request( + request, + body, + http_kwargs.get('_request_timeout'), + http_kwargs.get('urlopen_kw', None)) except ApiException as e: raise self._translate_write_exception(e, use_v2_api) - def _call_api( - self, resource_path, method, - query_params=None, header_params=None, body=None, - _request_timeout=None, urlopen_kw=None): + def _build_write_request(self, org, bucket, precision, no_sync, accept_partial, use_v2_api, **kwargs): + if org is None: + raise ValueError("Missing the required parameter `org` when calling `_build_write_request`") + if bucket is None: + raise ValueError("Missing the required parameter `bucket` when calling `_build_write_request`") + + query_params = [('org', org)] + if kwargs.get('org_id') is not None: + query_params.append(('orgID', kwargs['org_id'])) + + if use_v2_api: + path = '/api/v2/write' + query_params.append(('bucket', bucket)) + if precision is not None: + query_params.append(('precision', WritePrecisionConverter.to_v2_api_string(precision))) + else: + path = '/api/v3/write_lp' + query_params.append(('db', bucket)) + if precision is not None: + query_params.append(('precision', WritePrecisionConverter.to_v3_api_string(precision))) + if no_sync: + query_params.append(('no_sync', 'true')) + if accept_partial is False: + query_params.append(('accept_partial', 'false')) + + header_params = dict(self.default_header) if self.default_header is not None else {} + if kwargs.get('accept') is not None: + header_params['Accept'] = kwargs['accept'] + else: + header_params.setdefault('Accept', 'application/json') + if kwargs.get('content_type') is not None: + header_params['Content-Type'] = kwargs['content_type'] + else: + header_params.setdefault('Content-Type', 'text/plain; charset=utf-8') + if kwargs.get('content_length') is not None: + header_params['Content-Length'] = kwargs['content_length'] + if kwargs.get('content_encoding') is not None: + header_params['Content-Encoding'] = kwargs['content_encoding'] + + return path, query_params, header_params + + @staticmethod + def _validate_post_write_kwargs(kwargs): + for key in kwargs: + if key not in POST_WRITE_KWARGS: + raise TypeError( + f"Got an unexpected keyword argument '{key}' to method _post_write" + ) - # body + def _request(self, request, body, _request_timeout=None, urlopen_kw=None): + resource_path, query_params, header_params = request should_gzip = False if body: should_gzip = self._should_gzip(body, self.enable_gzip, self.gzip_threshold) - body = self._sanitize_for_serialization(body) - body = self._update_request_body(resource_path, body, should_gzip) - - # header parameters - header_params = header_params or {} - self._update_request_header_params(resource_path, header_params, should_gzip) - if header_params: - header_params = self._sanitize_for_serialization(header_params) - - # query parameters - if query_params: - query_params = self._sanitize_for_serialization(query_params) + if should_gzip: + import gzip + body = gzip.compress(body if isinstance(body, bytes) else bytes(body, _UTF_8_encoding)) + header_params = dict(header_params) + header_params['Content-Encoding'] = 'gzip' + header_params['Accept-Encoding'] = 'identity' urlopen_kw = urlopen_kw or {} @@ -760,7 +770,7 @@ def _call_api( # perform request and return response response_data = self.rest_client.request( - method=method, + method='POST', path=resource_path, query_params=query_params, headers=header_params, @@ -773,159 +783,6 @@ def _call_api( return response_data - def _post_write_prepare(self, org, bucket, body, default_header, **kwargs): # noqa: E501,D401,D403 - local_var_params = dict(locals()) - - all_params = ['org', 'bucket', 'body', 'content_encoding', 'content_type', 'content_length', - 'accept', 'org_id', 'precision', 'no_sync', 'accept_partial', 'use_v2_api'] # noqa: E501 - self._check_operation_params('_post_write', all_params, local_var_params) - local_var_params.setdefault('use_v2_api', DEFAULT_WRITE_USE_V2_API) - local_var_params.setdefault('no_sync', DEFAULT_WRITE_NO_SYNC) - local_var_params.setdefault('accept_partial', DEFAULT_WRITE_ACCEPT_PARTIAL) - # verify the required parameter 'org' is set - if ('org' not in local_var_params or - local_var_params['org'] is None): - raise ValueError("Missing the required parameter `org` when calling `_post_write`") # noqa: E501 - # verify the required parameter 'bucket' is set - if ('bucket' not in local_var_params or - local_var_params['bucket'] is None): - raise ValueError("Missing the required parameter `bucket` when calling `_post_write`") # noqa: E501 - # verify the required parameter 'body' is set - if ('body' not in local_var_params or - local_var_params['body'] is None): - raise ValueError("Missing the required parameter `body` when calling `_post_write`") # noqa: E501 - - path_params = {} - query_params = [] - - use_v2_api = local_var_params['use_v2_api'] - no_sync = local_var_params['no_sync'] - accept_partial = local_var_params['accept_partial'] - if 'org' in local_var_params: - query_params.append(('org', local_var_params['org'])) # noqa: E501 - if 'org_id' in local_var_params: - query_params.append(('orgID', local_var_params['org_id'])) # noqa: E501 - if 'bucket' in local_var_params: - query_params.append(('bucket' if use_v2_api else 'db', local_var_params['bucket'])) # noqa: E501 - - if use_v2_api: - path = '/api/v2/write' - if 'precision' in local_var_params: - precision = local_var_params['precision'] - query_params.append(('precision', WritePrecisionConverter.to_v2_api_string(precision))) # noqa: E501 - else: - path = '/api/v3/write_lp' - if 'precision' in local_var_params: - precision = local_var_params['precision'] - query_params.append(('precision', WritePrecisionConverter.to_v3_api_string(precision))) # noqa: E501 - if no_sync: - query_params.append(('no_sync', 'true')) - if accept_partial is False: - query_params.append(('accept_partial', 'false')) - - header_params = dict(default_header) if default_header is not None else {} - if local_var_params.get('accept') is not None: - header_params['Accept'] = local_var_params['accept'] - else: - header_params.setdefault('Accept', 'application/json') - if local_var_params.get('content_type') is not None: - header_params['Content-Type'] = local_var_params['content_type'] - else: - header_params.setdefault('Content-Type', 'text/plain; charset=utf-8') - if local_var_params.get('content_length') is not None: - header_params['Content-Length'] = local_var_params['content_length'] - - if local_var_params.get('content_encoding') is not None: - header_params['Content-Encoding'] = local_var_params['content_encoding'] # noqa: E501 - - body_params = None - if 'body' in local_var_params: - body_params = local_var_params['body'] - - return local_var_params, path, path_params, query_params, header_params, body_params - - def _check_operation_params(self, operation_id, supported_params, local_params): - supported_params.append('async_req') - supported_params.append('_request_timeout') - supported_params.append('urlopen_kw') - for key, val in local_params['kwargs'].items(): - if key not in supported_params: - raise TypeError( - f"Got an unexpected keyword argument '{key}'" - f" to method {operation_id}" - ) - local_params[key] = val - del local_params['kwargs'] - - def _update_request_header_params(self, path: str, params: dict, should_gzip: bool = False): - if should_gzip: - # GZIP Request - if path == '/api/v2/write' or path == '/api/v3/write_lp': - params["Content-Encoding"] = "gzip" - params["Accept-Encoding"] = "identity" - pass - # GZIP Response - if path == '/api/v2/query': - # params["Content-Encoding"] = "gzip" - params["Accept-Encoding"] = "gzip" - pass - pass - pass - - def _update_request_body(self, path: str, body, should_gzip: bool = False): - _body = body - if should_gzip: - # GZIP Request - if path == '/api/v2/write' or path == '/api/v3/write_lp': - import gzip - if isinstance(_body, bytes): - return gzip.compress(data=_body) - else: - return gzip.compress(bytes(_body, _UTF_8_encoding)) - - return _body - - def _sanitize_for_serialization(self, obj): - """Build a JSON POST object. - - If obj is None, return None.\n - If obj is str, int, long, float, bool, return directly.\n - If obj is datetime.datetime, datetime.date converts to string in iso8601 format.\n - If obj is a list, sanitize each element in the list.\n - If obj is dict, return the dict.\n - If obj is an OpenAPI model, return the properties dict.\n - - :param obj: The data to serialize. - :return: The serialized form of data. - """ - if obj is None: - return None - elif isinstance(obj, self.PRIMITIVE_TYPES): - return obj - elif isinstance(obj, list): - return [self._sanitize_for_serialization(sub_obj) - for sub_obj in obj] - elif isinstance(obj, tuple): - return tuple(self._sanitize_for_serialization(sub_obj) - for sub_obj in obj) - elif isinstance(obj, (datetime.datetime, datetime.date)): - return obj.isoformat() - - if isinstance(obj, dict): - obj_dict = obj - else: - # Convert model obj to dict except - # attributes `openapi_types`, `attribute_map` - # and attributes which value is not None. - # Convert attribute name to json key in - # model definition for request. - obj_dict = {obj.attribute_map[attr]: getattr(obj, attr) - for attr, _ in obj.openapi_types.items() - if getattr(obj, attr) is not None} - - return {key: self._sanitize_for_serialization(val) - for key, val in obj_dict.items()} - def _translate_write_exception(self, exc, use_v2_api): if use_v2_api and exc.status == HTTPStatus.METHOD_NOT_ALLOWED: message = ("Server doesn't support the V2 API endpoint (/api/v2/write). " diff --git a/tests/test_polars.py b/tests/test_polars.py index da8bad2..7b14cf6 100644 --- a/tests/test_polars.py +++ b/tests/test_polars.py @@ -164,7 +164,7 @@ def test_write_polars(self): "temperature": [22.4, 21.8], }) - self.client._write_api.call_api = mock.Mock() + self.client._write_api.rest_client.request = mock.Mock() self.client.write( database="database", @@ -173,7 +173,7 @@ def test_write_polars(self): data_frame_timestamp_column="time", ) - actual = self.client._write_api.call_api.call_args.kwargs['body'] + actual = self.client._write_api.rest_client.request.call_args.kwargs['body'] self.assertEqual(b'measurement temperature=22.4 1722470400000000000\n' b'measurement temperature=21.8 1722474000000000000', actual) @@ -193,7 +193,7 @@ def test_write_polars_batching(self): ) self.client._write_api._write_options = WriteOptions(batch_size=2) - self.client._write_api._call_api = mock.Mock() + self.client._write_api.rest_client.request = mock.Mock() self.client.write( database="database", @@ -203,8 +203,7 @@ def test_write_polars_batching(self): ) time.sleep(0.5) - args = self.client._write_api._call_api.call_args.args - body = args[4] - self.assertEqual(self.client._write_api._call_api.call_count, 1) + body = self.client._write_api.rest_client.request.call_args.kwargs['body'] + self.assertEqual(self.client._write_api.rest_client.request.call_count, 1) self.assertEqual(b'measurement temperature=22.4 1722470400000000000\nmeasurement ' b'temperature=21.8 1722474000000000000', body) diff --git a/tests/test_write_api.py b/tests/test_write_api.py index 9460ecc..2f3335a 100644 --- a/tests/test_write_api.py +++ b/tests/test_write_api.py @@ -55,6 +55,139 @@ def test_default_headers(self): self.assertEqual(f"{_package}/{VERSION}", write_api.default_header["User-Agent"]) self.assertEqual("Token my-token", write_api.default_header["Authorization"]) + def test_build_write_request_preserves_v2_and_v3_request_matrix(self): + client = InfluxDBClient3( + host='http://localhost:8181', + token='my-token', + database='my-bucket', + org='my-org' + ) + write_api = client._write_api + + cases = [ + ( + "v2 request", + True, + False, + False, + 'us', + '/api/v2/write', + [('org', 'TEST_ORG'), ('bucket', 'TEST_BUCKET'), ('precision', 'us')], + ), + ( + "v3 request", + False, + True, + False, + 'us', + '/api/v3/write_lp', + [('org', 'TEST_ORG'), ('db', 'TEST_BUCKET'), ('precision', 'microsecond'), + ('no_sync', 'true'), ('accept_partial', 'false')], + ), + ( + "v3 strict write request", + False, + False, + False, + 'ns', + '/api/v3/write_lp', + [('org', 'TEST_ORG'), ('db', 'TEST_BUCKET'), ('precision', 'nanosecond'), + ('accept_partial', 'false')], + ), + ] + + for name, use_v2_api, no_sync, accept_partial, precision, expected_path, expected_query in cases: + with self.subTest(name): + path, query_params, headers = write_api._build_write_request( + org='TEST_ORG', + bucket='TEST_BUCKET', + precision=precision, + no_sync=no_sync, + accept_partial=accept_partial, + use_v2_api=use_v2_api, + ) + self.assertEqual(expected_path, path) + self.assertEqual(expected_query, query_params) + self.assertEqual('application/json', headers['Accept']) + self.assertEqual('text/plain; charset=utf-8', headers['Content-Type']) + + def test_build_write_request_preserves_header_and_org_id_options(self): + client = InfluxDBClient3( + host='http://localhost:8181', + token='my-token', + database='my-bucket', + org='my-org' + ) + + path, query_params, headers = client._write_api._build_write_request( + org='TEST_ORG', + bucket='TEST_BUCKET', + precision='ns', + no_sync=False, + accept_partial=True, + use_v2_api=True, + org_id='ORG_ID', + accept='text/csv', + content_type='application/custom', + content_length=42, + content_encoding='gzip', + ) + + self.assertEqual('/api/v2/write', path) + self.assertEqual([ + ('org', 'TEST_ORG'), + ('orgID', 'ORG_ID'), + ('bucket', 'TEST_BUCKET'), + ('precision', 'ns'), + ], query_params) + self.assertEqual('text/csv', headers['Accept']) + self.assertEqual('application/custom', headers['Content-Type']) + self.assertEqual(42, headers['Content-Length']) + self.assertEqual('gzip', headers['Content-Encoding']) + + def test_build_write_request_requires_org_and_bucket(self): + client = InfluxDBClient3( + host='http://localhost:8181', + token='my-token', + database='my-bucket', + org='my-org' + ) + + with self.assertRaisesRegex(ValueError, r"required parameter `org`"): + client._write_api._build_write_request( + org=None, + bucket='TEST_BUCKET', + precision='ns', + no_sync=False, + accept_partial=True, + use_v2_api=True, + ) + + with self.assertRaisesRegex(ValueError, r"required parameter `bucket`"): + client._write_api._build_write_request( + org='TEST_ORG', + bucket=None, + precision='ns', + no_sync=False, + accept_partial=True, + use_v2_api=True, + ) + + _, query_params, _ = client._write_api._build_write_request( + org='TEST_ORG', + bucket='TEST_BUCKET', + precision='ns', + no_sync=False, + accept_partial=True, + use_v2_api=True, + org_id=None, + ) + self.assertEqual([ + ('org', 'TEST_ORG'), + ('bucket', 'TEST_BUCKET'), + ('precision', 'ns'), + ], query_params) + def test_api_error_cloud(self): response_body = '{"message": "parsing failed for write_lp endpoint"}' with self.assertRaises(InfluxDBError) as err: @@ -370,10 +503,9 @@ def test_post_write_async_translates_exceptions(self): org='my-org' ) write_api = client._write_api - write_api.call_api = mock.Mock() - thread = mock.Mock() - thread.get.side_effect = ApiException(http_resp=http_resp) - write_api.call_api.return_value = thread + write_api.rest_client.request = mock.Mock( + side_effect=ApiException(http_resp=http_resp) + ) result = write_api._post_write( org="TEST_ORG", bucket="TEST_BUCKET", @@ -393,6 +525,84 @@ def test_post_write_async_translates_exceptions(self): else: self.assertEqual(1, len(err.exception.line_errors)) + def test_post_write_async_requires_body(self): + client = InfluxDBClient3( + host='http://localhost:8181', + token='my-token', + database='my-bucket', + org='my-org', + ) + + async def run(): + await client._write_api.post_write_async( + "TEST_ORG", + "TEST_BUCKET", + None, + ) + + with self.assertRaisesRegex(ValueError, r"post_write_async"): + asyncio.run(run()) + + def test_post_write_requires_body(self): + client = InfluxDBClient3( + host='http://localhost:8181', + token='my-token', + database='my-bucket', + org='my-org', + ) + + with self.assertRaisesRegex(ValueError, r"_post_write"): + client._write_api._post_write( + False, + 'TEST_BUCKET', + 'TEST_ORG', + None, + 'ns', + False, + True, + True, + ) + + def test_post_write_rejects_unknown_keyword(self): + client = InfluxDBClient3( + host='http://localhost:8181', + token='my-token', + database='my-bucket', + org='my-org', + ) + + with self.assertRaisesRegex(TypeError, r"unexpected keyword argument 'unsupported'"): + client._write_api._post_write( + False, + 'TEST_BUCKET', + 'TEST_ORG', + 'home temp=96', + 'ns', + False, + True, + True, + unsupported=True, + ) + + def test_post_write_async_rejects_unknown_keyword(self): + client = InfluxDBClient3( + host='http://localhost:8181', + token='my-token', + database='my-bucket', + org='my-org', + ) + + async def run(): + await client._write_api.post_write_async( + 'TEST_ORG', + 'TEST_BUCKET', + 'home temp=96', + unsupported=True, + ) + + with self.assertRaisesRegex(TypeError, r"unexpected keyword argument 'unsupported'"): + asyncio.run(run()) + def test_post_write_async_translates_v3_unsupported(self): client = InfluxDBClient3( host='http://localhost:8181',