base_mysql.py 文件源码

def execute(self, query, *multiparams, **params):
        """
            ???????????? ??? ??????? ???????? conn ? ????? ?????
        """
        conn = params.pop('conn')
        if conn is None:
            raise MySQLNotConnectedException('MySQL not connected')

        def coro_finished(f: asyncio.Future):
            if f.cancelled():
                self.logger.warning(
                    'execute of query is cancelled. q: %s [%s, %s]',
                    query, multiparams, params)
                return
            e = f.exception()
            if e is not None and not isinstance(e,
                                                pymysql.err.OperationalError):
                self.logger.error(
                    'Exception happened while '
                    'executing query \'%s\': %s', query, str(e),
                    exc_info=e)

        coro = asyncio.ensure_future(
            conn.execute(query, *multiparams, **params),
            loop=self.loop
        )
        coro.add_done_callback(coro_finished)
        try:
            res = await coro
            return res
        except asyncio.futures.TimeoutError as e:
            raise MySQLTimeoutError('Timeout error') from e
        except pymysql.InternalError as e:
            err_code, err_msg = e.args
            if err_code == MySQLError.DEADLOCK:
                self.logger.error(
                    'Deadlock happened while executing query %s: %s',
                    query, e.args)
                raise DeadlockError() from e
            raise e