def execute()

in ossdbtoolsservice/query/batch.py [0:0]

• 23 lines of code
• 15 McCabe index (conditional complexity)

    def execute(self, conn: ServerConnection) -> None:
        """
        Execute the batch using a cursor retrieved from the given connection

        :raises DatabaseError: if an error is encountered while running the batch's query
        """
        self._execution_start_time = datetime.now()

        if self._batch_events and self._batch_events._on_execution_started:
            self._batch_events._on_execution_started(self)

        cursor = self.get_cursor(conn)
        try:
            cursor.execute(self.batch_text)

            # Commit the transaction if autocommit is True
            if conn.autocommit:
                conn.commit()

            self.after_execute(cursor)
        except conn.database_error as error:
            self._has_error = True
            raise error
        finally:
            # We are doing this because when the execute fails for named cursors
            # cursor is not activated on the server which results in failure on close
            # Hence we are checking if the cursor was really executed for us to close it
            if cursor and cursor.rowcount != -1 and cursor.rowcount is not None:
                cursor.close()
            self._has_executed = True
            self._execution_end_time = datetime.now()

            # TODO: PyMySQL doesn't support notices from a connection
            if conn._provider_name == PG_PROVIDER_NAME:
                self._notices = cursor.connection.notices
                cursor.connection.notices = []

            if self._batch_events and self._batch_events._on_execution_completed:
                self._batch_events._on_execution_completed(self)