diff --git a/neutron/common/ovn/utils.py b/neutron/common/ovn/utils.py
index 2ccc89d50da..482f6a128fd 100644
--- a/neutron/common/ovn/utils.py
+++ b/neutron/common/ovn/utils.py
@@ -12,6 +12,7 @@
 
 import collections
 import copy
+import functools
 import inspect
 import os
 import random
@@ -37,6 +38,7 @@ from oslo_utils import netutils
 from oslo_utils import strutils
 from ovsdbapp.backend.ovs_idl import idlutils
 from ovsdbapp import constants as ovsdbapp_const
+from pecan import util as pecan_util
 import tenacity
 
 from neutron._i18n import _
@@ -131,6 +133,68 @@ class OvsdbClientTransactCommand(OvsdbClientCommand):
     COMMAND = 'transact'
 
 
+def ovn_context(txn_var_name='txn', idl_var_name='idl'):
+    """Provide an OVN IDL transaction context
+
+    This decorator provides an OVN IDL database transaction context if the
+    'txn_var_name' variable, that should have an
+    ``ovsdbapp.backend.ovs_idl.transaction.Transaction`` derived object, is
+    empty. In that case (an empty transaction), that means the decorated method
+    has been called outside a transaction. In this case, the decorator creates
+    a transaction from the provided IDL and assigns it to the 'txn_var_name'
+    variable.
+    """
+    def decorator(f):
+        signature = inspect.signature(f)
+        if (txn_var_name not in signature.parameters or
+                idl_var_name not in signature.parameters):
+            msg = (_('Could not find variables %s and %s in the method '
+                     'signature') %
+                   (txn_var_name, idl_var_name))
+            raise RuntimeError(msg)
+
+        def retrieve_parameter(param_name, _args, _kwargs):
+            # Position of the parameter "param_name" in the "args" tuple.
+            param_index = pecan_util.getargspec(f).args.index(param_name)
+            try:  # Parameter passed as a positional argument.
+                value = _args[param_index]
+            except IndexError:  # Parameter passed as keyword argument.
+                # Reset the "param_index" value, that means the parameter is
+                # passed as keyword and if needed, it will be replaced in
+                # "kwargs".
+                param_index = None
+                try:
+                    value = _kwargs[param_name]
+                except KeyError:
+                    # Parameter is not passed as keyword nor positional, read
+                    # the keyword default value.
+                    value = signature.parameters[param_name].default
+
+            return value, param_index
+
+        @functools.wraps(f)
+        def wrapped(*args, **kwargs):
+            _txn, txn_index = retrieve_parameter(txn_var_name, args, kwargs)
+            _idl, idl_index = retrieve_parameter(idl_var_name, args, kwargs)
+            if not _txn and not _idl:
+                msg = (_('If no transaction is defined, it is needed at least '
+                         'an IDL connection'))
+                raise RuntimeError(msg)
+
+            if not _txn:
+                with _idl.transaction(check_error=True) as new_txn:
+                    if txn_index:
+                        args = (args[:txn_index] + (new_txn,) +
+                                args[txn_index + 1:])
+                    else:
+                        kwargs[txn_var_name] = new_txn
+                    return f(*args, **kwargs)
+            else:
+                return f(*args, **kwargs)
+        return wrapped
+    return decorator
+
+
 def ovn_name(id):
     # The name of the OVN entry will be neutron-<UUID>
     # This is due to the fact that the OVN application checks if the name
diff --git a/neutron/tests/functional/common/ovn/test_utils.py b/neutron/tests/functional/common/ovn/test_utils.py
index 88d16c11988..509e4c6c3af 100644
--- a/neutron/tests/functional/common/ovn/test_utils.py
+++ b/neutron/tests/functional/common/ovn/test_utils.py
@@ -12,7 +12,9 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import ddt
 from neutron_lib.api.definitions import portbindings
+from oslo_utils import uuidutils
 from ovsdbapp.backend.ovs_idl import idlutils
 
 from neutron.common.ovn import constants as ovn_const
@@ -192,3 +194,114 @@ class TestSyncHaChassisGroup(base.TestOVNFunctionalBase):
             idlutils.RowNotFound,
             self.nb_api.ha_chassis_group_get(hcg_name).execute,
             check_error=True)
+
+
+@utils.ovn_context()
+def method_with_idl_and_default_txn(ls_name, idl, txn=None):
+    txn.add(idl.ls_add(ls_name))
+
+
+@utils.ovn_context()
+def method_with_txn_and_default_idl(ls_name, txn, idl=None):
+    # NOTE(ralonsoh): the test with the default "idl" cannot be executed. A
+    # default value should be provided in a non-testing implementation.
+    txn.add(idl.ls_add(ls_name))
+
+
+@utils.ovn_context()
+def method_with_idl_and_txn(ls_name, idl, txn):
+    txn.add(idl.ls_add(ls_name))
+
+
+@utils.ovn_context(txn_var_name='custom_txn', idl_var_name='custom_idl')
+def method_with_custom_idl_and_custom_txn(ls_name, custom_idl, custom_txn):
+    custom_txn.add(custom_idl.ls_add(ls_name))
+
+
+@utils.ovn_context()
+def update_ls(ls_name, idl, txn):
+    txn.add(idl.db_set('Logical_Switch', ls_name,
+                       ('external_ids', {'random_key': 'random_value'})
+                       )
+            )
+
+
+@ddt.ddt()
+class TestOvnContext(base.TestOVNFunctionalBase):
+
+    scenarios = (
+        {'name': 'idl_and_default_txn',
+         'method': method_with_idl_and_default_txn,
+         '_args': ['ls_name', 'idl'], '_kwargs': ['txn']},
+        {'name': 'idl_and_default_txn__positional_txn',
+         'method': method_with_idl_and_default_txn,
+         '_args': ['ls_name', 'idl', 'txn'], '_kwargs': []},
+        {'name': 'idl_and_default_txn__default_txn',
+         'method': method_with_idl_and_default_txn,
+         '_args': ['ls_name', 'idl'], '_kwargs': []},
+
+        {'name': 'txn_and_default_idl',
+         'method': method_with_txn_and_default_idl,
+         '_args': ['ls_name', 'txn'], '_kwargs': ['idl']},
+        {'name': 'txn_and_default_idl__positional_idl',
+         'method': method_with_txn_and_default_idl,
+         '_args': ['ls_name', 'txn', 'idl'], '_kwargs': []},
+
+        {'name': 'txn_and_idl',
+         'method': method_with_idl_and_txn,
+         '_args': ['ls_name', 'idl', 'txn'], '_kwargs': []},
+
+        {'name': 'custom_idl_and_custom_txn',
+         'method': method_with_custom_idl_and_custom_txn,
+         '_args': ['ls_name', 'custom_idl', 'custom_txn'], '_kwargs': []},
+    )
+
+    scenarios2 = (
+        {'name': method_with_idl_and_default_txn.__name__,
+         'method': method_with_idl_and_default_txn},
+        {'name': method_with_txn_and_default_idl.__name__,
+         'method': method_with_txn_and_default_idl},
+        {'name': method_with_idl_and_txn.__name__,
+         'method': method_with_idl_and_txn},
+        {'name': method_with_custom_idl_and_custom_txn.__name__,
+         'method': method_with_custom_idl_and_custom_txn},
+    )
+
+    @ddt.unpack
+    @ddt.named_data(*scenarios)
+    def test_with_transaction(self, method, _args, _kwargs):
+        ls_name = uuidutils.generate_uuid()
+        custom_idl = idl = self.nb_api
+        with self.nb_api.transaction(check_error=True) as txn:
+            custom_txn = txn
+            _locals = locals()
+            args = [_locals[_arg] for _arg in _args]
+            kwargs = {_kwarg: _locals[_kwarg] for _kwarg in _kwargs}
+            # Create a LS and update it.
+            method(*args, **kwargs)
+            update_ls(ls_name, self.nb_api, txn)
+
+        ls = self.nb_api.lookup('Logical_Switch', ls_name)
+        self.assertEqual('random_value', ls.external_ids['random_key'])
+
+    @ddt.unpack
+    @ddt.named_data(*scenarios)
+    def test_without_transaction(self, method, _args, _kwargs):
+        ls_name = uuidutils.generate_uuid()
+        custom_idl = idl = self.nb_api
+        custom_txn = txn = None
+        _locals = locals()
+        args = [_locals[_arg] for _arg in _args]
+        kwargs = {_kwarg: _locals[_kwarg] for _kwarg in _kwargs}
+        # Create a LS and update it.
+        method(*args, **kwargs)
+        update_ls(ls_name, self.nb_api, txn)
+
+        ls = self.nb_api.lookup('Logical_Switch', ls_name)
+        self.assertEqual('random_value', ls.external_ids['random_key'])
+
+    @ddt.unpack
+    @ddt.named_data(*scenarios2)
+    def test_needed_parameters(self, method):
+        self.assertRaises(RuntimeError, method, uuidutils.generate_uuid(),
+                          None, None)
diff --git a/neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_impl_idl.py b/neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_impl_idl.py
index d0bbfebadc0..0ff869c4cef 100644
--- a/neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_impl_idl.py
+++ b/neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_impl_idl.py
@@ -590,6 +590,24 @@ class TestNbApi(BaseOvnIdlTest):
                 "BFD",
                 bfd_uuid)
 
+    def test_set_lsp_ha_chassis_group(self):
+        with self.nbapi.transaction(check_error=True) as txn:
+            ls_name = uuidutils.generate_uuid()
+            lsp_name = uuidutils.generate_uuid()
+            hcg_name = uuidutils.generate_uuid()
+            txn.add(self.nbapi.ls_add(ls_name))
+            txn.add(self.nbapi.lsp_add(ls_name, lsp_name))
+
+        with self.nbapi.transaction(check_error=True) as txn:
+            hcg = self.nbapi.ha_chassis_group_add(hcg_name)
+            txn.add(hcg)
+            lsp = self.nbapi.lookup('Logical_Switch_Port', lsp_name)
+            txn.add(self.nbapi.set_lswitch_port(lsp_name,
+                                                ha_chassis_group=hcg))
+
+        lsp = self.nbapi.lookup('Logical_Switch_Port', lsp_name)
+        self.assertEqual(hcg.result.uuid, lsp.ha_chassis_group[0].uuid)
+
 
 class TestIgnoreConnectionTimeout(BaseOvnIdlTest):
     @classmethod