Блокировка выполнения потока до завершения в тесте Python Django

У меня есть приложение Django, которому нужно сделать несколько вызовов 3d-сервиса для вычисления встраивания. Этот вызов занимает время и блокирует выполнение, а я хочу сделать его асинхронным. Для этого я использую Thread.

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()

Поток начинает вычислять встраивание, а представление возвращает ответ. По окончании вычисления в БД создается новый объект AnswerEmbedding.

Теперь мне нужно проверить, правильно ли он работает. Итак, если я просто запущу тест, который делает вызов api:

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

Этот тест не работает, потому что он проверяет AnswerEmbedding.objects.get до того, как он фактически создан в потоке.

Итак, мне нужно как-то либо заставить этот Thread работать синхронно, либо найти способ правильно над ним поиздеваться. Я могу просто передразнить поток, но при этом тест не будет проверять, создано ли вложение или нет.

Есть идеи? Спасибо

Мне интересно, в чем смысл использования нити?

если вы все еще хотите заблокировать его, я думаю, вы можете использовать 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()

или если ваша проблема заключается в ожидании создания embedding, вы можете использовать проверку состояния 

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()
Вернуться на верх