Django语法纠错API出现SQLite线程错误，求助排查

我正在开发一个名为spell_grammer_check的Django API，包含spell_correction和grammar_correction两个端点。拼写纠错功能可正常返回结果，但语法纠错功能触发错误：ProgrammingError at /grammar_correction/，提示信息为"SQLite objects created in a thread can only be used in that same thread"。

以下是相关代码：

import sqlite3
from django.http import JsonResponse
from whoosh.index import open_dir
from whoosh.qparser import QueryParser
from whoosh.qparser import plugins
from fuzzywuzzy import fuzz
import heapq
import re

ix = open_dir('C:/Users/Na Da/Documents/Final Semester/final project/myproject/indexed_unigram')
searcher = ix.searcher()
conn = sqlite3.connect("C:/Users/Na Da/Documents/Final Semester/final project/myproject/bigramDatabase")
c = conn.cursor()

def word_search(word):
    query_parser = QueryParser("ngram", schema=ix.schema)
    query = query_parser.parse(word)
    count = 0
    word = ' '
    for r in  searcher.search(query,limit = 1):
        count = r['count']
    return count

def  word_suggestions_special_char(word,pos):
    replaces = [( L + c + R[1:]) for L, R in  [(word[:pos],word[pos:])] if R for c in  'اأإ' for word in (L + c + R[1:],)]
    word_prob={}
    for w in replaces:
        count = word_search(w)
        if count!=0:
            word_prob[w] = count
    return heapq.nlargest(1, word_prob,key=word_prob.get)


def word_suggestions(word):
    my_parser = QueryParser("ngram", schema=ix.schema)
    my_parser.add_plugin(plugins.FuzzyTermPlugin())
    query = my_parser.parse(word+'~')
    results = searcher.search(query,limit = None)
    word_prob={}
    for r in  results:
        word_prob[r['ngram']] = r['count'] 
    return  heapq.nlargest(5, word_prob,key=word_prob.get)

def spelling_correction(request, sentence):
    suggestions = {}
    re.sub('[?.\"",;:]', '', (sentence)) + ' '
    words = sentence.split()
    for word in words:
        if word == 'و':
            continue
        count = word_search(word)
        if count<1000:
             suggestions[word] = word_suggestions(word)
        else:
            if word [0] == 'ا' and word[2] =='ا' :
                suggestions[word] = word_suggestions_special_char(word,2)
                continue
            if word[0] =='ا' :
                suggestions[word] = word_suggestions_special_char(word,0)
                continue
            suggestions[word] = word
    return JsonResponse(suggestions)

def get_corrections(word):
    query = "SELECT word2, count FROM mytable WHERE word1 = ?"
    c.execute(query, (word,))
    bigrams = c.fetchall()
    total_count = sum([count for _, count in bigrams])
    probabilities = [(w2, count / total_count) for w2, count in bigrams]
    probabilities.sort(key=lambda x: x[1], reverse=True)
    suggestions = [w2 for w2, _ in probabilities[:10]]
    return  suggestions 

def get_bigram_counts( word1, word2):
    c = conn.cursor()
    query = 'SELECT* FROM mytable WHERE word1=? AND word2=?;'
    c.execute(query, (word1, word2))
    count = c.fetchall()
    if count:
        return count
    else:
        return 0


def get_corrections_with_thirdWord(word,word3):
    query = "SELECT word2, count FROM mytable WHERE word1 = ?"
    c.execute(query, (word,))
    bigrams = c.fetchall()
    results=[]
    for w in bigrams:
        r = get_bigram_counts( w[0], word3)
        if r:
            results.append (w)
    total_count = sum([count for _, count in results])
    probabilities = [(w2, count / total_count) for w2, count in results]
    probabilities.sort(key=lambda x: x[1], reverse=True)
    suggestions = [w2 for w2, _ in probabilities[:]]
    return  suggestions 

def grammar_correction(request,sentence):
    words =sentence.split()
    corrected_sentence = []
    if len(words)==1:
        return sentence
    for i in range(len(words)):
        result = get_bigram_counts( words[i], words[i+1])
        if result:
            if i ==len(words)-2:
                corrected_sentence.append(words[i])
                corrected_sentence.append(words[i+1])
                return  ' '.join(corrected_sentence)
            corrected_sentence.append(words[i])
        else:
            if(i==len(words)-2):
                suggestions =get_corrections(words[i])
                matched_word = max( suggestions, key=lambda w: fuzz.ratio(words[i+1],w))
                corrected_sentence.append(words[i])
                corrected_sentence.append(matched_word)
                return' '.join(corrected_sentence)
            corrected_sentence.append(words[i])
            suggestions =get_corrections_with_thirdWord(words[i],words[i+2])
            if  suggestions:  
                matched_word = max( suggestions, key=lambda w: fuzz.ratio(words[i+1],w))
                words[i+1] = matched_word
            else:
                corrected_sentence +=words[i+1]
                continue
    return JsonResponse(' '.join(corrected_sentence)) 

我无法定位问题所在，请求帮助排查。

问题根源 #

Django是多线程框架，你在全局作用域初始化了SQLite连接conn和游标c，这些对象会被多个请求线程共享。而SQLite的默认规则是：连接和游标只能在创建它们的线程中使用，跨线程访问就会触发这个ProgrammingError。

拼写纠错功能没报错只是巧合——它没用到SQLite连接，而语法纠错的几个函数（get_corrections、get_bigram_counts等）都直接调用了全局游标c，所以触发了线程安全问题。

修复步骤 #

1. 移除全局的SQLite连接和游标 #

删除代码开头的全局初始化代码：

# 删掉这两行
conn = sqlite3.connect("C:/Users/Na Da/Documents/Final Semester/final project/myproject/bigramDatabase")
c = conn.cursor()

2. 在需要数据库操作的函数内创建连接和游标 #

每次数据库操作都创建新的连接，用完后关闭，确保每个线程使用自己的连接实例：

• 修改get_corrections函数：

def get_corrections(word):
    # 函数内创建连接和游标
    conn = sqlite3.connect("C:/Users/Na Da/Documents/Final Semester/final project/myproject/bigramDatabase")
    c = conn.cursor()
    
    query = "SELECT word2, count FROM mytable WHERE word1 = ?"
    c.execute(query, (word,))
    bigrams = c.fetchall()
    total_count = sum([count for _, count in bigrams])
    probabilities = [(w2, count / total_count) for w2, count in bigrams]
    probabilities.sort(key=lambda x: x[1], reverse=True)
    suggestions = [w2 for w2, _ in probabilities[:10]]
    
    # 关闭连接
    conn.close()
    return suggestions 

• 修改get_bigram_counts函数：

def get_bigram_counts(word1, word2):
    conn = sqlite3.connect("C:/Users/Na Da/Documents/Final Semester/final project/myproject/bigramDatabase")
    c = conn.cursor()
    
    query = 'SELECT* FROM mytable WHERE word1=? AND word2=?;'
    c.execute(query, (word1, word2))
    count = c.fetchall()
    
    conn.close()
    return count if count else 0

• 修改get_corrections_with_thirdWord函数：

def get_corrections_with_thirdWord(word, word3):
    conn = sqlite3.connect("C:/Users/Na Da/Documents/Final Semester/final project/myproject/bigramDatabase")
    c = conn.cursor()
    
    query = "SELECT word2, count FROM mytable WHERE word1 = ?"
    c.execute(query, (word,))
    bigrams = c.fetchall()
    results=[]
    for w in bigrams:
        r = get_bigram_counts(w[0], word3)
        if r:
            results.append(w)
    total_count = sum([count for _, count in results])
    probabilities = [(w2, count / total_count) for w2, count in results]
    probabilities.sort(key=lambda x: x[1], reverse=True)
    suggestions = [w2 for w2, _ in probabilities[:]]
    
    conn.close()
    return suggestions 

3. 额外优化：添加线程安全参数（可选） #

如果想复用连接（减少连接开销），可以在创建连接时添加check_same_thread=False参数，但注意这会带来线程安全风险，仅适用于单线程测试场景：

conn = sqlite3.connect("your_db_path", check_same_thread=False)

生产环境不建议使用这个参数，最好还是保持“按需创建连接、用完即关”的模式。

4. 修复grammar_correction的返回值与索引越界问题 #

原函数存在两处问题：一是部分分支直接返回字符串（不符合Django视图返回规范），二是循环范围会导致索引越界，修复后代码如下：

def grammar_correction(request,sentence):
    words = sentence.split()
    corrected_sentence = []
    if len(words)==1:
        return JsonResponse({"corrected": sentence})
    # 循环范围改为len(words)-1，避免i+1越界
    for i in range(len(words)-1):
        result = get_bigram_counts(words[i], words[i+1])
        if result:
            corrected_sentence.append(words[i])
            if i == len(words)-2:
                corrected_sentence.append(words[i+1])
        else:
            corrected_sentence.append(words[i])
            if i == len(words)-2:
                suggestions = get_corrections(words[i])
                matched_word = max(suggestions, key=lambda w: fuzz.ratio(words[i+1],w))
                corrected_sentence.append(matched_word)
            else:
                suggestions = get_corrections_with_thirdWord(words[i], words[i+2])
                if suggestions:
                    matched_word = max(suggestions, key=lambda w: fuzz.ratio(words[i+1],w))
                    corrected_sentence.append(matched_word)
                else:
                    corrected_sentence.append(words[i+1])
    return JsonResponse({"corrected": ' '.join(corrected_sentence)})

内容的提问来源于stack exchange，提问作者Alaa Refaat