公開されたAPIにログを送信するクラスがあります。今私がやりたいのは、失敗したログをリストに保存/保存して、再びサーバーに再送信できるようにすることです。永続的リストPythonの実装
これはこれまで私が行ってきたことです。 私はクラス変数を宣言していることに気付くかもしれませんが、私はreadがこれは本当にお勧めできません。
私の懸念は、リストやキューを永続化する良い方法があるということですか?唯一の変数のLogManager店以来
from collections import *
import time
import threading
import requests
import json
URL_POST = "http:/some/url/here"
class LogManager:
listQueue = deque()
def post(self, log):
headers = {"Content-type": "application/json",
"Accept": "text/plain", "User-Agent": "Test user agent"}
resp = requests.post(URL_POST, data=json.dumps(log), headers=headers)
return resp.status_code
def send_log(self, log):
try:
print ("Sending log to backend")
self.post(log)
except: # sending to server fails for some reason
print ("Sending logs to server fail, appending to Queue")
LogManager.listQueue.append(log)
def resend_log(self, log):
print ("checking if deque has values")
if LogManager.listQueue:
logToPost = LogManager.listQueue.popleft()
try:
self.post(logToPost)
print ("resending failed logs")
except: #for some reason it fails again
LogManager.listQueue.appendleft(logToPost)
print ("appending log back to deque")
def run(self,log):
t1 = threading.Thread(target=self.send_log, args=(log,))
t2 = threading.Thread(target=self.resend_log,args=(log,))
t1.start()
time.sleep(2)
t2.start()
t1.join()
t2.join()
if __name__ == "__main__":
while True:
logs = LogManager()
logs.run({"some log": "test logs"})
私が使用したいロジックを検証していただきありがとうございます。 – essramos