How to use the apsw.Error function in apsw

return
        with self.library_broker:
            default_prefs = None
            try:
                olddb = self.library_view.model().db
                if copy_structure:
                    default_prefs = olddb.prefs
            except:
                olddb = None
            if copy_structure and olddb is not None and default_prefs is not None:
                default_prefs['field_metadata'] = olddb.new_api.field_metadata.all_metadata()
            db = self.library_broker.prepare_for_gui_library_change(newloc)
            if db is None:
                try:
                    db = LibraryDatabase(newloc, default_prefs=default_prefs)
                except apsw.Error:
                    if not allow_rebuild:
                        raise
                    import traceback
                    repair = question_dialog(self, _('Corrupted database'),
                            _('The library database at %s appears to be corrupted. Do '
                            'you want calibre to try and rebuild it automatically? '
                            'The rebuild may not be completely successful.')
                            % force_unicode(newloc, filesystem_encoding),
                            det_msg=traceback.format_exc()
                            )
                    if repair:
                        from calibre.gui2.dialogs.restore_library import repair_library_at
                        if repair_library_at(newloc, parent=self):
                            db = LibraryDatabase(newloc, default_prefs=default_prefs)
                        else:
                            return

from contextvars import ContextVar
from functools import wraps
from dataclasses import dataclass

from torba.client.basedatabase import query, interpolate

from lbry.schema.url import URL, normalize_name
from lbry.schema.tags import clean_tags
from lbry.schema.result import Outputs
from lbry.wallet.ledger import BaseLedger, MainNetLedger, RegTestLedger

from .common import CLAIM_TYPES, STREAM_TYPES, COMMON_TAGS
from .full_text_search import FTS_ORDER_BY


class SQLiteOperationalError(apsw.Error):
    def __init__(self, metrics):
        super().__init__('sqlite query errored')
        self.metrics = metrics


class SQLiteInterruptedError(apsw.InterruptError):
    def __init__(self, metrics):
        super().__init__('sqlite query interrupted')
        self.metrics = metrics


ATTRIBUTE_ARRAY_MAX_LENGTH = 100

INTEGER_PARAMS = {
    'height', 'creation_height', 'activation_height', 'expiration_height',
    'timestamp', 'creation_timestamp', 'release_time', 'fee_amount',

def run_query(self, token, query, values, error_handler):
        """Run a query.

        Args:
            token: A uuid object of the query you want returned.
            query: A sql query with ? placeholders for values.
            values: A tuple of values to replace "?" in query.
            error_handler: A function to handle error. None executes the default.
        """
        if query.lower().strip().startswith("select"):
            try:
                self.sqlite3_cursor.execute(query, values)
                self.results[token] = self.sqlite3_cursor.fetchall()
            except apsw.Error as err:
                # Put the error into the output queue since a response
                # is required.
                self.results[token] = ("Query returned error: %s: %s: %s" % (query, values, err))
                if error_handler is None:
                    self.logger.error("Query returned error: %s: %s: %s".format(query, values, err))
                else:
                    error_handler(self, query, values, err)
        else:
            try:
                self.sqlite3_cursor.execute("begin")
                self.sqlite3_cursor.execute(query, values)
                self.sqlite3_cursor.execute("commit")
            except apsw.Error as err:
                self.results[token] = err
                self.sqlite3_cursor.execute("rollback")
                if error_handler is None:

return str(d)

                    # We are not allowed to modify the query itself, so we're forced to truncate
                    # long lists of results with Python.
                    for _ in range(1000):
                        row = c.fetchone()
                        if not row:
                            break
                        else:
                            query_headers = c.getdescription()
                        query_rows.append(map(stringify, row))
                    if c.fetchone():
                        query_more = True
                    if not query_rows:
                        query_error = "No results"
        except apsw.Error:
            query_error = traceback.format_exc()

        if action == "export" and query_headers:
            result_string = StringIO.StringIO()
            result_writer = csv.writer(result_string, delimiter=",", quotechar='"')
            result_writer.writerow([header_name for header_name, header_type in query_headers])
            if query_rows:
                result_writer.writerows(query_rows)
            resp = Response(result_string.getvalue())
            resp.headers["Content-Type"] = "text/csv"
            resp.headers["Content-Disposition"] = "attachment; filename=query_results.csv"
            return resp

    return render_template("ta/sql.html",
                           query=query,
                           query_headers=query_headers,

def initialize_db(self):
        from calibre.db.legacy import LibraryDatabase
        db = None
        self.timed_print('Initializing db...')
        try:
            db = LibraryDatabase(self.library_path)
        except apsw.Error:
            with self.app:
                self.hide_splash_screen()
                repair = question_dialog(self.splash_screen, _('Corrupted database'),
                        _('The library database at %s appears to be corrupted. Do '
                        'you want calibre to try and rebuild it automatically? '
                        'The rebuild may not be completely successful. '
                        'If you say No, a new empty calibre library will be created.')
                        % force_unicode(self.library_path, filesystem_encoding),
                        det_msg=traceback.format_exc()
                        )
            if repair:
                if iswindows:
                    # On some windows systems the existing db file gets locked
                    # by something when running restore from the main process.
                    # So run the restore in a separate process.
                    windows_repair(self.library_path)

def initialize_db(self):
        from calibre.db.legacy import LibraryDatabase
        db = None
        self.timed_print('Initializing db...')
        try:
            db = LibraryDatabase(self.library_path)
        except apsw.Error:
            with self.app:
                self.hide_splash_screen()
                repair = question_dialog(self.splash_screen, _('Corrupted database'),
                        _('The library database at %s appears to be corrupted. Do '
                        'you want calibre to try and rebuild it automatically? '
                        'The rebuild may not be completely successful. '
                        'If you say No, a new empty calibre library will be created.')
                        % force_unicode(self.library_path, filesystem_encoding),
                        det_msg=traceback.format_exc()
                        )
            if repair:
                if iswindows:
                    # On some windows systems the existing db file gets locked
                    # by something when running restore from the main process.
                    # So run the restore in a separate process.
                    windows_repair(self.library_path)

return
        with self.library_broker:
            default_prefs = None
            try:
                olddb = self.library_view.model().db
                if copy_structure:
                    default_prefs = olddb.prefs
            except:
                olddb = None
            if copy_structure and olddb is not None and default_prefs is not None:
                default_prefs['field_metadata'] = olddb.new_api.field_metadata.all_metadata()
            db = self.library_broker.prepare_for_gui_library_change(newloc)
            if db is None:
                try:
                    db = LibraryDatabase(newloc, default_prefs=default_prefs)
                except apsw.Error:
                    if not allow_rebuild:
                        raise
                    import traceback
                    repair = question_dialog(self, _('Corrupted database'),
                            _('The library database at %s appears to be corrupted. Do '
                            'you want calibre to try and rebuild it automatically? '
                            'The rebuild may not be completely successful.')
                            % force_unicode(newloc, filesystem_encoding),
                            det_msg=traceback.format_exc()
                            )
                    if repair:
                        from calibre.gui2.dialogs.restore_library import repair_library_at
                        if repair_library_at(newloc, parent=self):
                            db = LibraryDatabase(newloc, default_prefs=default_prefs)
                        else:
                            return

def run_query(self, token, query, values, error_handler):
        """Run a query.

        Args:
            token: A uuid object of the query you want returned.
            query: A sql query with ? placeholders for values.
            values: A tuple of values to replace "?" in query.
            error_handler: A function to handle error. None executes the default.
        """
        if query.lower().strip().startswith("select"):
            try:
                self.sqlite3_cursor.execute(query, values)
                self.results[token] = self.sqlite3_cursor.fetchall()
            except apsw.Error as err:
                # Put the error into the output queue since a response
                # is required.
                self.results[token] = ("Query returned error: %s: %s: %s" % (query, values, err))
                if error_handler is None:
                    self.logger.error("Query returned error: %s: %s: %s".format(query, values, err))
                else:
                    error_handler(self, query, values, err)
        else:
            try:
                self.sqlite3_cursor.execute("begin")
                self.sqlite3_cursor.execute(query, values)
                self.sqlite3_cursor.execute("commit")
            except apsw.Error as err:
                self.results[token] = err
                self.sqlite3_cursor.execute("rollback")
                if error_handler is None:

def execute(self, sql, args=()):
        try:
            t = time.time()
            try:
                return self._cursor.execute(sql, args)
            finally:
                log.msg(interface=iaxiom.IStatEvent,
                        stat_cursor_execute_time=time.time() - t)
        except apsw.Error, e:
            raise errors.SQLError(sql, args, e)

def execute_query(sql, values) -> List:
    context = ctx.get()
    context.set_query_timeout()
    try:
        return context.db.cursor().execute(sql, values).fetchall()
    except apsw.Error as err:
        plain_sql = interpolate(sql, values)
        if context.is_tracking_metrics:
            context.metrics['execute_query'][-1]['sql'] = plain_sql
        if isinstance(err, apsw.InterruptError):
            context.log.warning("interrupted slow sqlite query:\n%s", plain_sql)
            raise SQLiteInterruptedError(context.metrics)
        context.log.exception('failed running query', exc_info=err)
        raise SQLiteOperationalError(context.metrics)