Gg/handle stream exception (#431)

* handle low level http error as stream end

* fix deprecation warnings

Author: genzgd

CHANGELOG.md

@@ -11,11 +11,16 @@ release (0.9.0), unrecognized arguments/keywords for these methods of creating a
 instead of being passed as ClickHouse server settings. This is in conjunction with some refactoring in Client construction.
 The supported method of passing ClickHouse server settings is to prefix such arguments/query parameters with`ch_`.
 
+## 0.8.8, 2024-11-27
+### Improvement
+- Handle low level HTTP errors as "Stream Complete".  This provides better compatibility with the most recent
+ClickHouse version when the HTTP stream is abruptly closed after a server error.
+
 ## 0.8.7, 2024-11-21
 ### Improvement
 - Added basic support for ClickHouse geometric types Ring, Polygon, MultiPolygon, LineString, and MultiLineString.
 Closes https://github.com/ClickHouse/clickhouse-connect/issues/427
-- 
+
 ### Bug Fix
 - Settings/parameters from one Client will no longer leak into later client instantiations.  Fixes
 https://github.com/ClickHouse/clickhouse-connect/issues/426

clickhouse_connect/__version__.py

@@ -1 +1 @@
-version = '0.8.7'
+version = '0.8.8'

clickhouse_connect/datatypes/temporal.py

@@ -79,7 +79,6 @@ def _read_column_binary(self, source: ByteSource, num_rows: int, ctx: QueryConte
         return data_conv.read_date32_col(source, num_rows)
 
 
-from_ts_naive = datetime.utcfromtimestamp
 from_ts_tz = datetime.fromtimestamp
 
 
@@ -193,7 +192,7 @@ def _read_binary_tz(self, column: Sequence, tz_info: tzinfo):
     def _read_binary_naive(self, column: Sequence):
         new_col = []
         app = new_col.append
-        dt_from = datetime.utcfromtimestamp
+        dt_from = datetime.fromtimestamp
         prec = self.prec
         for ticks in column:
             seconds = ticks // prec

clickhouse_connect/driver/dataconv.py

@@ -28,9 +28,6 @@ def read_ipv4_col(source: ByteSource, num_rows: int):
 
 def read_datetime_col(source: ByteSource, num_rows: int, tz_info: Optional[tzinfo]):
     src_array = source.read_array('I', num_rows)
-    if tz_info is None:
-        fts = datetime.utcfromtimestamp
-        return [fts(ts) for ts in src_array]
     fts = datetime.fromtimestamp
     return [fts(ts, tz_info) for ts in src_array]
 

clickhouse_connect/driver/httputil.py

@@ -228,7 +228,12 @@ def buffered():
             read_gen = response.stream(chunk_size, decompress is None)
             while True:
                 while not done:
-                    chunk = next(read_gen, None) # Always try to read at least one chunk if there are any left
+                    try:
+                        chunk = next(read_gen, None) # Always try to read at least one chunk if there are any left
+                    except Exception: # pylint: disable=broad-except
+                        # By swallowing an unexpected exception reading the stream, we will let consumers decide how to
+                        # handle the unexpected end of stream
+                        pass
                     if not chunk:
                         done = True
                         break

clickhouse_connect/tools/datagen.py

@@ -18,8 +18,7 @@
 from clickhouse_connect.datatypes.temporal import Date, Date32, DateTime, DateTime64
 from clickhouse_connect.driver.common import array_sizes
 
-dt_from_ts = datetime.utcfromtimestamp
-dt_from_ts_tz = datetime.fromtimestamp
+dt_from_ts = datetime.fromtimestamp
 epoch_date = date(1970, 1, 1)
 date32_start_date = date(1925, 1, 1)
 
@@ -138,7 +137,7 @@ def random_datetime():
 
 
 def random_datetime_tz(timezone: tzinfo):
-    return dt_from_ts_tz(int(random() * 2 ** 32), timezone).replace(microsecond=0)
+    return dt_from_ts(int(random() * 2 ** 32), timezone).replace(microsecond=0)
 
 
 def random_ascii_str(max_len: int = 200, min_len: int = 0):
@@ -172,7 +171,7 @@ def random_datetime64_tz(prec: int, timezone: tzinfo):
         u_sec = int(random() * 1000) * 1000
     else:
         u_sec = int(random() * 1000000)
-    return dt_from_ts_tz(int(random() * 4294967296), timezone).replace(microsecond=u_sec)
+    return dt_from_ts(int(random() * 4294967296), timezone).replace(microsecond=u_sec)
 
 
 def random_ipv6():

tests/integration_tests/test_pandas.py

@@ -257,7 +257,7 @@ def test_pandas_row_df(test_client: Client, table_context:Callable):
         source_df = df.copy()
         test_client.insert_df('test_pandas_row_df', df)
         result_df = test_client.query_df('SELECT * FROM test_pandas_row_df', column_formats={'fs': 'string'})
-        assert str(result_df.dtypes[2]) == 'string'
+        assert str(result_df.dtypes.iloc[2]) == 'string'
         assert result_df.iloc[0]['key'] == 2
         assert result_df.iloc[0]['dt'] == pd.Timestamp(2023, 10, 15, 14, 50, 2, 4038)
         assert result_df.iloc[0]['fs'] == 'bit'