块线程执行直到在测试python django中完成

Question

我有一个Django应用程序，需要进行一些3D服务调用以计算嵌入。这个呼叫需要时间并阻止执行，我想使其异步。为此，我正在使用线程。

class MyView():
    def create_answer_embedding(answer):
        embedding = long_computing_call(answer)
        db.write(embedding) <- pseudo code, the idea is that after creating the embedding, we create an AnswerEmbedding obj. in the db
    
    def post(request, *args, **kwargs):
        answer = request.data.get("answer")
        Thread(target=self.create_answer_embedding, args=(answer, )).start()
        return Response()

因此，线程开始计算嵌入，视图返回响应。计算完成后，将在DB中创建一个新的回答对象。

现在我需要测试它正常工作。因此，如果我只是运行一个进行API调用的测试：

response = client.post("/my-view")
answer_embedding = AnswerEmbedding.objects.get(answer__id=response.json()['id'])

此测试会失败，因为它在实际创建线程中创建之前检查了Answerembedding.objects.get。

因此，我需要以某种方式使此线程运行同步或找到正确模拟它的方法。我只能模拟该线程，但是这样，测试将无法检查是否创建了嵌入。

有什么想法吗？ 谢谢

Answer 1

我想知道，使用线程有什么意义？

如果您仍然想阻止它，我认为您可以使用join（）

class MyView():
    def create_answer_embedding(answer):
        embedding = long_computing_call(answer)
        db.write(embedding)
    
    def post(request, *args, **kwargs):
        answer = request.data.get("answer")
        th = Thread(target=self.create_answer_embedding, args=(answer, ))
        th.start()
        th.join()
        return Response()

，或者您的问题在等待嵌入要创建的

class MyView():
    is_embedding_created = False

    def create_answer_embedding(answer):
        embedding = long_computing_call(answer)
        MyView.is_embedding_created = True
        db.write(embedding)
        MyView.is_embedding_created = False
    
    def post(request, *args, **kwargs):
        answer = request.data.get("answer")
        Thread(target=self.create_answer_embedding, args=(answer, )).start()
        while not MyView.is_embedding_created:
            time.sleep(0.1) # use a delay while waiting to reduce resource consumption
        return Response()