diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml index cb529fb1..b9784f91 100644 --- a/.github/workflows/pylint.yml +++ b/.github/workflows/pylint.yml @@ -7,7 +7,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ["3.8", "3.9", "3.10", "3.11"] + python-version: ["3.9", "3.10", "3.11"] steps: - uses: actions/checkout@v7 - name: Set up Python ${{ matrix.python-version }} diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ce37c61..76a9f63e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## 0.22.0 [unreleased] +### Bug Fixes + +1. [#239](https://github.com/InfluxCommunity/influxdb3-python/pull/239): + - Only throws `InfluxDBPartialWriteException` when: + - Error response status code is `400`. + - Error response format `{"error":"...","data":[{"error_message":"...","line_number":2,"original_line": "..."}]}` is returned with `data` must be an array. + - `accept_partial` is set to `true`. + - Write endpoint must be `api/v3/write_lp`. + ## 0.21.0 [2026-08-27] ### Bug Fixes diff --git a/influxdb_client_3/exceptions/exceptions.py b/influxdb_client_3/exceptions/exceptions.py index 2492fe3f..c6ab82fa 100644 --- a/influxdb_client_3/exceptions/exceptions.py +++ b/influxdb_client_3/exceptions/exceptions.py @@ -1,9 +1,8 @@ """Exceptions utils for InfluxDB.""" -import json import logging from dataclasses import dataclass -from typing import List, Optional, Tuple +from typing import List, Optional from urllib3 import HTTPResponse @@ -42,107 +41,6 @@ def __init__(self, error_message, *args, **kwargs): self.message = error_message -def _is_partial_write_error(error_message) -> bool: - if not isinstance(error_message, str) or not error_message: - return False - normalized = error_message.lower() - return ( - "partial write of line protocol occurred" in normalized or - "parsing failed for write_lp endpoint" in normalized - ) - - -def _parse_partial_write_data_item(item) -> Optional[Tuple[str, int, str]]: - if item is None: - return None - if not isinstance(item, dict): - raise ValueError("array item is not an object") - - error_message = item.get("error_message") - if not isinstance(error_message, str): - raise ValueError("error_message must be string") - if not error_message: - return None - - line_number_raw = item.get("line_number") - if line_number_raw is None: - line_number = 0 - elif isinstance(line_number_raw, int): - line_number = line_number_raw - else: - raise ValueError("line_number must be int") - - original_line_raw = item.get("original_line") - if original_line_raw is None: - original_line = "" - elif isinstance(original_line_raw, str): - original_line = original_line_raw - else: - raise ValueError("original_line must be string") - - return error_message, line_number, original_line - - -def _parse_typed_partial_write_array(data) -> Optional[List[Tuple[str, int, str]]]: - if not isinstance(data, list): - return None - line_errors: List[Tuple[str, int, str]] = [] - try: - for item in data: - parsed = _parse_partial_write_data_item(item) - if parsed is None: - continue - line_errors.append(parsed) - except ValueError: - return None - return line_errors if len(line_errors) > 0 else None - - -def _parse_typed_partial_write_object_or_none(data) -> Optional[Tuple[str, int, str]]: - try: - return _parse_partial_write_data_item(data) - except ValueError: - return None - - -def _format_partial_write_details(line_errors: List[Tuple[str, int, str]]) -> List[str]: - details: List[str] = [] - for error_message, line_number, original_line in line_errors: - if line_number != 0: - if original_line != "": - details.append(f"\tline {line_number}: {error_message} ({original_line})") - else: - details.append(f"\tline {line_number}: {error_message}") - elif error_message: - details.append(f"\t{error_message}") - return details - - -def _parse_partial_write_line_error_info(data) -> Tuple[List[Tuple[str, int, str]], List[str]]: - if data is None: - return [], [] - - typed_array = _parse_typed_partial_write_array(data) - if typed_array is not None: - return typed_array, _format_partial_write_details(typed_array) - - if isinstance(data, list): - details: List[str] = [] - for item in data: - if item is None: - continue - raw = json.dumps(item, separators=(',', ':')) - if raw and raw.lower() != "null": - details.append(raw) - return [], details - - typed_single = _parse_typed_partial_write_object_or_none(data) - if typed_single is not None: - return [typed_single], _format_partial_write_details([typed_single]) - - return [], [] - - # This error is for all write operations class InfluxDBError(InfluxDB3ClientError): """Raised when a server error occurs.""" @@ -151,7 +49,7 @@ def __init__(self, response: HTTPResponse = None, message: str = None): """Initialize the InfluxDBError handler.""" if response is not None: self.response = response - self.message = self._get_message(response) + self.message = message self.retry_after = response.getheader('Retry-After') else: self.response = None @@ -159,56 +57,6 @@ def __init__(self, response: HTTPResponse = None, message: str = None): self.retry_after = None super().__init__(self.message) - def _get_message(self, response): - if response.data: - def get(d, key): - if not key or d is None: - return d - if not isinstance(d, dict): - return None - return get(d.get(key[0]), key[1:]) - try: - node = json.loads(response.data) - if isinstance(node, dict): - # InfluxDB v3 error format: { "code": "...", "message": "..." } - code = node.get("code") - message = node.get("message") - if message: - return f"{code}: {message}" if code else message - # InfluxDB v3 write error format: - # { - # "error": "...", - # "data": [ { "error_message": "...", "line_number": 2, "original_line": "..." }, ... ] - # } - error_text = node.get("error") - if error_text and _is_partial_write_error(error_text): - _, details = _parse_partial_write_line_error_info(node.get("data")) - if details: - return error_text + ":\n" + "\n".join( - detail if detail.startswith("\t") else f"\t{detail}" - for detail in details - ) - return error_text - if error_text: - return error_text - for key in [['message'], ['data', 'error_message'], ['error']]: - value = get(node, key) - if value is not None: - return value - return response.data - except Exception as e: - logging.debug(f"Cannot parse error response to JSON: {response.data}, {e}") - return response.data - - # Header - for header_key in ["X-Platform-Error-Code", "X-Influx-Error", "X-InfluxDb-Error"]: - header_value = response.getheader(header_key) - if header_value is not None: - return header_value - - # Http Status - return response.reason - def getheaders(self): """Helper method to make response headers more accessible.""" return self.response.getheaders() @@ -216,40 +64,14 @@ def getheaders(self): @dataclass(frozen=True) class InfluxDBPartialWriteLineError: - line_number: int - error_message: str - original_line: str + line_number: Optional[int] + error_message: Optional[str] + original_line: Optional[str] class InfluxDBPartialWriteError(InfluxDBError): """Structured partial-write error with per-line failures.""" - def __init__(self, response: HTTPResponse, line_errors: List[InfluxDBPartialWriteLineError]): - super().__init__(response=response) + def __init__(self, response: HTTPResponse, message: str, line_errors: List[InfluxDBPartialWriteLineError]): + super().__init__(response=response, message=message) self.line_errors = line_errors - - @classmethod - def from_response(cls, response: HTTPResponse): - if response is None or not response.data: - return None - try: - node = json.loads(response.data) - except Exception: - return None - if not isinstance(node, dict): - return None - error_text = node.get("error") - if not _is_partial_write_error(error_text): - return None - parsed_line_errors, _ = _parse_partial_write_line_error_info(node.get("data")) - if not parsed_line_errors: - return None - line_errors = [ - InfluxDBPartialWriteLineError( - line_number=line_number, - error_message=error_message, - original_line=original_line, - ) - for error_message, line_number, original_line in parsed_line_errors - ] - return cls(response=response, line_errors=line_errors) diff --git a/influxdb_client_3/write_client/_sync/rest_client.py b/influxdb_client_3/write_client/_sync/rest_client.py index 354b3e7a..8e9bd691 100644 --- a/influxdb_client_3/write_client/_sync/rest_client.py +++ b/influxdb_client_3/write_client/_sync/rest_client.py @@ -175,7 +175,10 @@ def request(self, method, path, query_params=None, headers=None, raise ApiException(status=0, reason=msg) r = RESTResponse(r) - r.data = r.data.decode('utf8') + if r.data is not None and r.data != "": + r.data = r.data.decode('utf8') + else: + r.data = None if self.debug: RestClient.log_response(r.status) diff --git a/influxdb_client_3/write_client/client/write_api.py b/influxdb_client_3/write_client/client/write_api.py index b794a328..65b73442 100644 --- a/influxdb_client_3/write_client/client/write_api.py +++ b/influxdb_client_3/write_client/client/write_api.py @@ -1,21 +1,22 @@ """Collect and write time series data to InfluxDB Cloud or InfluxDB OSS.""" from __future__ import absolute_import - # TODO Remove after this program no longer supports Python 3.8.* from __future__ import annotations import asyncio import datetime +import json import logging import os import warnings from collections import defaultdict from enum import Enum from http import HTTPStatus +from json import JSONDecodeError from multiprocessing.pool import ThreadPool from random import random from time import sleep -from typing import Union, Any, Iterable, NamedTuple +from typing import Union, Any, Iterable, NamedTuple, List, Tuple, Optional import reactivex as rx import urllib3 @@ -23,21 +24,20 @@ from reactivex.scheduler import ThreadPoolScheduler from reactivex.subject import Subject -from influxdb_client_3.exceptions import InfluxDBPartialWriteError +from influxdb_client_3.exceptions import InfluxDBPartialWriteError, InfluxDBPartialWriteLineError from influxdb_client_3.write_client._sync.rest_client import RestClient -# from influxdb_client_3.write_client.client._base import _HAS_DATACLASS from influxdb_client_3.write_client.client.write.dataframe_serializer import DataframeSerializer from influxdb_client_3.write_client.client.write.point import Point, DEFAULT_WRITE_PRECISION, sanitize_tag_order from influxdb_client_3.write_client.client.write.retry import WritesRetry from influxdb_client_3.write_client.domain import WritePrecision from influxdb_client_3.write_client.domain.write_precision_converter import WritePrecisionConverter -from influxdb_client_3.write_client.write_exceptions import _UTF_8_encoding, ApiException from influxdb_client_3.write_client.write_defaults import ( DEFAULT_WRITE_ACCEPT_PARTIAL as _DEFAULT_WRITE_ACCEPT_PARTIAL, DEFAULT_WRITE_NO_SYNC as _DEFAULT_WRITE_NO_SYNC, DEFAULT_WRITE_TIMEOUT as _DEFAULT_WRITE_TIMEOUT, DEFAULT_WRITE_USE_V2_API as _DEFAULT_WRITE_USE_V2_API, ) +from influxdb_client_3.write_client.write_exceptions import _UTF_8_encoding, ApiException # Deprecated compatibility aliases. # New code should import these defaults from `influxdb_client_3.write_client.write_defaults`. @@ -481,7 +481,7 @@ async def post_write_async(self, org, bucket, body, **kwargs): # noqa: E501,D40 kwargs.get('urlopen_kw', None), ) except ApiException as e: - raise self._translate_write_exception(e, use_v2_api) + raise self._translate_write_exception(e, use_v2_api, local_var_params['accept_partial']) def call_api(self, resource_path, method, query_params=None, header_params=None, @@ -717,12 +717,11 @@ def translated_get(timeout=None): try: return original_get(timeout=timeout) except ApiException as e: - raise self._translate_write_exception(e, use_v2_api) - + raise self._translate_write_exception(e, use_v2_api, local_var_params['accept_partial']) result.get = translated_get return result except ApiException as e: - raise self._translate_write_exception(e, use_v2_api) + raise self._translate_write_exception(e, use_v2_api, local_var_params['accept_partial']) def _call_api( self, resource_path, method, @@ -926,26 +925,55 @@ def _sanitize_for_serialization(self, obj): 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: + @staticmethod + def _create_unsupported_endpoint_exception(use_v2_api: bool) -> ApiException: + if use_v2_api: message = ("Server doesn't support the V2 API endpoint (/api/v2/write). " "Set use_v2_api=False to use the V3 API endpoint.") - ex = ApiException(status=0, reason=message) - ex.message = message - ex.args = (message,) - return ex - if not use_v2_api and exc.status == HTTPStatus.METHOD_NOT_ALLOWED: + else: message = ("Server doesn't support the V3 API endpoint (/api/v3/write_lp). " "Set use_v2_api=True to use the V2 API endpoint.") - ex = ApiException(status=0, reason=message) - ex.message = message - ex.args = (message,) - return ex - partial = InfluxDBPartialWriteError.from_response(exc.response) - if partial is not None: - return partial + ex = ApiException(status=0, reason=message) + ex.message = message + ex.args = (message,) + return ex + + def _translate_write_exception( + self, + exc: ApiException, + use_v2_api: bool, + accept_partial: bool, + ) -> Union[ApiException, InfluxDBPartialWriteError]: + if exc.status == HTTPStatus.METHOD_NOT_ALLOWED: + return WriteApi._create_unsupported_endpoint_exception(use_v2_api) + + root = self._parse_json(exc.body) + if (root is None or + root == "" or + isinstance(root, dict) is False or + (isinstance(root, dict) and (not root.get("error") and not root.get("message")))): + message = WriteApi._extract_fallback_reason(exc.response) + exc.message = message + return exc + + if isinstance(root, dict) and WriteApi._is_partial_write_error(exc.status, use_v2_api, accept_partial, root): + # InfluxDB 3 Core/Enterprise partial write error format: + # {"error":"...","data":[{"error_message":"...","line_number":2,"original_line": "..."}]} + return self._handle_partial_write_error(exc.response, root) + + exc.message = WriteApi._get_message(root, exc.response) return exc + @staticmethod + def _parse_json(json_str: Optional[Union[str, bytes]]) -> Optional[Any]: + if not json_str: + return None + try: + return json.loads(json_str) + except (JSONDecodeError, TypeError, ValueError) as e: + logger.debug("Can't parse msg from response body %s: %s", json_str, e) + return None + def _should_gzip(self, payload: str, enable_gzip: bool = False, gzip_threshold: int = None) -> bool: """ Determines whether gzip compression should be applied to the given payload based @@ -981,6 +1009,118 @@ def _should_gzip(self, payload: str, enable_gzip: bool = False, gzip_threshold: return False + @staticmethod + def _handle_partial_write_error(response, root: dict) -> InfluxDBPartialWriteError: + reason = root.get("error") or "" + parse_res = WriteApi._parse_partial_write_line_errors(root.get("data")) + all_typed, line_errors = parse_res if parse_res is not None else (False, []) + line_error_details = WriteApi._format_partial_write_details(root, all_typed, line_errors) + if line_error_details: + details_str = "".join(f"\n\t{detail}" for detail in line_error_details) + reason = f"{reason}:{details_str}" + return InfluxDBPartialWriteError(response, reason, line_errors) + + @staticmethod + def _is_partial_write_error(status_code, use_v2_api, accept_partial, root) -> bool: + return (status_code == HTTPStatus.BAD_REQUEST and + accept_partial is True and + use_v2_api is False and + (isinstance(root.get('data'), list) and len(root.get('data')) > 0) + ) + + @staticmethod + def _parse_partial_write_data_item(item: Any) -> Optional[InfluxDBPartialWriteLineError]: + if not isinstance(item, dict): + return None + + line_number = item.get("line_number") + if line_number is not None and (not isinstance(line_number, int) or isinstance(line_number, bool)): + return None + + error_message = item.get("error_message") + if not error_message: + return None + + return InfluxDBPartialWriteLineError(line_number, error_message, item.get("original_line")) + + @staticmethod + def _parse_partial_write_line_errors(data: Any) -> Optional[Tuple[bool, List[InfluxDBPartialWriteLineError]]]: + if not isinstance(data, list): + return None + + parsed_items = [WriteApi._parse_partial_write_data_item(item) for item in data] + all_typed = all(item is not None for item in parsed_items) + line_errors = [item for item in parsed_items if item is not None] + return all_typed, line_errors + + @staticmethod + def _parse_typed_partial_write_object_or_none(data) -> Optional[InfluxDBPartialWriteLineError]: + try: + return WriteApi._parse_partial_write_data_item(data) + except ValueError: + return None + + @staticmethod + def _format_line_error(line_error: InfluxDBPartialWriteLineError) -> str: + if line_error.line_number is not None and line_error.original_line is not None: + return f"line {line_error.line_number}: {line_error.error_message} ({line_error.original_line})" + if line_error.line_number is not None: + return f"line {line_error.line_number}: {line_error.error_message}" + return f"{line_error.error_message}" + + @staticmethod + def _format_partial_write_details( + root: dict, all_typed: bool, line_errors: List[InfluxDBPartialWriteLineError] + ) -> List[str]: + if all_typed: + return [WriteApi._format_line_error(err) for err in line_errors] + + return [ + json.dumps(raw, separators=(',', ':')) + for raw in (root.get('data') or []) + if raw is not None and raw != "null" + ] + + @staticmethod + def _get_message(root, response): + if root: + try: + if isinstance(root, dict): + # InfluxDB v3 error format: { "code": "...", "message": "..." } + message = root.get("message") + if message: + code = root.get("code") + return f"{code}: {message}" if code else message + + # Core/Enterprise object format: + # {"error":"...","data":{"error_message":"..."}} + error_text = root.get("error") + if error_text: + data = root.get("data") + if isinstance(data, dict): + line_error = WriteApi._parse_typed_partial_write_object_or_none(data) + if line_error is not None and line_error.error_message: + return f"{error_text}:\n\t{WriteApi._format_line_error(line_error)}" + return error_text + except Exception as e: + logger.debug("Cannot parse error response to JSON: %s, %s", response.data, e) + return response.data + + @staticmethod + def _extract_fallback_reason(response) -> str: + # Fallback to header + for header_key in ["X-Platform-Error-Code", "X-Influx-Error", "X-InfluxDb-Error"]: + header_value = response.getheader(header_key) + if header_value is not None: + return header_value + + # Fallback to raw body + if response.data is not None and response.data != "": + return response.data + + # Fallback to http Status + return response.reason + @staticmethod def _on_error(ex): logger.error("unexpected error during batching: %s", ex) diff --git a/tests/test_influxdb_client_3_integration.py b/tests/test_influxdb_client_3_integration.py index c7dfe8d5..b087baad 100644 --- a/tests/test_influxdb_client_3_integration.py +++ b/tests/test_influxdb_client_3_integration.py @@ -199,7 +199,7 @@ def test_v3_error(self): client.write(lp) self.assertEqual(400, err.exception.status) - self.assertEqual("line protocol parsing error", err.exception.message) + self.assertIn("line protocol parsing error", err.exception.message) body = json.loads(err.exception.body) self.assertEqual("line protocol parsing error", body["error"]) self.assertEqual(2, body["data"]["line_number"]) diff --git a/tests/test_write_api.py b/tests/test_write_api.py index 9460ecc3..fcb51197 100644 --- a/tests/test_write_api.py +++ b/tests/test_write_api.py @@ -1,7 +1,10 @@ import asyncio +import http import json import unittest import uuid +from dataclasses import dataclass, field +from typing import Optional, List from unittest import mock import pytest @@ -9,7 +12,7 @@ from urllib3.exceptions import ConnectTimeoutError from influxdb_client_3 import InfluxDBClient3, InfluxDBError -from influxdb_client_3.exceptions import InfluxDBPartialWriteError +from influxdb_client_3.exceptions import InfluxDBPartialWriteError, InfluxDBPartialWriteLineError from influxdb_client_3.version import VERSION from influxdb_client_3.write_client.write_exceptions import ApiException @@ -17,6 +20,269 @@ _sentHeaders = {} +@dataclass +class TestCase: + __test__ = False + name: str + status_code: int + response_body: str + content_type: Optional[str] = None + use_v2_api: bool = False + accept_partial: bool = False + expected_msg: str = "" + expect_partial: bool = False + expected_lines: List[InfluxDBPartialWriteLineError] = field(default_factory=list) + + def __str__(self): + return self.name + + +POINTS = ( + "home,room=Sunroom temp=96 1735545600\n" + 'home,room=Sunroom temp="hi" 1735545610\n' + "home,room=Sunroom temp=88i 1735545620" +) +REJECTED_LINE = 'home,room=Sunroom temp="hi" 1735545610' +REJECTED_LINE_JSON = 'home,room=Sunroom temp=\\"hi\\" 1735545610' +LINE_ERROR = ( + "invalid column type for column 'temp', expected " + "iox::column_type::field::float, got iox::column_type::field::string" +) + +TEST_CASES = [ + TestCase( + name="V3 accept partial with renamed error and non-empty array", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=( + f'{{"error":"write completed with rejected rows","data":[' + f'{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}' + f"]}}" + ), + accept_partial=True, + expected_msg=f"write completed with rejected rows:\n\tline 2: {LINE_ERROR} ({REJECTED_LINE})", + expect_partial=True, + expected_lines=[ + InfluxDBPartialWriteLineError( + error_message=LINE_ERROR, + line_number=2, + original_line=REJECTED_LINE, + ) + ], + ), + TestCase( + name="V3 accept partial without content type", + status_code=http.client.BAD_REQUEST, + content_type=None, + response_body=( + f'{{"error":"write completed with rejected rows","data":[' + f'{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}' + f"]}}" + ), + accept_partial=True, + expected_msg=f"write completed with rejected rows:\n\tline 2: {LINE_ERROR} ({REJECTED_LINE})", + expect_partial=True, + expected_lines=[ + InfluxDBPartialWriteLineError( + error_message=LINE_ERROR, + line_number=2, + original_line=REJECTED_LINE, + ) + ], + ), + TestCase( + name="V3 accept partial with malformed non-empty array", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=( + f'{{"error":"write completed with rejected rows","data":[' + f'{{"line_number":"invalid","original_line":"{REJECTED_LINE_JSON}"}}' + f"]}}" + ), + accept_partial=True, + expected_msg=f'write completed with rejected rows:\n\t{{"line_number":"invalid","original_line":' + f'"{REJECTED_LINE_JSON}"}}', + expect_partial=True, + expected_lines=[], + ), + TestCase( + name="V3 accept partial with mixed primitive and typed entries", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=( + f'{{"error":"write completed with rejected rows","data":[' + f'1,{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}' + f"]}}" + ), + accept_partial=True, + expected_msg=( + f"write completed with rejected rows:\n\t1\n\t" + f'{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}' + ), + expect_partial=True, + expected_lines=[ + InfluxDBPartialWriteLineError( + error_message=LINE_ERROR, + line_number=2, + original_line=REJECTED_LINE, + ) + ], + ), + TestCase( + name="V3 accept partial with string entries", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=f'{{"error":"write completed with rejected rows","data":["{REJECTED_LINE_JSON}"]}}', + accept_partial=True, + expected_msg=f'write completed with rejected rows:\n\t"{REJECTED_LINE_JSON}"', + expect_partial=True, + expected_lines=[], + ), + TestCase( + name="V3 accept partial with error message only", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=f'{{"error":"write completed with rejected rows",' + f'"data":[{{"error_message":"{LINE_ERROR}"}}]}}', + accept_partial=True, + expected_msg=f"write completed with rejected rows:\n\t{LINE_ERROR}", + expect_partial=True, + expected_lines=[ + InfluxDBPartialWriteLineError( + error_message=LINE_ERROR, + line_number=None, + original_line=None, + ) + ], + ), + TestCase( + name="V3 accept partial with line number but no original line", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=f'{{"error":"write completed with rejected rows","data":[{{"error_message":"{LINE_ERROR}",' + f'"line_number":2}}]}}', + accept_partial=True, + expected_msg=f"write completed with rejected rows:\n\tline 2: {LINE_ERROR}", + expect_partial=True, + expected_lines=[ + InfluxDBPartialWriteLineError( + error_message=LINE_ERROR, + line_number=2, + original_line=None, + ) + ], + ), + TestCase( + name="V3 accept partial with entry missing error message", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=f'{{"error":"write completed with rejected rows",' + f'"data":[{{"line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}]}}', + accept_partial=True, + expected_msg=f'write completed with rejected rows:\n\t{{"line_number":2,' + f'"original_line":"{REJECTED_LINE_JSON}"}}', + expect_partial=True, + expected_lines=[], + ), + TestCase( + name="V3 accept partial with empty array", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body='{"error":"write failed","data":[]}', + accept_partial=True, + expected_msg="write failed", + expect_partial=False, + ), + TestCase( + name="V3 accept partial with object details remains generic", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=( + f'{{"error":"line protocol parsing error","data":' + f'{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}}}' + ), + accept_partial=True, + expected_msg=f"line protocol parsing error:\n\tline 2: {LINE_ERROR} ({REJECTED_LINE})", + expect_partial=False, + ), + TestCase( + name="V3 reject partial with object details", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=( + f'{{"error":"line protocol parsing error","data":' + f'{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}}}' + ), + accept_partial=False, + expected_msg=f"line protocol parsing error:\n\tline 2: {LINE_ERROR} ({REJECTED_LINE})", + expect_partial=False, + ), + TestCase( + name="V2 never returns partial write error", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body=( + f'{{"error":"partial write of line protocol occurred","data":[' + f'{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}' + f"]}}" + ), + use_v2_api=True, + accept_partial=True, + expected_msg="partial write of line protocol occurred", + expect_partial=False, + ), + TestCase( + name="V3 non-400 never returns partial write error", + status_code=http.client.INTERNAL_SERVER_ERROR, + content_type="application/json", + response_body=( + f'{{"error":"partial write of line protocol occurred","data":[' + f'{{"error_message":"{LINE_ERROR}","line_number":2,"original_line":"{REJECTED_LINE_JSON}"}}' + f"]}}" + ), + accept_partial=True, + expected_msg="partial write of line protocol occurred", + expect_partial=False, + ), + TestCase( + name="V3 scalar data remains generic", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body='{"error":"write failed","data":"invalid"}', + accept_partial=True, + expected_msg="write failed", + expect_partial=False, + ), + TestCase( + name="V3 empty object data remains generic", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body='{"error":"write failed","data":{}}', + accept_partial=True, + expected_msg="write failed", + expect_partial=False, + ), + TestCase( + name="V3 null data remains generic", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body='{"error":"write failed","data":null}', + accept_partial=True, + expected_msg="write failed", + expect_partial=False, + ), + TestCase( + name="V3 malformed JSON preserves raw response", + status_code=http.client.BAD_REQUEST, + content_type="application/json", + response_body='{"error":"write failed"', + accept_partial=True, + expected_msg='{"error":"write failed"', + expect_partial=False, + ), +] + + class WriteApiTests(unittest.TestCase): received_timeout_total = None @@ -29,18 +295,22 @@ def mock_urllib3_timeout_request(method, return response.HTTPResponse(status=200, version=4, reason="OK", decode_content=False, request_url=url) - def _test_api_error(self, body): + def _test_api_error(self, body, header=None, accept_partial=None, use_v2_api=None): client = InfluxDBClient3( host='http://localhost:8181', token='my-token', database='my-bucket', org='my-org' ) + if body is not None: + body = body.encode() + client._write_api.rest_client.pool_manager.request \ = mock.Mock(return_value=response.HTTPResponse(status=400, + headers=header or {}, reason='Bad Request', - body=body.encode())) - client._write_api.write(record="data,foo=bar val=3.14") + body=body)) + client._write_api.write(record="data,foo=bar val=3.14", accept_partial=accept_partial, use_v2_api=use_v2_api) def test_default_headers(self): client = InfluxDBClient3( @@ -98,6 +368,8 @@ def test_api_error_v3_with_detail(self): "\tline 3: invalid column type for column 'v', expected iox::column_type::field::float, " "got iox::column_type::field::uinteger (***.INF.remote_***)", True, + False, + 1 ), # error_message only (no line_number/original_line) ( @@ -107,6 +379,8 @@ def test_api_error_v3_with_detail(self): "partial write of line protocol occurred:\n" "\tonly error message", True, + False, + 1 ), # non-dict item in data list is skipped ( @@ -114,8 +388,10 @@ def test_api_error_v3_with_detail(self): '{"error":"partial write of line protocol occurred","data":[null,' '{"error_message":"bad line","line_number":2,"original_line":"bad lp"}]}', "partial write of line protocol occurred:\n" - "\tline 2: bad line (bad lp)", + "\t{\"error_message\":\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}", True, + False, + 1 ), # details empty -> return error_text ( @@ -123,7 +399,9 @@ def test_api_error_v3_with_detail(self): '{"error":"partial write of line protocol occurred","data":[{"line_number":2}]}', "partial write of line protocol occurred:\n" "\t{\"line_number\":2}", + True, False, + 0 ), # typed parse fails due line_number type -> raw fallback details ( @@ -132,7 +410,9 @@ def test_api_error_v3_with_detail(self): '[{"error_message":"bad line","line_number":"x","original_line":"bad lp"}]}', "partial write of line protocol occurred:\n" "\t{\"error_message\":\"bad line\",\"line_number\":\"x\",\"original_line\":\"bad lp\"}", + True, False, + 0 ), # mixed valid + malformed in array -> raw fallback for whole array ( @@ -142,7 +422,9 @@ def test_api_error_v3_with_detail(self): "partial write of line protocol occurred:\n" "\t{\"error_message\":\"bad line\",\"line_number\":2,\"original_line\":\"bad lp\"}\n" "\t1", + True, False, + 1 ), # data is not a dict when resolving fallback keys ( @@ -150,14 +432,18 @@ def test_api_error_v3_with_detail(self): '{"error":"data not list","data":"oops"}', "data not list", False, + True, + 0 ), # typed object with empty message is dropped ( "empty error_message in object", - '{"error":"partial write of line protocol occurred","data":' + '{"error":"parsing failed for write_lp endpoint","data":' '{"error_message":"","line_number":2,"original_line":"bad lp"}}', - "partial write of line protocol occurred", + "parsing failed for write_lp endpoint", False, + True, + 0 ), # typed array parse fails, raw fallback skips null item ( @@ -166,64 +452,79 @@ def test_api_error_v3_with_detail(self): '[null,{"error_message":123}]}', "partial write of line protocol occurred:\n" "\t{\"error_message\":123}", + True, False, + 1 ), ] - for name, response_body, expected, is_partial in cases: + for name, response_body, expected, is_partial, use_v2_api, expected_line_error_count in cases: with self.subTest(name): - with self.assertRaises(InfluxDBError) as err: - self._test_api_error(response_body) - self.assertEqual(expected, err.exception.message) if is_partial: + with self.assertRaises(InfluxDBPartialWriteError) as err: + self._test_api_error(body=response_body, accept_partial=is_partial, use_v2_api=use_v2_api) self.assertIsInstance(err.exception, InfluxDBPartialWriteError) - self.assertGreaterEqual(len(err.exception.line_errors), 1) + self.assertGreaterEqual(len(err.exception.line_errors), expected_line_error_count) else: - self.assertNotIsInstance(err.exception, InfluxDBPartialWriteError) + with self.assertRaises(ApiException) as err: + self._test_api_error(body=response_body, accept_partial=is_partial, use_v2_api=use_v2_api) + self.assertEqual(expected, err.exception.message) - def test_api_error_v3_parsing_failed_object_returns_partial_error(self): + def test_api_error_v3_parsing_failed_object_returns_error(self): response_body = ('{"error":"parsing failed for write_lp endpoint","data":' '{"error_message":"invalid field value","line_number":2,"original_line":"m,t=a f=bad"}}') - with self.assertRaises(InfluxDBPartialWriteError) as err: + with self.assertRaises(ApiException) as err: self._test_api_error(response_body) - self.assertEqual(1, len(err.exception.line_errors)) - self.assertEqual(2, err.exception.line_errors[0].line_number) + self.assertEqual('parsing failed for write_lp endpoint:\n\tline 2: invalid field value (m,t=a f=bad)', + err.exception.message) - def test_api_error_v3_partial_write_with_message_only_object_returns_partial_error(self): - response_body = ('{"error":"partial write of line protocol occurred","data":' + def test_api_error_v3_write_with_message_only_object_returns(self): + response_body = ('{"error":"parsing failed for write_lp endpoint","data":' '{"error_message":"only error message"}}') - with self.assertRaises(InfluxDBPartialWriteError) as err: + with self.assertRaises(ApiException) as err: self._test_api_error(response_body) - self.assertEqual(1, len(err.exception.line_errors)) - self.assertEqual(0, err.exception.line_errors[0].line_number) - self.assertEqual("", err.exception.line_errors[0].original_line) + self.assertEqual("parsing failed for write_lp endpoint:\n\tonly error message", err.exception.message) - def test_api_error_v3_partial_write_with_line_number_without_original_line(self): - response_body = ('{"error":"partial write of line protocol occurred","data":' + def test_api_error_v3_write_with_line_number_without_original_line(self): + response_body = ('{"error":"parsing failed for write_lp endpoint","data":' '{"error_message":"invalid field value","line_number":2}}') - with self.assertRaises(InfluxDBPartialWriteError) as err: + with self.assertRaises(ApiException) as err: self._test_api_error(response_body) - self.assertEqual(1, len(err.exception.line_errors)) - self.assertEqual("partial write of line protocol occurred:\n\tline 2: invalid field value", + self.assertEqual("parsing failed for write_lp endpoint:\n\tline 2: invalid field value", err.exception.message) - def test_partial_write_from_response_guards(self): - self.assertIsNone(InfluxDBPartialWriteError.from_response(None)) - - empty_body = response.HTTPResponse(status=400, reason="Bad Request", body=b"") - self.assertIsNone(InfluxDBPartialWriteError.from_response(empty_body)) - - invalid_json = response.HTTPResponse(status=400, reason="Bad Request", body=b"{") - self.assertIsNone(InfluxDBPartialWriteError.from_response(invalid_json)) - - non_dict_json = response.HTTPResponse(status=400, reason="Bad Request", body=b"[]") - self.assertIsNone(InfluxDBPartialWriteError.from_response(non_dict_json)) - - object_without_typed_line_error = response.HTTPResponse( - status=400, - reason="Bad Request", - body=b'{"error":"partial write of line protocol occurred","data":{"error_message":123}}', - ) - self.assertIsNone(InfluxDBPartialWriteError.from_response(object_without_typed_line_error)) + def test_fallback_header_or_body(self): + for body in ["{err", "[]", "{}"]: + for is_partial_write in [False, True]: + # Fallback to header message + with self.assertRaises(InfluxDBError) as err: + header = {"X-Influx-Error": "not used"} + self._test_api_error( + body=body, + header=header, + accept_partial=is_partial_write, + use_v2_api=False + ) + self.assertEqual(header["X-Influx-Error"], err.exception.message) + + # Fallback to raw body + with self.assertRaises(InfluxDBError) as err: + self._test_api_error( + body=body, + accept_partial=is_partial_write, + use_v2_api=False + ) + self.assertEqual(body, err.exception.message) + + def test_fallback_status_code_msg(self): + for body in ["", None]: + for is_partial_write in [False, True]: + with self.assertRaises(InfluxDBError) as err: + self._test_api_error( + body=body, + accept_partial=is_partial_write, + use_v2_api=False + ) + self.assertEqual('Bad Request', err.exception.message) def test_api_error_headers(self): body = '{"error": "test error"}' @@ -333,6 +634,7 @@ def test_post_write_async_translates_exceptions(self): ( "v2 on v3-only backend", True, + False, response.HTTPResponse(status=405, reason="Method Not Allowed", body=b""), ApiException, "Server doesn't support the V2 API endpoint (/api/v2/write). " @@ -341,6 +643,7 @@ def test_post_write_async_translates_exceptions(self): ( "v3 on v2-only backend", False, + False, response.HTTPResponse(status=405, reason="Method Not Allowed", body=b""), ApiException, "Server doesn't support the V3 API endpoint (/api/v3/write_lp). " @@ -349,6 +652,7 @@ def test_post_write_async_translates_exceptions(self): ( "v3 partial write response", False, + True, response.HTTPResponse( status=400, reason="Bad Request", @@ -361,7 +665,7 @@ def test_post_write_async_translates_exceptions(self): None, ), ] - for name, use_v2_api, http_resp, expected_type, expected_message in cases: + for name, use_v2_api, accept_partial, http_resp, expected_type, expected_message in cases: with self.subTest(name): client = InfluxDBClient3( host='http://localhost:8181', @@ -379,7 +683,7 @@ def test_post_write_async_translates_exceptions(self): bucket="TEST_BUCKET", body="home,room=Sunroom temp=96 1735545600", precision='s', - accept_partial=False, + accept_partial=accept_partial, no_sync=False, async_req=True, _async_req=True, @@ -423,3 +727,34 @@ async def run(): expected = ("Server doesn't support the V3 API endpoint (/api/v3/write_lp). " "Set use_v2_api=True to use the V2 API endpoint.") self.assertEqual(expected, err.exception.message) + + def test_write_error_classification(self): + for tc in TEST_CASES: + with self.subTest(tc.name): + headers = {"Content-Type": tc.content_type} if tc.content_type is not None else {} + + client = InfluxDBClient3( + host="http://localhost:8086", + token="token", + database="database", + ) + client._write_api.rest_client.pool_manager.request = mock.Mock( + return_value=response.HTTPResponse( + status=tc.status_code, + headers=headers, + body=tc.response_body.encode("utf-8"), + ) + ) + + expected_exc = InfluxDBPartialWriteError if tc.expect_partial else InfluxDBError + with self.assertRaises(expected_exc) as cm: + client.write( + record=POINTS, + use_v2_api=tc.use_v2_api, + accept_partial=tc.accept_partial, + ) + + err = cm.exception + self.assertEqual(tc.expected_msg, err.message) + if tc.expect_partial: + self.assertEqual(tc.expected_lines, err.line_errors)