Add optional parameter dynamic_read_timeout for functions that upload files to azure storage account

Author: wojciech-bochniarz-4ss

datareservoirio/client.py

@@ -104,7 +104,7 @@ def ping(self):
         return response.json()
 
     @log_decorator("exception")
-    def create(self, series=None, wait_on_verification=True):
+    def create(self, series=None, wait_on_verification=True, dynamic_read_timeout=False):
         """
         Create a new series in DataReservoir.io from a pandas.Series. If no
         data is provided, an empty series is created.
@@ -124,6 +124,12 @@ def create(self, series=None, wait_on_verification=True):
             validation is successful. The latter is significantly faster, but
             is recommended when the data is "validated" in advance.
             Default is True.
+        dynamic_read_timeout : bool (optional)
+            While uploading file there is no timeout for read operations which can cause
+            problems when there is no response from the server. If this flag is set to true,
+            application will calculate and apply timeout for read operations and retry upload 
+            if necessary.
+            Default is False.
 
         Returns
         -------
@@ -157,7 +163,7 @@ def create(self, series=None, wait_on_verification=True):
             environment.api_base_url + "files/commit",
             {"json": {"FileId": file_id}, "timeout": _TIMEOUT_DEAULT},
         )
-        self._storage.put(df, target_url, commit_request)
+        self._storage.put(df, target_url, commit_request, dynamic_read_timeout)
 
         if wait_on_verification:
             status = self._wait_until_file_ready(file_id)
@@ -173,7 +179,7 @@ def create(self, series=None, wait_on_verification=True):
         return response.json()
 
     @log_decorator("exception")
-    def append(self, series, series_id, wait_on_verification=True):
+    def append(self, series, series_id, wait_on_verification=True, dynamic_read_timeout=False):
         """
         Append data to an already existing series.
 
@@ -193,6 +199,12 @@ def append(self, series, series_id, wait_on_verification=True):
             validation is successful. The latter is significantly faster, but
             is recommended when the data is "validated" in advance.
             Default is True.
+        dynamic_read_timeout : bool (optional)
+            While uploading file there is no timeout for read operations which can cause
+            problems when there is no response from the server. If this flag is set to true,
+            application will calculate and apply timeout for read operations and retry upload 
+            if necessary.
+            Default is False.
 
         Returns
         -------
@@ -217,7 +229,7 @@ def append(self, series, series_id, wait_on_verification=True):
             {"json": {"FileId": file_id}, "timeout": _TIMEOUT_DEAULT},
         )
 
-        self._storage.put(df, target_url, commit_request)
+        self._storage.put(df, target_url, commit_request, dynamic_read_timeout)
 
         if wait_on_verification:
             status = self._wait_until_file_ready(file_id)

datareservoirio/storage/storage.py

@@ -59,7 +59,7 @@ def __init__(self, session, cache=True, cache_opt=None):
 
         self._session = session
 
-    def put(self, df, target_url, commit_request):
+    def put(self, df, target_url, commit_request, dynamic_read_timeout=False):
         """
         Put a Pandas DataFrame into storage.
 
@@ -72,9 +72,12 @@ def put(self, df, target_url, commit_request):
         commit_request : tuple
             Parameteres for "commit" request. Given as `(METHOD, URL, kwargs)`.
             The tuple is passed forward to `session.request(method=METHOD, url=URL, **kwargs)`
+        dynamic_read_timeout : bool
+            Flag that enables timeout calculation for read operation while
+            uploading file
 
         """
-        _df_to_blob(df, target_url)
+        _df_to_blob(df, target_url, _BLOBSTORAGE_SESSION, dynamic_read_timeout)
 
         method, url, kwargs = commit_request
         response = self._session.request(method=method, url=url, **kwargs)
@@ -322,7 +325,7 @@ def _blob_to_df(blob_url, session=_BLOBSTORAGE_SESSION):
     return df
 
 
-def _df_to_blob(df, blob_url, session=_BLOBSTORAGE_SESSION):
+def _df_to_blob(df, blob_url, session=_BLOBSTORAGE_SESSION, dynamic_read_timeout=False):
     """
     Upload a Pandas Dataframe as blob to a remote storage.
 
@@ -354,6 +357,13 @@ def _df_to_blob(df, blob_url, session=_BLOBSTORAGE_SESSION):
             url=blob_url,
             headers={"x-ms-blob-type": "BlockBlob"},
             data=fp,
-            timeout=(30, None),
+            timeout=(30, _calculate_timeout(fp.getbuffer().nbytes) if dynamic_read_timeout else None),
         ).raise_for_status()
     return
+
+
+def _calculate_timeout(file_size_bytes):
+    bytes_per_second = 1 * 1024 * 1024 # 1MB/s
+    min_timeout = 30
+    timeout = max(min_timeout, (file_size_bytes / bytes_per_second) * 1.5)
+    return timeout