django-orchestra/orchestra/contrib/mailer/engine.py

77 lines
2.6 KiB
Python
Raw Normal View History

2015-05-04 19:52:53 +00:00
import smtplib
2015-05-07 19:00:02 +00:00
from datetime import timedelta
2015-05-04 19:52:53 +00:00
from socket import error as SocketError
from django.core.mail import get_connection
2015-05-07 19:00:02 +00:00
from django.db.models import Q
from django.utils import timezone
2015-05-04 19:52:53 +00:00
from django.utils.encoding import smart_str
2015-06-05 12:39:58 +00:00
from orchestra.utils.sys import LockFile, OperationLocked
2015-05-06 10:51:12 +00:00
2015-05-07 19:00:02 +00:00
from . import settings
2015-05-04 19:52:53 +00:00
from .models import Message
2015-08-31 11:58:59 +00:00
def send_message(message, connection=None, bulk=settings.MAILER_BULK_MESSAGES):
2015-08-05 22:58:35 +00:00
message.last_try = timezone.now()
update_fields = ['last_try']
if message.state != message.QUEUED:
message.retries += 1
update_fields.append('retries')
message.save(update_fields=update_fields)
2015-11-09 19:07:06 +00:00
if connection is None:
connection = get_connection(backend='django.core.mail.backends.smtp.EmailBackend')
if connection.connection is None:
try:
connection.open()
except Exception as err:
message.defer()
message.log(error)
return
error = None
2015-05-04 19:52:53 +00:00
try:
connection.connection.sendmail(message.from_address, [message.to_address], smart_str(message.content))
2015-07-02 10:49:44 +00:00
except (SocketError,
smtplib.SMTPSenderRefused,
2015-05-04 19:52:53 +00:00
smtplib.SMTPRecipientsRefused,
smtplib.SMTPAuthenticationError) as err:
message.defer()
error = err
else:
message.sent()
message.log(error)
2015-07-02 10:49:44 +00:00
return connection
2015-05-04 19:52:53 +00:00
2015-06-05 12:39:58 +00:00
def send_pending(bulk=settings.MAILER_BULK_MESSAGES):
try:
with LockFile('/dev/shm/mailer.send_pending.lock'):
2015-11-09 19:07:06 +00:00
connection = get_connection(backend='django.core.mail.backends.smtp.EmailBackend')
2015-08-31 11:58:59 +00:00
cur, total = 0, 0
for message in Message.objects.filter(state=Message.QUEUED).order_by('priority', 'last_try', 'created_at'):
2015-11-09 19:07:06 +00:00
if cur >= bulk:
2015-08-31 11:58:59 +00:00
connection.close()
cur = 0
2015-11-09 19:07:06 +00:00
send_message(message, connection, bulk)
2015-08-31 11:58:59 +00:00
cur += 1
total += 1
2015-06-05 12:39:58 +00:00
now = timezone.now()
qs = Q()
for retries, seconds in enumerate(settings.MAILER_DEFERE_SECONDS):
delta = timedelta(seconds=seconds)
2015-08-05 22:58:35 +00:00
qs = qs | Q(retries=retries, last_try__lte=now-delta)
2015-08-31 11:58:59 +00:00
for message in Message.objects.filter(state=Message.DEFERRED).filter(qs).order_by('priority', 'last_try'):
2015-11-09 19:07:06 +00:00
if cur >= bulk:
2015-08-31 11:58:59 +00:00
connection.close()
cur = 0
2015-11-09 19:07:06 +00:00
send_message(message, connection, bulk)
2015-08-31 11:58:59 +00:00
cur += 1
total += 1
return total
2015-06-05 12:39:58 +00:00
except OperationLocked:
pass
2015-11-09 19:07:06 +00:00
finally:
if connection.connection is not None:
connection.close()