sqlite3.py 文件源码

def cursor(self, sql: str, params: typing.Union[typing.Mapping, typing.Iterable] = None) \
            -> 'Sqlite3ResultSet':
        """
        Gets a cursor for the specified SQL.
        """

        logger.debug("Running SQL {} with params {}".format(sql, params))

        async with self._lock:
            async with threadpool():
                for stmt in separate_statements(sql):
                    cur = self.connection.cursor()
                    try:
                        if params is None:
                            cur.execute(stmt)
                        else:
                            cur.execute(stmt, params)
                    except sqlite3.OperationalError as e:
                        raise DatabaseException(*e.args)

        return Sqlite3ResultSet(cur)