我们从Python开源项目中，提取了以下49个代码示例，用于说明如何使用smtplib.SMTPServerDisconnected()。
def close(self): """Closes the connection to the email server.""" if self.connection is None: return try: try: self.connection.quit() except (ssl.SSLError, smtplib.SMTPServerDisconnected): # This happens when calling quit() on a TLS connection # sometimes, or when the connection was already disconnected # by the server. self.connection.close() except smtplib.SMTPException: if self.fail_silently: return raise finally: self.connection = None
def _send_email(self, from_addr, to_addrs, msg): if self._connect_smtp(): try: try: self.smtp_connection.sendmail(from_addr, to_addrs, msg) return except smtplib.SMTPServerDisconnected as e: logging.debug("SMTP disconnected: {} - reconnecting".format(str(e))) if self._connect_smtp(force=True): self.smtp_connection.sendmail(from_addr, to_addrs, msg) return except Exception as e: logging.error("Error while sending email to {}: " "{}".format(email_addresses, str(e)), exc_info=True)
def send_mail(self, sender, receiver, subject="", body=""): """Send email from sender to receiver """ mime_msg = MIMEMultipart('related') mime_msg['Subject'] = subject mime_msg['From'] = sender mime_msg['To'] = receiver msg_txt = MIMEText(body, 'plain') mime_msg.attach(msg_txt) try: host = getToolByName(self, 'MailHost') host.send(mime_msg.as_string(), immediate=True) except SMTPServerDisconnected as msg: logger.warn("SMTPServerDisconnected: %s." % msg) except SMTPRecipientsRefused as msg: raise WorkflowException(str(msg))
def send_email_new_item(self, FROM, TO, TEXT, item_name): SUBJECT = 'New potencial item: %s' % item_name try: message = """\From: %s\nTo: %s\nSubject: %s\n\n%s""" % (FROM, ", ".join(TO), SUBJECT, TEXT) self.server.sendmail(FROM, TO, message) self.email_number += 1 print 'Email number '+ str(self.email_number) + ' was sent for the item ' + item_name except smtplib.SMTPRecipientsRefused: print 'The email was not sent, the target refused' return False except smtplib.SMTPServerDisconnected: print 'The server is disconnected.' return False except: print 'SHIT HAPPENED DEAL WITH IT' return False
def create(self, validated_data): if self._get_cache() is not None: raise Throttled() callback = validated_data['callback'] signed_data = signing.dumps(dict(callback=callback, user_id=self.user.pk)) callback = add_params(callback, sign=signed_data) email_message = get_password_reset_email(self.user, callback) try: email_message.send() except smtplib.SMTPServerDisconnected as e: raise serializers.ValidationError( 'Mail sending timeout.', code=status.HTTP_500_INTERNAL_SERVER_ERROR) except smtplib.SMTPException as e: raise serializers.ValidationError( 'Unknown SMTP error: %s.' % str(e), code=status.HTTP_500_INTERNAL_SERVER_ERROR) else: self._set_cache() return 'OK'
def close(self): """Closes the connection to the email server.""" if self.connection is None: return try: try: self.connection.quit() except (ssl.SSLError, smtplib.SMTPServerDisconnected): # This happens when calling quit() on a TLS connection # sometimes, or when the connection was already disconnected # by the server. self.connection.close() except: if self.fail_silently: return raise finally: self.connection = None
def send_message(self, target): try: msg = self.messages[random.randrange(len(self.messages))] self.server.sendmail(self.email, target, msg) self.messages_sent += 1 self.write_to_messages_sent() self.write_to_log("%s sent to %s" % (self.email, target)) except smtplib.SMTPServerDisconnected: self.write_to_log("Disconnected by ban, email %s" % (self.email)) self.initialize_server() self.write_to_log("Logged in again to %s" % (self.email)) #Put universal except in thread class #Consider calling write_to_messages_sent only when thread is being shut down
def testNotConnected(self): # Test various operations on an unconnected SMTP object that # should raise exceptions (at present the attempt in SMTP.send # to reference the nonexistent 'sock' attribute of the SMTP object # causes an AttributeError) smtp = smtplib.SMTP() self.assertRaises(smtplib.SMTPServerDisconnected, smtp.ehlo) self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, 'test msg')
def test_with_statement(self): with smtplib.SMTP(HOST, self.port) as smtp: code, message = smtp.noop() self.assertEqual(code, 250) self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo') with smtplib.SMTP(HOST, self.port) as smtp: smtp.close() self.assertRaises(smtplib.SMTPServerDisconnected, smtp.send, b'foo')
def disconnect(self): if self.connected: try: self.smtp_server.quit() except smtplib.SMTPServerDisconnected as e: pass self.connected = False self.smtp_server = None
def test_with_statement_QUIT_failure(self): with self.assertRaises(smtplib.SMTPResponseException) as error: with smtplib.SMTP(HOST, self.port) as smtp: smtp.noop() self.serv._SMTPchannel.quit_response = '421 QUIT FAILED' self.assertEqual(error.exception.smtp_code, 421) self.assertEqual(error.exception.smtp_error, b'QUIT FAILED') #TODO: add tests for correct AUTH method fallback now that the #test infrastructure can support it. # Issue 17498: make sure _rset does not raise SMTPServerDisconnected exception
def start(self): smtp = smtplib.SMTP(self.mta) while self.running: try: (sender, receivers, subject, message, attachments) = MAIL_QUEUE.get(block=True, timeout=2) self._send_mail(smtp, sender, receivers, subject, message, attachments) except Empty: pass except smtplib.SMTPServerDisconnected: # Reconnect and retry once smtp.connect(self.mta) self._send_mail(smtp, sender, receivers, subject, message, attachments) except Exception: logger.exception("Unhandled exception in mail thread") smtp.quit()
