mkashwin
diff --git a/‎.vscode/settings.json
Lines changed: 1 addition & 0 deletions b/‎.vscode/settings.json
Lines changed: 1 addition & 0 deletions
diff --git a/‎02_mqtt-cluster/src/uns_mqtt/mqtt_listener.py
Lines changed: 2 additions & 2 deletions b/‎02_mqtt-cluster/src/uns_mqtt/mqtt_listener.py
Lines changed: 2 additions & 2 deletions
diff --git a/‎02_mqtt-cluster/test/test_mqtt_listner.py
Lines changed: 7 additions & 0 deletions b/‎02_mqtt-cluster/test/test_mqtt_listner.py
Lines changed: 7 additions & 0 deletions
diff --git a/‎02_mqtt-cluster/test/test_uns_spb_enums.py
Lines changed: 2 additions & 2 deletions b/‎02_mqtt-cluster/test/test_uns_spb_enums.py
Lines changed: 2 additions & 2 deletions
diff --git a/‎02_mqtt-cluster/test/test_uns_spb_helper_metrics.py
Lines changed: 3 additions & 6 deletions b/‎02_mqtt-cluster/test/test_uns_spb_helper_metrics.py
Lines changed: 3 additions & 6 deletions
diff --git a/‎03_uns_graphdb/src/uns_graphdb/graphdb_handler.py
Lines changed: 3 additions & 3 deletions b/‎03_uns_graphdb/src/uns_graphdb/graphdb_handler.py
Lines changed: 3 additions & 3 deletions
diff --git a/‎05_sparkplugb/test/test_spb2unspublisher.py
Lines changed: 3 additions & 0 deletions b/‎05_sparkplugb/test/test_spb2unspublisher.py
Lines changed: 3 additions & 0 deletions
diff --git a/‎07_uns_graphql/schema/uns_schema.graphql
Lines changed: 99 additions & 9 deletions b/‎07_uns_graphql/schema/uns_schema.graphql
Lines changed: 99 additions & 9 deletions
diff --git a/‎07_uns_graphql/src/uns_graphql/backend/graphdb.py
Lines changed: 44 additions & 50 deletions b/‎07_uns_graphql/src/uns_graphql/backend/graphdb.py
Lines changed: 44 additions & 50 deletions
@@ -19,6 +19,7 @@
         "createdb",
         "createrole",
         "createuser",
+        "cypher",
         "DBIRTH",
         "dbuser",
         "DCMD",
 
@@ -359,9 +359,9 @@ def get_regex_for_topic_with_wildcard(topic_with_wildcard) -> str:
                 regex_exp += "[^/]*/"
             elif curr_value == "#":
                 if last_value is None:  # Handle if subscribed to #
-                    regex_exp += "(.)*/"
+                    regex_exp += "(.*)*/"
                 else:  # need to wrap the last / in optional too i.e. 'a/#' should map to just 'a' too
-                    regex_exp = regex_exp[:-1] + "(/.)*"
+                    regex_exp = regex_exp[:-1] + "(/.*)*"
             else:
                 regex_exp += curr_value + "/"
             last_value = curr_value
 
@@ -200,10 +200,17 @@ def on_connect(client, userdata, flags, result_code, properties=None):
         ("#", "a/b/c", True),
         ("a/#", "a", True),
         ("a/#", "a/b/c", True),
+        ("a/#", "a/bbb", True),
         ("a/#", "ab/a", False),
+        ("a/#/c", "ab/a", False),
+        ("a/#/c", "a/b/c", True),
+        ("aa/#/cc", "aa/abc/cc", True),
         ("a/+", "a/b", True),
+        ("a/+", "a/bbb", True),
         ("a/+/c", "a/b/c", True),
+        ("a/+/c", "a/bbb/c", True),
         ("+", "a", True),
+        ("+", "abc", True),
         ("+", "a/b/c", False),
         (None, "a/b/c", False),
         ("+", "a/b/c", False),
 
@@ -319,9 +319,9 @@ def test_dataset_via_enum(value: sparkplug_b_pb2.Payload.DataSet, spb_obj):
         ("String", sparkplug_b_pb2.Payload.Metric()),  # negative test case
     ],
 )
-def test_template_via_enum(value: sparkplug_b_pb2.Template, spb_obj):
+def test_template_via_enum(value, spb_obj):
     """
-    Test case for value setting Template via the ENUMs
+    Test case for value setting Template via the ENUMs  sparkplug_b_pb2.Template
     """
     if isinstance(value, sparkplug_b_pb2.Payload.Template):
         SPBMetricDataTypes(sparkplug_b_pb2.Template).set_value_in_sparkplug(value=value, spb_object=spb_obj)
 
@@ -318,13 +318,12 @@ def test_add_metric_and_ddata_msg(timestamp: float, metrics: list[dict]):
     sparkplug_message = SpBMessageGenerator()
     payload = sparkplug_message.get_device_data_payload(timestamp=timestamp)
     alias = 0
+    # creating the payload from the dict object provided as parameter
     for metric in metrics:
         name: str = metric["name"]
         datatype: int = metric["datatype"]
         value = metric.get("value", None)
-        metric_timestamp = metric.get("timestamp", None)
-        if metric_timestamp is None:
-            metric_timestamp = timestamp
+        metric_timestamp = metric.get("timestamp", timestamp)
         sparkplug_message.add_metric(
             payload_or_template=payload, name=name, alias=alias, datatype=datatype, value=value, timestamp=metric_timestamp
         )
@@ -594,9 +593,7 @@ def test_add_historical_metric_and_ddata_msg(metrics: list[dict]):
         name: str = metric["name"]
         datatype: int = metric["datatype"]
         value = metric.get("value", None)
-        metric_timestamp = metric.get("timestamp", None)
-        if metric_timestamp is None:
-            metric_timestamp = timestamp
+        metric_timestamp = metric.get("timestamp", timestamp)
         sparkplug_message.add_historical_metric(
             payload=payload, name=name, datatype=datatype, value=value, timestamp=metric_timestamp
         )
 
@@ -103,8 +103,8 @@ def connect(self, retry: int = 0) -> neo4j.Driver:
         ----------
         retry: int
             Optional parameters to retry making a connection in case of errors.
-            The max number of retry is `GraphDBHandler.MAX_RETRIES`
-            The time between attempts is  `GraphDBHandler.SLEEP_BTW_ATTEMPT`
+            The max number of retry is `GraphDBHandler.max_retry`
+            The time between attempts is  `GraphDBHandler.sleep_btw_attempts`
         Returns:
             neo4j.Driver: The Neo4j driver object.
 
@@ -117,7 +117,7 @@ def connect(self, retry: int = 0) -> neo4j.Driver:
         """
         try:
             if self.driver is None:
-                self.driver = neo4j.GraphDatabase.driver(self.uri, auth=self.auth)
+                self.driver = neo4j.GraphDatabase.driver(self.uri, auth=self.auth, database=self.database)
             self.driver.verify_connectivity()
         except (
             exceptions.DatabaseError,
 
@@ -166,6 +166,9 @@ def test_negative_get_metric_name(cache_key, metric):
 
 
 @pytest.mark.parametrize("cache_key", ["group1_/edge_node_1/None", "group1_/edge_node_1/MyDevice"])
+# FIXME not working with VsCode https://github.com/microsoft/vscode-python/issues/19374
+# Comment this marker and run test individually in VSCode. Uncomment for running from command line / CI
+@pytest.mark.xdist_group(name="spb_publisher")
 @pytest.mark.parametrize(
     "metric,name, alias",
     [
 
@@ -77,15 +77,102 @@ type Query {
   """
   getUnsNodes(topics: [MQTTTopicInput!]!): [UNSNode!]!
 
-  """
-  Get all UNSNodes published which have specific attributes.Option binary_operator input allows chaining the list of property_keys. If NULL, property_keys will be ORed- OR: Either of the property_keys must exist in the same node. If only one property_keys provided will be ignored- AND:  All property_keys must exist in same node. If only one property_keys provided will be ignored- NOT: None of the provided property_keys should exist in the same nodeOther criteria - topics will always be ANDed to the query filter
-  """
-  getUnsNodesByProperty(propertyKeys: [String!]!, binaryOperator: BinaryOperator, topics: [MQTTTopicInput!]): [UNSNode!]!
+  "Get all UNSNodes published which have specific attribute name as 'propertyKeys'. \n Optionally Filter results with list of  topics. \nIf topics are provided then optional boolean value 'excludeTopics' attribute can be set to true .\nto exclude the nodes which match the topics.\n"
+  getUnsNodesByProperty(propertyKeys: [String!]!, topics: [MQTTTopicInput!], excludeTopics: Boolean = false): [UNSNode!]!
 
-  """
-  Get consolidation of sparkplugB nodes. MQTT wildcards are supportedEnsure that the topics are all in the SparkplugB namespace i.e. starts with 'spBv1.0/' Format of an SpB topic is spBv1.0/ <group_id> / <message_type> / <edge_node_id> / [<device_id>]
-  """
-  getSpbNodes(topics: [MQTTTopicInput!]!): [UNSNode!]!
+  """Get all the SPBNode by the provided metric name"""
+  getSpbNodesByMetric(metricNames: [String!]!): [SPBNode!]!
+}
+
+type SPBDataSet {
+  numOfColumns: Union!
+  columns: [String!]!
+  types: [String!]!
+  rows: [SPBDataSetRow!]!
+}
+
+type SPBDataSetRow {
+  elements: [SPBDataSetValue!]!
+}
+
+type SPBDataSetValue {
+  value: SPBPrimitive!
+}
+
+type SPBMetadata {
+  isMultiPart: Boolean
+  contentType: String
+  size: Union
+  seq: Union
+  fileName: String
+  fileType: String
+  md5: String
+  description: String
+}
+
+type SPBMetric {
+  name: String!
+  alias: Union
+  timestamp: DateTime!
+  datatype: String!
+  isHistorical: Boolean
+  isTransient: Boolean
+  isNull: Boolean
+  metadata: SPBMetadata
+  properties: SPBPropertySet
+  value: SPBPrimitiveBytesPayloadSPBDataSetSPBTemplate!
+}
+
+type SPBNode {
+  topic: String!
+  timestamp: DateTime!
+  metrics: [SPBMetric!]!
+  seq: Union!
+  uuid: ID
+  body: Base64
+}
+
+"""
+Wrapper for primitive types in Sparkplug.: int, float, str, bool, list
+                 Needed because GraphQL does not support str for unions.
+                 Data is converted to its String representation for convenience.
+                 Use the datatype to convert to actual type if needed 
+"""
+type SPBPrimitive {
+  data: String!
+}
+
+union SPBPrimitiveBytesPayloadSPBDataSetSPBTemplate = SPBPrimitive | BytesPayload | SPBDataSet | SPBTemplate
+
+union SPBPrimitiveSPBPropertySetSPBPropertySetList = SPBPrimitive | SPBPropertySet | SPBPropertySetList
+
+type SPBPropertySet {
+  keys: [String!]!
+  values: [SPBPropertyValue!]!
+}
+
+type SPBPropertySetList {
+  propertysets: [SPBPropertySet!]!
+}
+
+type SPBPropertyValue {
+  isNull: Boolean
+  datatype: String!
+  value: SPBPrimitiveSPBPropertySetSPBPropertySetList!
+}
+
+type SPBTemplate {
+  version: String
+  metrics: [SPBMetric!]!
+  parameters: [SPBTemplateParameter!]
+  templateRef: String
+  isDefinition: Boolean
+}
+
+type SPBTemplateParameter {
+  name: String!
+  datatype: String!
+  value: SPBPrimitive!
 }
 
 type StreamingMessage {
@@ -112,4 +199,7 @@ type UNSNode {
   payload: JSONPayload!
   created: DateTime!
   lastUpdated: DateTime!
-}
+}
+
+"""Int 64 field since GraphQL doesn't support int64, only int 32"""
+scalar Union
@@ -18,14 +18,19 @@
 Encapsulates integration with the Graph database database
 """
 
+import asyncio
 import logging
 
-from neo4j import AsyncDriver, AsyncGraphDatabase, AsyncResult, Record
+from neo4j import AsyncDriver, AsyncGraphDatabase, Record
+from neo4j.exceptions import Neo4jError
 
 from uns_graphql.graphql_config import GraphDBConfig
 
 LOGGER = logging.getLogger(__name__)
 
+MAX_RETRIES = 5
+SLEEP_BTW_ATTEMPT = 10
+
 
 class GraphDB:
     """
@@ -37,34 +42,45 @@ class GraphDB:
     _graphdb_driver: AsyncDriver = None
 
     @classmethod
-    async def get_graphdb_driver(cls) -> AsyncDriver:
+    async def get_graphdb_driver(cls, retry: int = 0) -> AsyncDriver:
         """
         Returns the Neo4j async driver which is the connection to the database.
 
         Validates if the current driver is still connected, and if not, creates a new connection.
         The driver is cached and reused for subsequent requests to avoid creating multiple connections.
+        Parameters
+        ----------
+        retry: int
+            Optional parameters to retry making a connection in case of errors.
+            The max number of retry is `MAX_RETRIES`
+            The time between attempts is  `SLEEP_BTW_ATTEMPT` seconds
 
         Returns:
             AsyncDriver: The Neo4j async driver.
         """
         LOGGER.debug("GraphDB driver requested")
-        if cls._graphdb_driver is None:
-            LOGGER.info("Creating a new GraphDB driver")
-            cls._graphdb_driver = AsyncGraphDatabase.driver(
-                uri=GraphDBConfig.conn_url, auth=(GraphDBConfig.user, GraphDBConfig.password)
-            )
         try:
+            if cls._graphdb_driver is None:
+                LOGGER.info("Creating a new GraphDB driver")
+                cls._graphdb_driver = AsyncGraphDatabase.driver(
+                    uri=GraphDBConfig.conn_url,
+                    auth=(GraphDBConfig.user, GraphDBConfig.password),
+                    database=GraphDBConfig.database,
+                )
+
             await cls._graphdb_driver.verify_connectivity()
             LOGGER.debug("GraphDB driver connectivity verified")
-        except Exception as ex:
+        except Neo4jError as ex:
             LOGGER.error("Failed to verify GraphDB driver connectivity: %s", str(ex), stack_info=True, exc_info=True)
             # In case of connectivity failure, close the existing driver and create a new one
             await cls.release_graphdb_driver()
-            cls._graphdb_driver = AsyncGraphDatabase.driver(
-                uri=GraphDBConfig.conn_url, auth=(GraphDBConfig.user, GraphDBConfig.password)
-            )
-            await cls._graphdb_driver.verify_connectivity()
-            LOGGER.info("Reconnected to GraphDB driver")
+            if retry >= MAX_RETRIES:
+                LOGGER.error("No. of retries exceeded %s", str(MAX_RETRIES), stack_info=True, exc_info=True)
+                raise ex  # Re-raise the exception after cleanup
+
+            await asyncio.sleep(SLEEP_BTW_ATTEMPT)
+            cls._graphdb_driver = await cls.get_graphdb_driver(retry + 1)
+            LOGGER.info("Trying to Reconnect to GraphDB driver")
 
         return cls._graphdb_driver
 
@@ -79,19 +95,19 @@ async def release_graphdb_driver(cls):
             try:
                 await cls._graphdb_driver.close()
                 LOGGER.info("GraphDB driver closed successfully")
-            except Exception as ex:
+            except Neo4jError as ex:
                 LOGGER.error("Error closing the GraphDB driver: %s", str(ex), stack_info=True, exc_info=True)
             finally:
                 cls._graphdb_driver = None
         else:
             LOGGER.warning("Trying to close an already closed driver")
 
-    async def execute_read_query(self, query: str, *args, **kwargs) -> list[Record]:  # -> list:
+    async def execute_read_query(self, query: str, *args, **kwargs) -> list[Record]:
         """
-        Executes a read (CQL) query with the provided positional and keyword parameters.
+        Executes a (CQL) query with the provided positional and keyword parameters.
 
         Args:
-            query (str): The CQL query to execute.
+            query (str): The CQL query to execute. Using the auto commit feature for transaction
             *args: Positional parameters for the CQL query.
             **kwargs: Keyword parameters for the CQL query.
 
@@ -100,43 +116,21 @@ async def execute_read_query(self, query: str, *args, **kwargs) -> list[Record]:
         """
 
         LOGGER.debug("Executing query: %s with args: %s and kwargs: %s", query, args, kwargs)
-        async with self.get_graphdb_driver().session() as session:
+        driver = await self.get_graphdb_driver()
+
+        async def read(read_tx, query: str, *args, **kwargs) -> list[Record]:
+            result = await read_tx.run(query, *args, **kwargs)
+            records: list[Record] = [record async for record in result]
+            return records
+
+        async with driver.session() as session:
             try:
-                result: AsyncResult = await session.run(query, *args, **kwargs)
-                records: list[Record] = [record async for record in result]
+                records = await session.execute_read(read, query=query, *args, **kwargs)  # noqa: B026
+
                 LOGGER.debug("Query executed successfully, retrieved records: %s", records)
                 return records
             except Exception as ex:
                 LOGGER.error("Error executing query: %s", str(ex), stack_info=True, exc_info=True)
                 raise
             finally:
-                result.aclose()
-
-    async def execute_write_query(self, query: str, *args, **kwargs):
-        """
-        Executes a write (CQL) query with the provided positional and keyword parameters.
-        Supports transactional operations.
-
-        Args:
-            query (str): The CQL query to execute.
-            *args: Positional parameters for the CQL query.
-            **kwargs: Keyword parameters for the CQL query.
-
-        Returns:
-            list: The results of the query.
-        """
-
-        LOGGER.debug("Executing write query: %s with args: %s and kwargs: %s", query, args, kwargs)
-        async with self.get_graphdb_driver().session() as session:
-            try:
-                async with session.begin_transaction() as tx:
-                    result = await tx.run(query, *args, **kwargs)
-                    await tx.commit()
-                    records = [record async for record in result]
-                    LOGGER.debug("Write query executed successfully, committed records: %s", records)
-                    return records
-            except Exception as ex:
-                LOGGER.error("Error executing write query: %s", str(ex), stack_info=True, exc_info=True)
-                raise
-            finally:
-                result.aclose()
+                await session.close()
Original file line number	Diff line number	Diff line change
`@@ -166,6 +166,9 @@ def test_negative_get_metric_name(cache_key, metric):`
`166`	`166`
`167`	`167`
`168`	`168`	`@pytest.mark.parametrize("cache_key", ["group1_/edge_node_1/None", "group1_/edge_node_1/MyDevice"])`
	`169`	`+# FIXME not working with VsCode https://github.com/microsoft/vscode-python/issues/19374`
	`170`	`+# Comment this marker and run test individually in VSCode. Uncomment for running from command line / CI`
	`171`	`+@pytest.mark.xdist_group(name="spb_publisher")`
`169`	`172`	`@pytest.mark.parametrize(`
`170`	`173`	`"metric,name, alias",`
`171`	`174`	`[`