Removed implicit query fallback execution from query_database

Author: alexthomas93

src/neo4j_graphrag/schema.py

@@ -18,7 +18,7 @@
 
 import neo4j
 from neo4j import Query
-from neo4j.exceptions import ClientError, CypherTypeError, Neo4jError
+from neo4j.exceptions import ClientError, CypherTypeError
 
 BASE_KG_BUILDER_LABEL = "__KGBuilder__"
 BASE_ENTITY_LABEL = "__Entity__"
@@ -161,38 +161,16 @@ def query_database(
         list[dict[str, Any]]: the result of the query in json format.
     """
     if not session_params:
-        try:
-            data = driver.execute_query(
-                Query(text=query, timeout=timeout),
-                database_=database,
-                parameters_=params,
-            )
-            json_data = [r.data() for r in data.records]
-            if sanitize:
-                json_data = [_value_sanitize(el) for el in json_data]
-            return json_data
-        except Neo4jError as e:
-            if not (
-                (
-                    (  # isCallInTransactionError
-                        e.code == "Neo.DatabaseError.Statement.ExecutionFailed"
-                        or e.code
-                        == "Neo.DatabaseError.Transaction.TransactionStartFailed"
-                    )
-                    and e.message is not None
-                    and "in an implicit transaction" in e.message
-                )
-                or (  # isPeriodicCommitError
-                    e.code == "Neo.ClientError.Statement.SemanticError"
-                    and e.message is not None
-                    and (
-                        "in an open transaction is not possible" in e.message
-                        or "tried to execute in an explicit transaction" in e.message
-                    )
-                )
-            ):
-                raise
-    # Fallback to allow implicit transactions
+        data = driver.execute_query(
+            Query(text=query, timeout=timeout),
+            database_=database,
+            parameters_=params,
+        )
+        json_data = [r.data() for r in data.records]
+        if sanitize:
+            json_data = [_value_sanitize(el) for el in json_data]
+        return json_data
+
     session_params.setdefault("database", database)
     with driver.session(**session_params) as session:
         result = session.run(Query(text=query, timeout=timeout), params)

tests/unit/test_schema.py

@@ -19,7 +19,6 @@
 
 import pytest
 from neo4j import Driver, Query
-from neo4j.exceptions import Neo4jError
 from neo4j_graphrag.schema import (
     BASE_ENTITY_LABEL,
     BASE_KG_BUILDER_LABEL,
@@ -30,12 +29,11 @@
     NODE_PROPERTIES_QUERY,
     REL_PROPERTIES_QUERY,
     REL_QUERY,
+    _value_sanitize,
     format_schema,
     get_enhanced_schema_cypher,
     get_schema,
     get_structured_schema,
-    query_database,
-    _value_sanitize,
 )
 
 
@@ -212,34 +210,6 @@ def test__value_sanitize(
     ), f"Failed test case: {description}"
 
 
-def test_query_fallback_execution(driver: MagicMock) -> None:
-    """Test the fallback to allow for implicit transactions in query."""
-    err = Neo4jError()
-    err.code = "Neo.DatabaseError.Statement.ExecutionFailed"
-    err.message = "in an implicit transaction"
-    driver.execute_query.side_effect = err
-    mock_session = MagicMock()
-    mock_result = MagicMock()
-    mock_result.data.return_value = {
-        "key1": "value1",
-        "oversized_list": list(range(LIST_LIMIT + 1)),
-    }
-    mock_session.run.return_value = [mock_result]
-    driver.session.return_value.__enter__.return_value = mock_session
-    driver.session.return_value.__exit__.return_value = None
-    query = "MATCH (n) RETURN n;"
-    params = {"param1": "value1"}
-    json_data = query_database(
-        driver=driver, query=query, database="test_db", params=params, sanitize=True
-    )
-    driver.session.assert_called_with(database="test_db")
-    called_args, _ = mock_session.run.call_args
-    called_query = called_args[0]
-    assert called_query.text == query
-    assert called_args[1] == params
-    assert json_data == [{"key1": "value1"}]
-
-
 @pytest.mark.parametrize(
     "description, schema, is_enhanced, expected_output",
     [