def get_conn()

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

• 24 lines of code
• 4 McCabe index (conditional complexity)

    def get_conn(self, dbname=""):
        """
        Create the connection to MySQL instance, there will be only one
        connection during the whole schema change
        """
        try:
            conn = db_lib.MySQLSocketConnection(
                self.mysql_user,
                self.mysql_pass,
                self.socket,
                dbname,
                connect_function=self.get_conn_func,
                charset=self.charset,
            )
            if conn:
                conn.connect()
                if self.session_timeout:
                    conn.execute(
                        "SET SESSION wait_timeout = {}".format(self.session_timeout)
                    )
                return conn
        except MySQLdb.MySQLError as e:
            errcode, errmsg = e.args
            log.error("Error when connecting to MySQL [{}] {}".format(errcode, errmsg))
            raise OSCError(
                "GENERIC_MYSQL_ERROR",
                {"stage": "Connecting to MySQL", "errnum": errcode, "errmsg": errmsg},
            )