Bulk processor bugfixes (#34)

* error handling

* bump to next patch version

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "xata"
-version = "0.5.0"
+version = "0.5.1"
 description = "Python client for Xata.io"
 authors = ["Xata <[email protected]>"]
 license = "Apache-2.0"

xata/client.py

@@ -46,7 +46,7 @@
 
 # TODO this is a manual task, to keep in sync with pyproject.toml
 # could/should be automated to keep in sync
-__version__ = "0.5.0"
+__version__ = "0.5.1"
 
 PERSONAL_API_KEY_LOCATION = "~/.config/xata/key"
 DEFAULT_BASE_URL_DOMAIN = "xata.sh"

xata/helpers.py

@@ -109,10 +109,10 @@ def process(self, id: int):
                 if r.status_code != 200:
                     self.logger.error(
                         "thread #%d: unable to process batch for table '%s', with error: %d - %s"
-                        % (id, batch["table"], r.status_code, r.json()["message"])
+                        % (id, batch["table"], r.status_code, r.json())
                     )
                     # TODO add records to batch again or callback
-                    raise Exception(r.json()["message"])
+                    raise Exception(r.json())
 
                 self.logger.debug(
                     "thread #%d: pushed a batch of %d records to table %s"
@@ -165,7 +165,9 @@ def flush_queue(self):
         self.logger.debug("flushing queue with %d records .." % (self.records.size()))
         self.records.set_flush_interval(0)
         self.processing_timeout = 0
+
         while self.stats["queue"] > 0:
+            self.logger.debug("flushing queue with %d records." % self.stats["queue"])
             time.sleep(self.processing_timeout / len(self.thread_workers) + 0.01)
 
     class Records(object):