Adjust retry logic and add arguments for disabling automatic tag/signal creation (#62)

Author: hallarak

VERSION

@@ -1 +1 @@
-2.4.0
+2.5.0

exabel_data_sdk/client/api/bulk_insert.py

@@ -1,5 +1,5 @@
 from concurrent.futures.thread import ThreadPoolExecutor
-from time import time
+from time import sleep, time
 from typing import Callable, Sequence
 
 from exabel_data_sdk.client.api.data_classes.request_error import ErrorType, RequestError
@@ -90,10 +90,15 @@ def _bulk_insert(
             raise BulkInsertFailedError()
 
 
+def _get_backoff(trial: int, min_sleep: float = 1.0, max_sleep: float = 60.0) -> float:
+    """Return the backoff in seconds for the given trial."""
+    return min(min_sleep * 2 ** trial, max_sleep)
+
+
 def bulk_insert(
     resources: Sequence[TResource],
     insert_func: Callable[[TResource], ResourceCreationStatus],
-    retries: int = 2,
+    retries: int = 5,
     threads: int = 40,
 ) -> ResourceCreationResults[TResource]:
     """
@@ -119,6 +124,9 @@ def bulk_insert(
                 failures = results.extract_retryable_failures()
                 if not failures:
                     break
+                backoff = _get_backoff(trial)
+                print(f"Sleeping {backoff:.2f} seconds before retrying failed requests...")
+                sleep(backoff)
                 resources = [result.resource for result in failures]
                 print(f"Retry #{trial} with {len(resources)} resources:")
             _bulk_insert(results, resources, insert_func, threads=threads)

exabel_data_sdk/client/api/entity_api.py

@@ -220,6 +220,7 @@ def bulk_create_entities(
         entity_type: str,
         threads: int = 40,
         upsert: bool = False,
+        retries: int = 5,
     ) -> ResourceCreationResults[Entity]:
         """
         Check if the provided entities exist, and create them if they don't.
@@ -238,4 +239,4 @@ def insert(entity: Entity) -> ResourceCreationStatus:
             self.create_entity(entity=entity, entity_type=entity_type)
             return ResourceCreationStatus.CREATED
 
-        return bulk_insert(entities, insert, threads=threads)
+        return bulk_insert(entities, insert, threads=threads, retries=retries)

exabel_data_sdk/client/api/relationship_api.py

@@ -310,6 +310,7 @@ def bulk_create_relationships(
         relationships: Sequence[Relationship],
         threads: int = 40,
         upsert: bool = False,
+        retries: int = 5,
     ) -> ResourceCreationResults[Relationship]:
         """
         Check if the provided relationships exist, and create them if they don't.
@@ -327,4 +328,4 @@ def insert(relationship: Relationship) -> ResourceCreationStatus:
             self.create_relationship(relationship=relationship)
             return ResourceCreationStatus.CREATED
 
-        return bulk_insert(relationships, insert, threads=threads)
+        return bulk_insert(relationships, insert, threads=threads, retries=retries)

exabel_data_sdk/client/api/time_series_api.py

@@ -313,6 +313,7 @@ def bulk_upsert_time_series(
         create_tag: bool = False,
         threads: int = 40,
         default_known_time: DefaultKnownTime = None,
+        retries: int = 5,
     ) -> ResourceCreationResults[pd.Series]:
         """
         Calls upsert_time_series for each of the provided time series,
@@ -332,6 +333,7 @@ def bulk_upsert_time_series(
                         the Known Time for data points where a specific known time timestamp
                         has not been given. If not provided, the Exabel API defaults to the
                         current time (upload time) as the Known Time.
+            retries:         Maximum number of retries to make for each failed request.
         """
 
         def insert(ts: pd.Series) -> ResourceCreationStatus:
@@ -340,7 +342,7 @@ def insert(ts: pd.Series) -> ResourceCreationStatus:
             )
             return ResourceCreationStatus.UPSERTED if existed else ResourceCreationStatus.CREATED
 
-        return bulk_insert(series, insert, threads=threads)
+        return bulk_insert(series, insert, threads=threads, retries=retries)
 
     @staticmethod
     def _series_to_time_series_points(series: pd.Series) -> Sequence[TimeSeriesPoint]:

exabel_data_sdk/scripts/csv_script.py

@@ -5,6 +5,10 @@
 import pandas as pd
 
 from exabel_data_sdk.scripts.base_script import BaseScript
+from exabel_data_sdk.services.csv_loading_constants import (
+    DEFAULT_NUMBER_OF_RETRIES,
+    DEFAULT_NUMBER_OF_THREADS,
+)
 
 
 class CsvScript(BaseScript):
@@ -56,8 +60,19 @@ def __init__(self, argv: Sequence[str], description: str):
             type=int,
             choices=range(1, 101),
             metavar="[1-100]",
-            default=40,
-            help="The number of parallel upload threads to run. Defaults to 40.",
+            default=DEFAULT_NUMBER_OF_THREADS,
+            help=f"The number of parallel upload threads to run. "
+            f"Defaults to {DEFAULT_NUMBER_OF_THREADS}.",
+        )
+        self.parser.add_argument(
+            "--retries",
+            required=False,
+            type=int,
+            choices=range(1, 51),
+            metavar="[1-50]",
+            default=DEFAULT_NUMBER_OF_RETRIES,
+            help=f"The maximum number of retries to make for each failed request. Defaults to "
+            f"{DEFAULT_NUMBER_OF_RETRIES}.",
         )
 
     def read_csv(

exabel_data_sdk/scripts/load_entities_from_csv.py

@@ -85,6 +85,7 @@ def run_script(self, client: ExabelClient, args: argparse.Namespace) -> None:
                 threads=args.threads,
                 upsert=args.upsert,
                 dry_run=args.dry_run,
+                retries=args.retries,
             )
         except CsvLoadingException as e:
             print(e)

exabel_data_sdk/scripts/load_relationships_from_csv.py

@@ -87,6 +87,7 @@ def run_script(self, client: ExabelClient, args: argparse.Namespace) -> None:
                 threads=args.threads,
                 upsert=args.upsert,
                 dry_run=args.dry_run,
+                retries=args.retries,
             )
         except CsvLoadingException as e:
             print(e)

exabel_data_sdk/scripts/load_time_series_from_csv.py

@@ -13,16 +13,18 @@ class LoadTimeSeriesFromCsv(CsvScriptWithEntityMapping):
     Processes a timeseries CSV file and uploads the time series to Exabel.
 
     The CSV file should have a header line on the format
-        entity;date;`signal_1`; ... ;`signal_n`
+        entityType;date;`signal_1`; ... ;`signal_n`
+
+    For example
+        brand;date;revenue;sales
 
     Each subsequent row consists of the following elements:
-      * the entity referred to by the entity’s resource name, e.g.,
-            entityTypes/company/entities/company_code
+      * the entity referred to by the entity’s identifier
       * the date on ISO format, e.g. 2020-12-31
       * one numerical value for each of the signals `signal_1` to `signal_n`
 
     Thus, a typical row would look like:
-        entityTypes/company/entities/company_code;2020-12-31;12;1234.56;1.23e6
+        brand_1;2020-12-31;12;1234.56;1.23e6
 
     The rows do not have to be sorted in any particular order.
     """
@@ -59,6 +61,22 @@ def __init__(self, argv: Sequence[str]):
                 "available to the user the day after, one would set --pit-offset 1"
             ),
         )
+        self.parser.add_argument(
+            "--no-create-library-signal",
+            dest="create_library_signal",
+            required=False,
+            action="store_false",
+            default=True,
+            help="Set to not create library signal DSL expressions.",
+        )
+        self.parser.add_argument(
+            "--no-create-tag",
+            dest="create_tag",
+            required=False,
+            action="store_false",
+            default=True,
+            help="Set to not create a tag for every entity type a signal has time series for.",
+        )
 
     def run_script(self, client: ExabelClient, args: argparse.Namespace) -> None:
         try:
@@ -70,8 +88,11 @@ def run_script(self, client: ExabelClient, args: argparse.Namespace) -> None:
                 pit_current_time=args.pit_current_time,
                 pit_offset=args.pit_offset,
                 create_missing_signals=args.create_missing_signals,
+                create_tag=args.create_tag,
+                create_library_signal=args.create_library_signal,
                 threads=args.threads,
                 dry_run=args.dry_run,
+                retries=args.retries,
             )
         except CsvLoadingException as e:
             print(e)

exabel_data_sdk/services/csv_entity_loader.py

@@ -2,7 +2,10 @@
 from exabel_data_sdk.client.api.bulk_insert import BulkInsertFailedError
 from exabel_data_sdk.client.api.data_classes.entity import Entity
 from exabel_data_sdk.services.csv_exception import CsvLoadingException
-from exabel_data_sdk.services.csv_loading_constants import DEFAULT_NUMBER_OF_THREADS
+from exabel_data_sdk.services.csv_loading_constants import (
+    DEFAULT_NUMBER_OF_RETRIES,
+    DEFAULT_NUMBER_OF_THREADS,
+)
 from exabel_data_sdk.services.csv_reader import CsvReader
 from exabel_data_sdk.util.resource_name_normalization import normalize_resource_name
 
@@ -29,6 +32,7 @@ def load_entities(
         upsert: bool,
         dry_run: bool = False,
         error_on_any_failure: bool = False,
+        retries: int = DEFAULT_NUMBER_OF_RETRIES,
     ) -> None:
         """
         Load a CSV file and upload the entities specified therein to the Exabel Data API.
@@ -49,6 +53,7 @@ def load_entities(
             upsert: whether entities should be updated if they already exist
             dry_run: if True, the file is processed, but no entities are actually uploaded
             error_on_any_failure: if True, an exception is raised if any entity failed to be created
+            retries: the maximum number of retries to make for each failed request
         """
         if dry_run:
             print("Running dry-run...")
@@ -61,7 +66,9 @@ def load_entities(
         }
         if description_column:
             string_columns.add(description_column)
-        entities_df = CsvReader.read_csv(filename, separator, string_columns=string_columns)
+        entities_df = CsvReader.read_csv(
+            filename, separator, string_columns=string_columns, keep_default_na=False
+        )
 
         name_col = name_column or entities_df.columns[0]
         display_name_col = display_name_column or name_col
@@ -91,7 +98,7 @@ def load_entities(
 
         try:
             result = self._client.entity_api.bulk_create_entities(
-                entities, entity_type_name, threads=threads, upsert=upsert
+                entities, entity_type_name, threads=threads, upsert=upsert, retries=retries
             )
             if error_on_any_failure and result.has_failure():
                 raise CsvLoadingException("An error occurred while uploading entities.")

exabel_data_sdk/services/csv_loading_constants.py

@@ -1 +1,2 @@
 DEFAULT_NUMBER_OF_THREADS = 40
+DEFAULT_NUMBER_OF_RETRIES = 5

exabel_data_sdk/services/csv_reader.py

@@ -8,7 +8,11 @@ class CsvReader:
 
     @staticmethod
     def read_csv(
-        filename: str, separator: str, string_columns: Iterable[Union[str, int]]
+        filename: str,
+        separator: str,
+        string_columns: Iterable[Union[str, int]],
+        *,
+        keep_default_na: bool
     ) -> pd.DataFrame:
         """
         Read the given file and return the content as a pandas DataFrame.
@@ -21,4 +25,6 @@ def read_csv(
         dtype: Optional[Mapping[Union[str, int], type]] = None
         if string_columns:
             dtype = {column: str for column in string_columns}
-        return pd.read_csv(filename, header=0, sep=separator, dtype=dtype, keep_default_na=False)
+        return pd.read_csv(
+            filename, header=0, sep=separator, dtype=dtype, keep_default_na=keep_default_na
+        )

exabel_data_sdk/services/csv_relationship_loader.py

@@ -3,7 +3,10 @@
 from exabel_data_sdk.client.api.data_classes.relationship import Relationship
 from exabel_data_sdk.client.api.data_classes.relationship_type import RelationshipType
 from exabel_data_sdk.services.csv_exception import CsvLoadingException
-from exabel_data_sdk.services.csv_loading_constants import DEFAULT_NUMBER_OF_THREADS
+from exabel_data_sdk.services.csv_loading_constants import (
+    DEFAULT_NUMBER_OF_RETRIES,
+    DEFAULT_NUMBER_OF_THREADS,
+)
 from exabel_data_sdk.services.csv_reader import CsvReader
 from exabel_data_sdk.services.entity_mapping_file_reader import EntityMappingFileReader
 from exabel_data_sdk.util.resource_name_normalization import to_entity_resource_names
@@ -32,6 +35,7 @@ def load_relationships(
         upsert: bool = False,
         dry_run: bool = False,
         error_on_any_failure: bool = False,
+        retries: int = DEFAULT_NUMBER_OF_RETRIES,
     ) -> None:
         """
         Load a CSV file and upload the relationships specified therein to the Exabel Data API.
@@ -53,6 +57,7 @@ def load_relationships(
             dry_run: if True, the file is processed, but no relationships are actually uploaded
             error_on_any_failure: if True, an exception is raised if any relationship failed to be
                 created
+            retries: the maximum number of retries to make for each failed request
         """
         if dry_run:
             print("Running dry-run...")
@@ -66,7 +71,9 @@ def load_relationships(
         if description_column:
             string_columns.add(description_column)
 
-        relationships_df = CsvReader.read_csv(filename, separator, string_columns=string_columns)
+        relationships_df = CsvReader.read_csv(
+            filename, separator, string_columns=string_columns, keep_default_na=False
+        )
 
         entity_from_col = entity_from_column
         entity_to_col = entity_to_column
@@ -119,7 +126,7 @@ def load_relationships(
 
         try:
             result = self._client.relationship_api.bulk_create_relationships(
-                relationships, threads=threads, upsert=upsert
+                relationships, threads=threads, upsert=upsert, retries=retries
             )
             if error_on_any_failure and result.has_failure():
                 raise CsvLoadingException("An error occurred while uploading relationships.")

exabel_data_sdk/services/csv_time_series_loader.py

@@ -10,7 +10,10 @@
 from exabel_data_sdk.client.api.bulk_insert import BulkInsertFailedError
 from exabel_data_sdk.client.api.data_classes.signal import Signal
 from exabel_data_sdk.services.csv_exception import CsvLoadingException
-from exabel_data_sdk.services.csv_loading_constants import DEFAULT_NUMBER_OF_THREADS
+from exabel_data_sdk.services.csv_loading_constants import (
+    DEFAULT_NUMBER_OF_RETRIES,
+    DEFAULT_NUMBER_OF_THREADS,
+)
 from exabel_data_sdk.services.csv_reader import CsvReader
 from exabel_data_sdk.services.entity_mapping_file_reader import EntityMappingFileReader
 from exabel_data_sdk.stubs.exabel.api.data.v1.time_series_messages_pb2 import DefaultKnownTime
@@ -40,9 +43,12 @@ def load_time_series(
         pit_current_time: bool = False,
         pit_offset: int = False,
         create_missing_signals: bool = False,
+        create_tag: bool = True,
+        create_library_signal: bool = True,
         threads: int = DEFAULT_NUMBER_OF_THREADS,
         dry_run: bool = False,
         error_on_any_failure: bool = False,
+        retries: int = DEFAULT_NUMBER_OF_RETRIES,
     ) -> None:
         """
         Load a CSV file and upload the time series to the Exabel Data API
@@ -64,6 +70,7 @@ def load_time_series(
             dry_run: if True, the file is processed, but no time series are actually uploaded
             error_on_any_failure: if True, an  exception is raised if any time series failed to be
                 created
+            retries: the maximum number of retries to make for each failed request
         """
         if dry_run:
             print("Running dry-run...")
@@ -79,7 +86,9 @@ def load_time_series(
             time_offset = Duration(seconds=86400 * pit_offset)
             default_known_time = DefaultKnownTime(time_offset=time_offset)
 
-        ts_data = CsvReader.read_csv(filename, separator=separator, string_columns=[0])
+        ts_data = CsvReader.read_csv(
+            filename, separator=separator, string_columns=[0], keep_default_na=True
+        )
         entity_mapping = EntityMappingFileReader.read_entity_mapping_file(
             entity_mapping_filename, separator=separator
         )
@@ -194,7 +203,7 @@ def load_time_series(
                     for signal in missing_signals:
                         self._client.signal_api.create_signal(
                             Signal(name=prefix + signal, display_name=signal),
-                            create_library_signal=True,
+                            create_library_signal=create_library_signal,
                         )
             else:
                 raise CsvLoadingException(
@@ -212,7 +221,11 @@ def load_time_series(
 
         try:
             result = self._client.time_series_api.bulk_upsert_time_series(
-                series, create_tag=True, threads=threads, default_known_time=default_known_time
+                series,
+                create_tag=create_tag,
+                threads=threads,
+                default_known_time=default_known_time,
+                retries=retries,
             )
             if error_on_any_failure and result.has_failure():
                 raise CsvLoadingException("An error occurred while uploading time series.")

exabel_data_sdk/tests/client/api/test_bulk_insert.py

@@ -0,0 +1,14 @@
+import unittest
+
+from exabel_data_sdk.client.api.bulk_insert import _get_backoff
+
+
+class TestEntityApi(unittest.TestCase):
+    def test_get_backoff(self):
+        self.assertEqual(1.0, _get_backoff(0))
+        self.assertEqual(2.0, _get_backoff(1))
+        self.assertEqual(4.0, _get_backoff(2))
+        self.assertEqual(8.0, _get_backoff(3))
+        self.assertEqual(16.0, _get_backoff(4))
+        self.assertEqual(32.0, _get_backoff(5))
+        self.assertEqual(60.0, _get_backoff(6))