def get_long_trx()

in core/lib/payload/copy.py [0:0]

• 20 lines of code
• 11 McCabe index (conditional complexity)

    def get_long_trx(self):
        """
        Return a long running transaction agaisnt the table we'll touch,
        if there's one.
        This is mainly for safety as long running transaction may block DDL,
        thus blocks more other requests
        """
        if self.skip_long_trx_check:
            return False
        processes = self.query(sql.show_processlist)
        for proc in processes:
            if not proc["Info"]:
                sql_statement = ""
            else:
                if isinstance(proc["Info"], bytes):
                    sql_statement = proc["Info"].decode("utf-8", "replace")
                else:
                    sql_statement = proc["Info"]

            proc["Info"] = sql_statement
            # Time can be None if the connection is in "Connect" state
            if (
                (proc.get("Time") or 0) > self.long_trx_time
                and proc.get("db", "") == self._current_db
                and self.table_name in "--" + sql_statement
                and not proc.get("Command", "") == "Sleep"
            ):
                return proc