传递给函数时 Sqlite 执行中的变量 (Python3)

Question

一段时间以来，我一直在研究如何从不同的线程 read/write 到 sqlite 数据库，我找到了很多答案和文档来描述需要发生的事情，但是我离实现目标还差得很远需要，所以我决定使用现有的 class 我发现别人制作的。

不好意思承认，但弄清楚如何让这个 class 工作已经花了几个小时，尽管我现在不知道为什么但是我无法在执行函数中获取变量

我通常会这样做：

c.execute("SELECT codeID FROM users WHERE codeID=:code", {'code':tag_attempt})

这工作正常但是当我用多线程尝试同样的事情时 class 它不会工作（class 有一个 'select' 函数获取和 returns数据，这是我能够从数据库中获取数据的唯一方法，但是执行函数有完全相同的问题）（我也试过这种使用变量的方法）

for q in sql.select(("select codeID from users where codeID=?", (tag_attempt)), 0):
    print(q)
# TypeError: 'NoneType' object is not iterable

这是使用的class，但我还会包括link

class MultiThreadOK(Thread):
    def __init__(self, db):
        super(MultiThreadOK, self).__init__()
        self.db = db
        self.reqs = Queue()
        self.start()

    def run(self):
        cnx = sqlite3.Connection(self.db)
        cursor = cnx.cursor()
        while True:
            req = self.reqs.get()
            if req == '--close--':
                break
            elif req == '--commit--':
                cnx.commit()
            try:
                cursor.executescript(
                    req) if ';' in req else cursor.execute(req)
            except sqlite3.OperationalError as err:
                self.escribir_error(
                    '{0} - Error {1}\n'.format(datetime.now(), err))
                self.escribir_error('{0} - {1}\n'.format(datetime.now(), req))
            except:
                self.escribir_error('{0} - Salida'.format(datetime.now()))
        cnx.close()

    def execute(self, req):
        self.reqs.put(req)

    def queries(self):
        return self.reqs.qsize()

    def empty(self):
        return self.reqs.empty()

    def select(self, req, results=0):
        cnx = sqlite3.Connection(self.db)
        cursor = cnx.cursor()
        try:
            if results == 0:
                cursor.execute(req)
                ret = [x for x in cursor.fetchall()]
                cnx.close()
                return ret
            else:
                cursor.execute(req)
                ret = [x for x in cursor.fetchall()[:results]]
                cnx.close()
                return ret
        except:
            print("Unexpected error: {0}".format(sys.exc_info()[0]))
            cnx.close()

    def commit(self):
        self.execute("--commit--")

    def close(self):
        self.execute('--close--')

    def escribir_error(self, texto):
        #with open(os.path.dirname(os.path.abspath(__file__)) + '\errores.txt', 'a') as archivo:
        #    archivo.write(texto)
        print(texto)

总结我希望能够在一个单独的线程中获取数据，这可以通过 class 我只是无法在以下位置包含变量任何阶段

信用 https://gist.github.com/User001501/3053f26100ddf281600668fed347e518

Answer 1

执行方法接收单个参数。

看起来你可以像这样使用字符串命名占位符

c.execute("SELECT codeID FROM users WHERE codeID='{code}'".format(**{'code': ag_attempt})

Answer 2

我找到了一个解决方案并想 post 以防将来它对任何人都有好处

我在 class 中将一个只是字典的新参数传递给函数，我将对更改的行进行评论。这个新代码适用于我在问题中尝试使用的行，因此函数的输入是相同的并且像常规 sqlite 一样工作。

class MultiThreadOK(Thread):
    def __init__(self, db):
        super(MultiThreadOK, self).__init__()
        self.db = db
        self.reqs = Queue()
        self.start()

    def run(self):
        cnx = sqlite3.Connection(self.db)
        cursor = cnx.cursor()
        while True:
            req, arg, res = self.reqs.get() # Add arg and res here
            if req == '--close--':
                break
            elif req == '--commit--':
                cnx.commit()
            try:
                cursor.executescript(
                    req, arg) if ';' in req else cursor.execute(req, arg) # Add arg in two places
            except sqlite3.OperationalError as err:
                self.escribir_error(
                    '{0} - Error {1}\n'.format(datetime.now(), err))
                self.escribir_error('{0} - {1}\n'.format(datetime.now(), req))
            except:
                self.escribir_error('{0} - Salida'.format(datetime.now()))
        cnx.close()

    def execute(self, req, arg=None, res=None):
        self.reqs.put((req, arg or tuple(), res)) # Add this

    def queries(self):
        return self.reqs.qsize()

    def empty(self):
        return self.reqs.empty()

    def select(self, req, arg, results=0): # Add arg here
        cnx = sqlite3.Connection(self.db)
        cursor = cnx.cursor()
        try:
            if results == 0:
                cursor.execute(req, arg) # Add arg here
                ret = [x for x in cursor.fetchall()]
                cnx.close()
                return ret
            else:
                cursor.execute(req, arg) # Add arg here
                ret = [x for x in cursor.fetchall()[:results]]
                cnx.close()
                return ret
        except:
            print("Unexpected error: {0}".format(sys.exc_info()[0]))
            cnx.close()

    def commit(self):
        self.execute("--commit--")

    def close(self):
        self.execute('--close--')

    def escribir_error(self, texto):
        #with open(os.path.dirname(os.path.abspath(__file__)) + '\errores.txt', 'a') as archivo:
        #    archivo.write(texto)
        print(texto)

传递给函数时 Sqlite 执行中的变量 (Python3)

Variables in Sqlite execution when being passed into a function (Python3)

python

sql

sqlite

multithreading