Merge pull request #373 from aperture-data/release-0.4.17

Release 0.4.17

Author: gsaluja9

aperturedb/BBoxDataCSV.py

@@ -14,10 +14,10 @@ class BBoxDataCSV(CSVParser.CSVParser):
     """
     **ApertureDB BBox Data.**
 
-    This class loads the Bounding Box Data which is present in a csv file,
+    This class loads the Bounding Box Data which is present in a CSV file,
     and converts it into a series of aperturedb queries.
 
-    :::note Is backed by a csv file with the following columns:
+    :::note Is backed by a CSV file with the following columns:
     ``IMG_KEY``, ``x_pos``, ``y_pos``, ``width``, ``height``, ``BBOX_PROP_NAME_1``, ... ``BBOX_PROP_NAME_N``, ``constraint_BBOX_PROP_NAME_1``
     :::
 
@@ -34,7 +34,7 @@ class BBoxDataCSV(CSVParser.CSVParser):
 
     **constraint_BBOX_PROP_NAME_1**: Constraints against specific property, used for conditionally adding a Bounding Box.
 
-    Example csv file::
+    Example CSV file::
 
         img_unique_id,x_pos,y_pos,width,height,type,dataset_id,constraint_dataset_id
         d5b25253-9c1e,257,154,84,125,manual,12345,12345

aperturedb/BlobDataCSV.py

@@ -12,10 +12,10 @@
 class BlobDataCSV(CSVParser.CSVParser):
     """**ApertureDB Blob Data.**
 
-    This class loads the Blob Data which is present in a csv file,
+    This class loads the Blob Data which is present in a CSV file,
     and converts it into a series of aperturedb queries.
 
-    :::note Is backed by a csv file with the following columns:
+    :::note Is backed by a CSV file with the following columns:
     ``FILENAME``, ``PROP_NAME_1``, ... ``PROP_NAME_N``, ``constraint_PROP_NAME_1``
     :::
 
@@ -25,7 +25,7 @@ class BlobDataCSV(CSVParser.CSVParser):
 
     **constraint_PROP_NAME_1**: Constraints against specific property, used for conditionally adding a Blob.
 
-    Example csv file::
+    Example CSV file::
 
         filename,name,lastname,age,id,constraint_id
         /mnt/blob1,John,Salchi,69,321423532,321423532

aperturedb/BlobNewestDataCSV.py

@@ -42,15 +42,15 @@ class BlobNewestDataCSV(CSVParser.CSVParser):
 
         the result is then used to identify if a blob has changed.
 
-        The generated fields are to be left empty in the csv input.
+        The generated fields are to be left empty in the CSV input.
 
         Summary
          This requires a constraint to be able to check if an id exists, and a generated prop
           to be able to detect if the blob matches. It will ensure only one entity exists with
           the constraints and matching the managed blob constraints.
 
     :::note
-    Is backed by a csv file with the following columns (format optional):
+    Is backed by a CSV file with the following columns (format optional):
 
         ``filename``, ``PROP_NAME_1``, ... ``PROP_NAME_N``, ``constraint_PROP1``, ``format``
 
@@ -68,7 +68,7 @@ class BlobNewestDataCSV(CSVParser.CSVParser):
         ...
     :::
 
-    Example csv file::
+    Example CSV file::
 
         filename,id,label,constraint_id,format,dataset_ver,updateif>_dataset_ver,gen_blobsha1_sha
         /home/user/file1.jpg,321423532,dog,321423532,jpg,2,2,

aperturedb/Configuration.py

@@ -14,10 +14,10 @@ class Configuration:
     use_ssl: bool = True
     use_rest: bool = False
     use_keepalive: bool = True
-    retry_connect_interval_seconds: int = 1
+    retry_interval_seconds: int = 1
     # Max number of attempts to retry the initial connection (0 means infinite)
     # This is useful when the aperturedb server is not ready yet.
-    retry_connect_max_attempts: int = 3
+    retry_max_attempts: int = 3
 
     def __repr__(self) -> str:
         mode = "REST" if self.use_rest else "TCP"

aperturedb/ConnectionDataCSV.py

@@ -12,14 +12,14 @@
 class ConnectionDataCSV(CSVParser):
     """**ApertureDB Connection Data.**
 
-    This class loads the Connection Data which is present in a csv file,
+    This class loads the Connection Data which is present in a CSV file,
     and converts it into a series of aperturedb queries.
 
-    :::note Is backed by a csv file with the following columns:
+    :::note Is backed by a CSV file with the following columns:
     ``ConnectionClass``, ``Class1@PROP_NAME``, ``Class2@PROP_NAME`` ... ``PROP_NAME_N``, ``constraint_PROP1``
     :::
 
-    Example csv file::
+    Example CSV file::
 
         ConnectionClass,_Image@id,_Descriptor@UUID,confidence,id,constraint_id
         has_image,321423532,AID-0X3E,0.4354,5432543254,5432543254
@@ -45,7 +45,7 @@ class ConnectionDataCSV(CSVParser):
 
 
     :::info
-    This example csv's first row creates connection between an Image(id=321423532) and a descriptor(UUID=AID-0X3E)
+    This example CSV's first row creates connection between an Image(id=321423532) and a descriptor(UUID=AID-0X3E)
     It also connects the images and descriptors in the subsequent rows.
 
     In the above example, the constraint_id ensures that a connection with the specified

aperturedb/Connector.py

@@ -52,6 +52,10 @@ class UnauthorizedException(Exception):
     pass
 
 
+class UnauthenticatedException(Exception):
+    pass
+
+
 @dataclass
 class Session():
 
@@ -94,14 +98,20 @@ class Connector(object):
 
     def __init__(self, host="localhost", port=55555,
                  user="", password="", token="",
-                 use_ssl=True, shared_data=None, authenticate=True,
+                 use_ssl=True,
+                 shared_data=None,
+                 authenticate=True,
                  use_keepalive=True,
-                 retry_connect_interval_seconds=1,
-                 retry_connect_max_attempts=3,
+                 retry_interval_seconds=1,
+                 retry_max_attempts=3,
                  config: Configuration = None):
+        """
+        Constructor for the Connector class.
+        """
         self.connected = False
-        self.last_response   = ''
+        self.last_response = ''
         self.last_query_time = 0
+        self.authenticated = False
 
         if config is None:
             self.host = host
@@ -117,8 +127,8 @@ def __init__(self, host="localhost", port=55555,
                 password=password,
                 name="runtime",
                 use_keepalive=use_keepalive,
-                retry_connect_interval_seconds=retry_connect_interval_seconds,
-                retry_connect_max_attempts=retry_connect_max_attempts
+                retry_interval_seconds=retry_interval_seconds,
+                retry_max_attempts=retry_max_attempts
             )
         else:
             self.config = config
@@ -128,34 +138,36 @@ def __init__(self, host="localhost", port=55555,
             self.use_keepalive = config.use_keepalive
 
         self.conn = None
-        connect_attempts = 1
-        while True:
-            self.connect(
-                details=f"Will retry in {self.config.retry_connect_interval_seconds} seconds")
-            if self.connected or connect_attempts == self.config.retry_connect_max_attempts:
-                break
-            time.sleep(self.config.retry_connect_interval_seconds)
-            connect_attempts += 1
 
-        if not self.connected:
-            raise Exception(
-                f"Could not connect to apertureDB server: {self.config}")
-
-        if authenticate:
-            self.authenticate(shared_data, user, password, token)
+        self.token = token
 
-    def authenticate(self, shared_data, user, password, token):
         if shared_data is None:
             self.shared_data = SimpleNamespace()
             self.shared_data.session = None
             self.shared_data.lock = Lock()
-            try:
-                self._authenticate(user, password, token)
-            except Exception as e:
-                raise Exception("Authentication failed:", str(e))
         else:
             self.shared_data = shared_data
 
+        self.should_authenticate = authenticate
+
+    def authenticate(self, shared_data, user, password, token):
+        """
+        Authenticate with the database. This will be called automatically from query.
+        This is separate from session refresh mechanism, and is called only once.
+        Failure leads to exception.
+        """
+        if not self.authenticated:
+            if shared_data.session is None:
+                self.shared_data.lock = Lock()
+                try:
+                    self._authenticate(user, password, token)
+                except Exception as e:
+                    raise UnauthenticatedException(
+                        "Authentication failed:", str(e))
+            else:
+                self.shared_data = shared_data
+            self.authenticated = True
+
     def __del__(self):
         if self.connected:
             self.conn.close()
@@ -300,7 +312,8 @@ def _connect(self):
                 self.conn = self.context.wrap_socket(self.conn)
 
         except BaseException as e:
-            logger.error(f"Error connecting to server: {str(e)} {self.config}")
+            logger.error(
+                f"Error connecting to server: {str(e)} {self.config}", exc_info=True, stack_info=True)
             self.conn.close()
             self.connected = False
             raise
@@ -343,7 +356,7 @@ def _query(self, query, blob_array = [], try_resume=True):
 
         # this is for session refresh attempts
         tries = 0
-        while tries < 3:
+        while tries < self.config.retry_max_attempts:
             try:
                 if self._send_msg(data):
                     response = self._recv_msg()
@@ -362,23 +375,32 @@ def _query(self, query, blob_array = [], try_resume=True):
             except OSError as ose:
                 logger.exception(ose)
                 logger.warning(f"OS error on process {os.getpid()}")
+            except AttributeError as ae:
+                if self.connected:
+                    # Only log if we got this while connected.
+                    # else it is expected after unification of query/connect
+                    logger.exception(ae)
+                    logger.warning(f"Attribute error on process {os.getpid()}")
+
             tries += 1
             logger.warning(
-                f"Connection broken. Reconnectng attempt [{tries}/{self.config.retry_connect_max_attempts}] .. PID = {os.getpid()}")
-            self.conn.close()
-            self.connected = False
+                f"Connection broken. Reconnecting attempt [{tries}/{self.config.retry_max_attempts}] .. PID = {os.getpid()}")
+
+            if self.connected:
+                self.conn.close()
+                self.connected = False
 
             self.connect(
-                details=f"Will retry in {self.config.retry_connect_interval_seconds} seconds")
-            time.sleep(self.config.retry_connect_interval_seconds)
+                details=f"Will retry in {self.config.retry_interval_seconds} seconds")
+            time.sleep(self.config.retry_interval_seconds)
 
             # Try to resume the session, in cases where the connection is severed.
             # For example aperturedb server is restarted, or network is lost.
             # While this is useful bit of code, when executed in a refresh token
             # path, this can cause a deadlock. Hence the try_resume flag.
             if try_resume:
                 self._renew_session()
-        if tries == self.config.retry_connect_max_attempts:
+        if tries == self.config.retry_max_attempts:
             raise Exception(
                 f"Could not query apertureDB using TCP.")
         return (self.last_response, response_blob_array)
@@ -399,7 +421,13 @@ def query(self, q, blobs=[]):
         Returns:
             _type_: _description_
         """
-        self._renew_session()
+        if self.should_authenticate:
+            self.authenticate(
+                shared_data=self.shared_data,
+                user=self.config.username,
+                password=self.config.password,
+                token=self.token)
+
         try:
             start = time.time()
             self.response, self.blobs = self._query(q, blobs)
@@ -410,14 +438,15 @@ def query(self, q, blobs=[]):
                 # Hope is that the query send won't be longer than the session
                 # ttl.
                 logger.warning(
-                    f"Session expired while query was sent. Retrying... {self.config}", stack_info=True)
+                    f"Session expired while query was sent. Retrying... {self.config}")
                 self._renew_session()
                 start = time.time()
                 self.response, self.blobs = self._query(q, blobs)
             self.last_query_time = time.time() - start
             return self.response, self.blobs
         except BaseException as e:
-            logger.critical("Failed to query", exc_info=True, stack_info=True)
+            logger.critical("Failed to query",
+                            exc_info=True, stack_info=True)
             raise
 
     def _renew_session(self):
@@ -436,6 +465,9 @@ def create_new_connection(self) -> Connector:
         return type(self)(
             self.host,
             self.port,
+            self.config.username,
+            self.config.password,
+            self.token,
             use_ssl=self.use_ssl,
             shared_data=self.shared_data)