fix: partial write error handling

.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 }}

CHANGELOG.md

@@ -2,6 +2,23 @@
 
 ## 0.22.0 [unreleased]
 
+### Breaking Changes
+
+1. [#239](https://github.com/InfluxCommunity/influxdb3-python/pull/239): Exception classes now function mostly like a data carrying object, It stores almost no logics.  
+    - `InfluxDBPartialWriteError` class constructor will now except `message` as an argument.
+    - `InfluxDBPartialWriteError.from_response(cls, response: HTTPResponse):` function was removed.
+
+### Bug Fixes
+
+1. [#237](https://github.com/InfluxCommunity/influxdb3-python/pull/237): Makes the writing API simpler and more consistent with other v3 clients:
+    - Further simplifies the `WriteApi` request path by constructing v2/v3 requests directly through `RestClient`, while preserving existing write behavior.
+1. [#239](https://github.com/InfluxCommunity/influxdb3-python/pull/239):
+    - Only throws `InfluxDBPartialWriteError` 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

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,105 +49,29 @@ 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
             self.message = message or 'no response'
             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()
 
 
 @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)

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)

influxdb_client_3/write_client/client/write/retry.py

@@ -10,6 +10,7 @@
 from urllib3.exceptions import MaxRetryError, ResponseError
 
 from influxdb_client_3.exceptions import InfluxDBError
+from influxdb_client_3.write_client.write_exceptions import ApiException, translate_write_exception
 
 logger = logging.getLogger('influxdb_client.client.write.retry')
 
@@ -124,7 +125,8 @@ def increment(self, method=None, url=None, response=None, error=None, _pool=None
         new_retry = super().increment(method, url, response, error, _pool, _stacktrace)
 
         if response is not None:
-            parsed_error = InfluxDBError(response=response)
+            api_exception = translate_write_exception(exc=ApiException(http_resp=response))
+            parsed_error = InfluxDBError(response=response, message=api_exception.message)
         elif error is not None:
             parsed_error = error
         else:

influxdb_client_3/write_client/client/write_api.py

@@ -1,6 +1,5 @@
 """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
 
@@ -11,7 +10,6 @@
 import warnings
 from collections import defaultdict
 from enum import Enum
-from http import HTTPStatus
 from multiprocessing.pool import ThreadPool
 from random import random
 from time import sleep
@@ -23,21 +21,19 @@
 from reactivex.scheduler import ThreadPoolScheduler
 from reactivex.subject import Subject
 
-from influxdb_client_3.exceptions import InfluxDBPartialWriteError
 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, translate_write_exception
 
 # Deprecated compatibility aliases.
 # New code should import these defaults from `influxdb_client_3.write_client.write_defaults`.
@@ -481,7 +477,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 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 +713,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 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 translate_write_exception(e, use_v2_api, local_var_params['accept_partial'])
 
     def _call_api(
             self, resource_path, method,
@@ -926,26 +921,6 @@ 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:
-            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:
-            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
-        return exc
-
     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