我有一個 Django 應用程式托管在遠程服務器上，它以相對較短的時間間隔運行一些 cron 作業。其中一個 cron 作業執行從資料庫獲取查詢集的命令，呼叫外部 API 并根據 API 的回應更改模型。如果我不小心，cron 作業將在 API 回應之前執行多次，從而導致并發問題和同一模型的多個實體同時更新。
我有不同的策略來避免這個問題，但我想撰寫可以在本地運行的測驗，模擬 API 呼叫并確保兩個 cron 任務不會同時嘗試處理一個物件。我怎么做？
我的代碼看起來像這樣（說明問題的目的）：
def task():
qs = MyModel.objects.filter(task_ran=False)
for model in qs:
resp = api_call(model.foo)
model.bar = resp
model.task_ran = True
model.save()
那么，我如何撰寫一個測驗，檢查是否task()在第一次呼叫完成之前第二次呼叫，然后它不會再次更新模型并且不會再次呼叫 API？下面是一個測驗的草圖，我試圖將呼叫task()放在單獨的執行緒中，但這會導致測驗凍結并且 - 在 - 之后KeyboardInterrupt- 失敗
django.db.utils.OperationalError：其他用戶正在訪問資料庫“test_db”
詳細資訊：還有 1 個其他會話正在使用該資料庫。```
@patch("api_call")
def test_task(self, mock_api_call):
def side_effect(number):
time.sleep(2)
return number 1
mock_api_call.side_effect = side_effect
# how to call these simultaneously? threading causes Django to get mad
task()
task()
mock_api_call.assert_called_once()
uj5u.com熱心網友回復：
好的，所以我根據這個答案找到了答案。基本上，測驗可以通過執行緒在 Django 中完成，但它需要做一些事情：
- 首先，測驗類本身必須是
TransactionTestCase（至少如果它涉及任何資料庫惡作劇或.select_for_update我的代碼所做的使用）的子類。 - 其次，在每個執行緒中打開的資料庫連接應該在執行緒終止時再次關閉。這可以通過使用 a
ThreadPoolExecutor創建Futures 然后添加一個回呼函式來完成，當執行緒通過.add_done_callback.
所以有了這個，測驗可以寫成如下：
import concurrent.futures
from django.db import connections
from django.test import TransactionTestCase
class CronTestCase(TransactionTestCase):
def on_done(self, future):
connections.close_all()
@patch("api_call")
def test_task(self, mock_api_call):
# setup the test
num_threads = 5
with concurrent.futures.ThreadPoolExecutor() as executor:
for _ in range(num_threads):
future = executor.submit(task)
future.add_done_callback(self.on_done)
mock_api_call.assert_called_once()
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/355614.html