Merge pull request #643 from aperture-data/release-0.4.54

Release 0.4.54

Author: gsaluja9

.github/workflows/pr.yaml

@@ -9,8 +9,7 @@ jobs:
   run_test:
 
     runs-on:
-    - self-hosted
-    - deployer
+    - gcp
 
     steps:
 

aperturedb/CSVParser.py

@@ -62,7 +62,17 @@ def __init__(self,
                 self.df = pd.read_csv(filename)
             else:
                 self.df = df
+
+            # we expect the df index to have 'start', which means RangeIndex.
+            # most users don't supply their own df, so this is mostly a sanity check
+            # for when an advanced user has done filtering and have a IntervalIndex.
+            if not isinstance(self.df.index, pd.RangeIndex):
+                raise TypeError(
+                    f"CSVParser requires a RangeIndex. the supplied DataFrame has a {type(self.df.index)} index.")
         else:
+            if df is not None:
+                raise ValueError(
+                    "Dask mode requires a CSV filename; DataFrame inputs are not supported.")
             # It'll impact the number of partitions, and memory usage.
             # TODO: tune this for the best performance.
             cores_used = int(CORES_USED_FOR_PARALLELIZATION * mp.cpu_count())

aperturedb/Configuration.py

@@ -116,7 +116,7 @@ def create_aperturedb_key(
 
         if host.endswith(APERTUREDB_CLOUD):
             host = host[:-1 * len(APERTUREDB_CLOUD)]
-            m = re.match("(.*)\.farm(\d+)$", host)
+            m = re.match(r"(.*)\.farm(\d+)$", host)
             if m is not None:
                 host = "{}.{}".format(m.group(1), int(m.group(2)))
                 compressed = True
@@ -170,7 +170,7 @@ def reinflate(cls, encoded_str: list) -> object:
         else:
             raise ValueError("Bad format for key list")
 
-        port_match = re.match(".*:(\d+)$", host)
+        port_match = re.match(r".*:(\d+)$", host)
         if port_match is not None:
             port = int(port_match.group(1))
             host = host[:-1 * (len(port_match.group(1)) + 1)]

aperturedb/DaskManager.py

@@ -42,7 +42,7 @@ def __del__(self):
         self._cluster.close()
 
     def run(self, QueryClass: type[ParallelQuery], client: Connector, generator, batchsize, stats):
-        def process(df, host, port, use_ssl, ca_cert, session, connnector_type):
+        def process(df, host, port, use_ssl, ca_cert, verify_hostname, session, connnector_type):
             metrics = Stats()
             # Dask reads data in partitions, and the first partition is of 2 rows, with all
             # values as 'foo'. This is for sampling the column names and types. Should not process
@@ -55,8 +55,12 @@ def process(df, host, port, use_ssl, ca_cert, session, connnector_type):
                 shared_data = SimpleNamespace()
                 shared_data.session = session
                 shared_data.lock = Lock()
-                client = connnector_type(host=host, port=port,
-                                         use_ssl=use_ssl, ca_cert=ca_cert, shared_data=shared_data)
+                client = connnector_type(
+                    host=host, port=port,
+                    use_ssl=use_ssl,
+                    ca_cert=ca_cert,
+                    verify_hostname=verify_hostname,
+                    shared_data=shared_data)
             except Exception as e:
                 logger.exception(e)
             #from aperturedb.ParallelLoader import ParallelLoader
@@ -88,6 +92,7 @@ def process(df, host, port, use_ssl, ca_cert, session, connnector_type):
             client.port,
             client.use_ssl,
             client.config.ca_cert,
+            client.config.verify_hostname,
             client.shared_data.session,
             type(client))
         computation = computation.persist()

aperturedb/Query.py

@@ -87,8 +87,8 @@ def get_specific(obj: BaseModel) -> dict:
         start, stop  = obj.start, obj.stop
         if obj.range_type == RangeType.TIME:
             start, stop = int(start), int(stop)
-            start = f"{start//60}:{start%60}"
-            stop = f"{stop//60}:{stop%60}"
+            start = f"{start//3600:0>2}:{start//60:0>2}:{start%60:0>2}"
+            stop = f"{stop//3600:0>2}:{stop//60:0>2}:{stop%60:0>2}"
         elif obj.range_type == RangeType.FRAME:
             start = int(obj.start)
             stop = int(obj.stop)
@@ -169,6 +169,8 @@ def generate_add_query(
     cindex = index
     specific_params = {}
     specific_blobs = []
+    if obj is None:
+        return query, blobs, index
     if obj.id not in cached:
         for p in obj.__dict__.keys():
             if "_" == p[0]:

aperturedb/SPARQL.py

@@ -13,6 +13,8 @@
 from aperturedb.CommonLibrary import create_connector, execute_query
 from aperturedb.Utils import Utils
 
+SAFE_PREFIX = "adb_"
+
 
 class SPARQL:
     def __init__(self, client=None, debug=False, log_level=None):
@@ -93,6 +95,12 @@ def _format_triple(self, triple):
     def _format_triples(self, triples):
         return " .\n".join([self._format_triple(triple) for triple in triples])
 
+    def _make_safe_prefix(self, suffix):
+        return f"{SAFE_PREFIX}{suffix[1:]}" if suffix.startswith("_") else suffix
+
+    def _make_command_name(self, t):
+        return "Find" + t[len(SAFE_PREFIX):] if t.startswith(f"{SAFE_PREFIX}") else "FindEntity"
+
     def _load_schema(self):
         self.schema = self._utils.get_schema()
         self.connections = {}
@@ -109,14 +117,17 @@ def _load_schema(self):
                 d_list = [d] if isinstance(d, dict) else d
 
                 for d in d_list:
-                    self.connections[uri][0].add(d["src"])
-                    self.connections[uri][1].add(d["dst"])
+                    self.connections[uri][0].add(
+                        self._make_safe_prefix(d["src"]))
+                    self.connections[uri][1].add(
+                        self._make_safe_prefix(d["dst"]))
         if not self.connections:
             self.logger.warning("No connections found in schema")
 
         self.properties = {}
         if "entities" in self.schema and self.schema["entities"] is not None and "classes" in self.schema["entities"]:
             for e, d in self.schema["entities"]["classes"].items():
+                e = self._make_safe_prefix(e)
                 for p in d["properties"]:
                     uri = self._make_uri("p", p)
                     if uri not in self.properties:
@@ -166,9 +177,9 @@ def evalBGP(self, ctx: "QueryContext",
         def add_find(v, t):
             """Create new Find* command for variable v with type t"""
             from rdflib.term import Variable
-            command_name = "FindEntity" if t[0] != "_" else "Find" + t[1:]
             body = {}
-            if t[0] != "_":
+            command_name = self._make_command_name(t)
+            if command_name == "FindEntity":
                 body["with_class"] = t
             body["_ref"] = len(query) + 1
             body["uniqueids"] = True
@@ -445,7 +456,7 @@ def add_types(k, tt):
                 add_types(s, [self._parse_uri_with_ns('t', o)])
             elif p in self.knn_properties:
                 if p == self.namespaces["knn"] + "similarTo":
-                    add_types(s, {"_Descriptor"})
+                    add_types(s, {self._make_safe_prefix("_Descriptor")})
             elif p == self.namespaces["c"] + "ANY":
                 pass
             else:
@@ -509,9 +520,10 @@ def get_blob(self, uri: Union[str, "URIRef"], type: Optional[str] = None) -> byt
             assert type == self._deduce_type(
                 uri), f"Type {type} does not match deduced type {self._deduce_type(uri)}"
         assert type is not None, f"Cannot get blob for entity URI: {uri}"
-        assert type[0] == "_", f"Cannot get blob for entity URI: {uri} with type {type}"
+        assert type.startswith(
+            f"{SAFE_PREFIX}"), f"Cannot get blob for entity URI: {uri} with type {type}"
         uniqueid = self._deduce_uniqueid(uri)
-        command_name = "Find" + type[1:]
+        command_name = self._make_command_name(type)
         query = [
             {command_name: {
                 "constraints": {
@@ -539,9 +551,10 @@ def get_blobs(self, uris: List[Union[str, "URIRef"]], type: Optional[str] = None
                 t == type for t in types), f"Types do not match: {types}"
 
         assert type is not None, f"Cannot get blob for entity URI: {uri}"
-        assert type[0] == "_", f"Cannot get blob for entity URI: {uri} with type {type}"
+        assert type.startswith(
+            f"{SAFE_PREFIX}"), f"Cannot get blob for entity URI: {uri} with type {type}"
         uniqueids = [self._deduce_uniqueid(uri) for uri in uris]
-        command_name = "Find" + type[1:]
+        command_name = self._make_command_name(type)
         query = [
             {command_name: {
                 "results": {"list": ["_uniqueid"]},
@@ -568,7 +581,7 @@ def get_image(self, uri: Union[str, "URIRef"]) -> "np.ndarray":
         import numpy as np
         import cv2
 
-        blob = self.get_blob(uri, type="_Image")
+        blob = self.get_blob(uri, type=self._make_safe_prefix("_Image"))
         nparr = np.fromstring(blob, np.uint8)
         image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
         image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
@@ -581,7 +594,7 @@ def get_images(self, uris: List[Union[str, "URIRef"]]) -> List["np.ndarray"]:
         import numpy as np
         import cv2
 
-        blobs = self.get_blobs(uris, type="_Image")
+        blobs = self.get_blobs(uris, type=self._make_safe_prefix("_Image"))
         images = []
         for blob in blobs:
             if blob is not None:
@@ -616,13 +629,14 @@ def get_descriptor(self, uri: Union[str, "URIRef"]) -> "np.ndarray":
         Get the descriptor associated with a URI or QName
         """
         import numpy as np
-        blob = self.get_blob(uri, type="_Descriptor")
+        blob = self.get_blob(uri, type=self._make_safe_prefix("_Descriptor"))
         return np.frombuffer(blob, dtype=np.float32)
 
     def get_descriptors(self, uris: List[Union[str, "URIRef"]]) -> List["np.ndarray"]:
         """
         Get the descriptors associated with a list of URI or QName
         """
         import numpy as np
-        blobs = self.get_blobs(uris, type="_Descriptor")
+        blobs = self.get_blobs(
+            uris, type=self._make_safe_prefix("_Descriptor"))
         return [np.frombuffer(blob, dtype=np.float32) for blob in blobs]

aperturedb/__init__.py

@@ -10,7 +10,7 @@
 import signal
 import sys
 
-__version__ = "0.4.53"
+__version__ = "0.4.54"
 
 logger = logging.getLogger(__name__)
 
@@ -47,13 +47,30 @@
         os.environ["ADB_LOG_FILE"]) == 0 else os.environ["ADB_LOG_FILE"]
 
 if error_file_name is not None:
+    # detachable file handler allows program to run from working directory
+    # without write access; adb is a use case for this.
+    class DetachableFileHandler(logging.FileHandler):
+        def __init__(self, file_path, delay=False):
+            super().__init__(file_path, delay=delay)
+
+        def emit(self, record):
+            try:
+                super().emit(record)
+            except PermissionError as pErr:
+                for h in logger.handlers[:]:
+                    if h == self:
+                        logger.removeHandler(h)
+                        break
+                logging.warning(
+                    f"Unable to write to {self.baseFilename}, removing file logging")
+
     error_file_tmpl = Template(error_file_name)
     template_items = {
         # python isodate has ':', not valid in files in windows.
         "now": str(datetime.datetime.now().isoformat()).replace(':', ''),
         "node": str(platform.node())
     }
-    error_file_handler = logging.FileHandler(error_file_tmpl.safe_substitute(
+    error_file_handler = DetachableFileHandler(error_file_tmpl.safe_substitute(
         **template_items), delay=True)
     error_file_handler.setFormatter(formatter)
     error_file_handler.setLevel(log_file_level)

aperturedb/queryMessage.py

@@ -24,6 +24,14 @@ def ParseFromString(msg, data):
     def queryMessage():
         return queryMessage5_pb2.queryMessage()
 
+    def ParseFromString(msg, data):
+        return msg.ParseFromString(data)
+elif google.protobuf.__version__.split(".")[0] == "6":
+    from . import queryMessage6_pb2
+
+    def queryMessage():
+        return queryMessage6_pb2.queryMessage()
+
     def ParseFromString(msg, data):
         return msg.ParseFromString(data)
 else:

aperturedb/queryMessage6_pb2.py

@@ -4,7 +4,7 @@ dynamic = ["version"]
 description = "ApertureDB Python SDK"
 
 readme = "README.md"
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 license = {file = "LICENSE"}
 keywords = ["aperturedb", "graph", "database",
  "image", "video", "metadata", "search", "indexing"]
@@ -16,7 +16,7 @@ authors = [
 dependencies = [
     # Pin to the bridge version.
     # https://github.com/tensorflow/tensorflow/issues/60320
-    'protobuf >=3.20.3,<6.0.0',
+    'protobuf >=3.20.3,<7.0.0',
     #Folllowing is needed parallel loaders, and basic things for
     # making the notebooks.
     'requests', 'boto3',