一段時間以來，我一直在研究如何從不同的執行緒讀取/寫入 sqlite 資料庫，我找到了很多描述需要發生的事情的答案和檔案，但是我遠無法實作所需的，所以我決定使用我發現其他人制作的現有課程。
很慚愧地承認，但弄清楚如何讓這個類作業已經花了幾個小時，盡管我現在不知道為什么但是我無法在執行函式中獲取變數
我通常會這樣做：
c.execute("SELECT codeID FROM users WHERE codeID=:code", {'code':tag_attempt})
這作業得很好，但是當我用多執行緒類嘗試同樣的事情時，它就不起作用了（該類有一個“選擇”函式，它可以獲取并回傳資料，這是我能夠從資料庫中獲取資料的唯一方法，但是執行函式有完全相同的問題）（我也嘗試過這種使用變數的方法）
for q in sql.select(("select codeID from users where codeID=?", (tag_attempt)), 0):
print(q)
# TypeError: 'NoneType' object is not iterable
這是使用的類，但我還將包含鏈接
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)
總結我希望能夠在一個單獨的執行緒中獲取資料，這對于這個類是可能的我只是無法在任何階段包含變數
信用 https://gist.github.com/User001501/3053f26100ddf281600668fed347e518
uj5u.com熱心網友回復：
方法執行接收單個引數。
看起來您可以使用名為占位符的字串，例如
c.execute("SELECT codeID FROM users WHERE codeID='{code}'".format(**{'code': ag_attempt})
uj5u.com熱心網友回復：
我找到了一個解決方案，并想發布它以防萬一將來它對任何人都有好處
我向類中的函式傳遞了一個新引數，它只是字典，我將對更改的行進行評論。這個新代碼適用于我在問題中嘗試使用的行，因此函式的輸入是相同的，并且像常規 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)
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/438833.html