in aws_advanced_python_wrapper/states/session_state_service.py [0:0]
def apply_pristine_session_state(self, new_connection: Connection):
if not self._transfer_state_enabled_setting():
return
func: Optional[Callable] = SessionStateTransferHandlers.get_transfer_session_state_on_switch_func()
if func is not None:
is_handled: bool = func(self._session_state, new_connection)
if is_handled:
# Custom function has handled session transfer
return
if self._copy_session_state is None:
return
if self._copy_session_state.auto_commit.can_restore_pristine():
try:
self._plugin_service.driver_dialect.set_autocommit(new_connection, self._copy_session_state.auto_commit.pristine_value)
except Exception:
# Ignore any exception.
pass
if self._copy_session_state.readonly.can_restore_pristine():
try:
self._plugin_service.driver_dialect.set_read_only(new_connection, self._copy_session_state.readonly.pristine_value)
except Exception:
# Ignore any exception.
pass