这是我用来使 API 操作速度提高 10 倍的函数:
def load_url(req, id, data, timeout):
headers = {'Authorization': 'AT-API 111111222222333333344444445555555'}
r = req.post("https://service.com/api/v1/compare", headers=headers, data=data, timeout=timeout)
data = r.json()
print id
if data['error']:
print data['error']
else:
c.execute("UPDATE offers SET valid = ? WHERE id = ?", ('valid' if data['data']['success'] else 'invalid', id))
print data['data']['success']
print data['data']['count']
return r.json()
if __name__ == '__main__':
...
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_url = {executor.submit(load_url, s, data['id'], data, 120): data for data in datas}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
我们每 4 小时定期运行此脚本。在 load_url 中，我们打印并检查请求的状态，data['data']['count'] - 是一个月剩余的调用量。我想在通话次数达到 5000 次或更少后发送电子邮件通知，但仅限一次。如何实现只发送一条消息，而不发送5条消息，不发送50条消息？我们使用 Sqlite3 来存储数据。
我们使用 Mailgun 发送电子邮件:
def send_simple_message(email_list):
for email in email_list:
response = requests.post(
"https://api.mailgun.net/v3/newsletter.company.com",
auth=("api", "key-1234567"),
data={"from": "Mailgun Sandbox <<a href="https://stackoverflow.com/cdn-cgi/l/email-protection" class="__cf_email__" data-cfemail="4d3d223e39202c3e39283f0d23283a3e21283939283f6320222f383d3d3e632e2220" rel="noreferrer noopener nofollow">[email protected]</a>>",
"to": email,
"subject": "Agent - we reached the limit by API",
"html": "We reached the limit for agent" })
最佳答案
您可以使用锁定机制来防止多个线程执行 send_simple_message 函数，并设置一个同步值来跟踪邮件是否已发送。
import threading
lock = threading.Lock()
has_been_sent = False
# then in your load_url function you could do something like
if condition on count:
with lock:
if not has_beend_sent:
# send mail
has_been_sent = True
关于python - 如何在线程池执行的函数中仅发送一次电子邮件？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/45031799/