Fix Neo4jOperator templated query validation

providers/neo4j/src/airflow/providers/neo4j/operators/neo4j.py

@@ -63,16 +63,20 @@ def __init__(
                 AirflowProviderDeprecationWarning,
                 stacklevel=2,
             )
-            if cypher is not None:
-                raise ValueError("Cannot provide both `sql` and `cypher`. Use `cypher` only.")
-            cypher = sql
-        if cypher is None:
-            raise ValueError("Parameter `cypher` is required.")
         self.neo4j_conn_id = neo4j_conn_id
         self.cypher = cypher
+        self.sql = sql
         self.parameters = parameters
 
     def execute(self, context: Context) -> None:
-        self.log.info("Executing: %s", self.cypher)
+        cypher = self.cypher
+        if self.sql is not None:
+            if cypher is not None:
+                raise ValueError("Cannot provide both `sql` and `cypher`. Use `cypher` only.")
+            cypher = self.render_template(self.sql, context)
+        if cypher is None:
+            raise ValueError("Parameter `cypher` is required.")
+
+        self.log.info("Executing: %s", cypher)
         hook = Neo4jHook(conn_id=self.neo4j_conn_id)
-        hook.run(self.cypher, self.parameters)
+        hook.run(cypher, self.parameters)

providers/neo4j/tests/unit/neo4j/operators/test_neo4j.py

@@ -61,16 +61,20 @@ def test_neo4j_operator_sql_param_is_deprecated(self, mock_hook):
             AirflowProviderDeprecationWarning,
             match="`sql` parameter is deprecated, please use `cypher` instead.",
         ):
-            op = Neo4jOperator(task_id="basic_neo4j", sql=cypher)
-        assert op.cypher == cypher
-        op.execute(mock.MagicMock())
+            op = Neo4jOperator(task_id="basic_neo4j", sql="{{ cypher }}")
+        assert op.sql == "{{ cypher }}"
+        op.execute({"cypher": cypher})
         mock_hook.return_value.run.assert_called_once_with(cypher, None)
 
-    def test_neo4j_operator_both_sql_and_cypher_raises(self):
+    def test_neo4j_operator_both_sql_and_cypher_raises_on_execute(self):
         with pytest.warns(AirflowProviderDeprecationWarning):
-            with pytest.raises(ValueError, match="Cannot provide both `sql` and `cypher`"):
-                Neo4jOperator(task_id="basic_neo4j", sql="a", cypher="b")
+            op = Neo4jOperator(task_id="basic_neo4j", sql="a", cypher="b")
+
+        with pytest.raises(ValueError, match="Cannot provide both `sql` and `cypher`"):
+            op.execute(mock.MagicMock())
+
+    def test_neo4j_operator_missing_cypher_raises_on_execute(self):
+        op = Neo4jOperator(task_id="basic_neo4j")
 
-    def test_neo4j_operator_missing_cypher_raises(self):
         with pytest.raises(ValueError, match="Parameter `cypher` is required."):
-            Neo4jOperator(task_id="basic_neo4j")
+            op.execute(mock.MagicMock())

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -77,7 +77,6 @@ providers/microsoft/azure/src/airflow/providers/microsoft/azure/sensors/compute.
 providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py::GCSToAzureBlobStorageOperator
 providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/oracle_to_azure_data_lake.py::OracleToAzureDataLakeOperator
 providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py::PsrpOperator
-providers/neo4j/src/airflow/providers/neo4j/operators/neo4j.py::Neo4jOperator
 providers/oracle/src/airflow/providers/oracle/transfers/oracle_to_oracle.py::OracleToOracleOperator
 providers/papermill/src/airflow/providers/papermill/operators/papermill.py::PapermillOperator
 providers/snowflake/src/airflow/providers/snowflake/operators/snowpark_containers.py::SnowparkContainerJobOperator