我们从Python开源项目中，提取了以下31个代码示例，用于说明如何使用flask_mail.Message()。
def send_email(subject, recipients, template, **kwargs): if not isinstance(recipients, list): recipients = list(recipients) msg = Message(subject, reply_to=current_app.config['MAIL_DEFAULT_SENDER'], recipients=recipients) msg.body = render_template(template + '.txt', **kwargs) msg.html = render_template(template + '.html', **kwargs) attachments = kwargs.get('attachments', []) mimes = MimeTypes() for file in attachments: path_ = os.path.join(current_app.config['APP_UPLOADS'], file) with current_app.open_resource(path_) as fp: mime = mimes.guess_type(fp.name) msg.attach(path_, mime[0], fp.read()) app = current_app._get_current_object() t = Thread(target=send_async_email, args=[app, msg]) t.start() return t
def send_email(to_email, subject, template_path, template_vars, attachment_path=None): if not ENV_DICT['SEND_EMAIL']: print '++ sending emails is currently disabled in this environment. Enable SEND_EMAIL to allow email sending' return # email address that emails will be sent from from_email = ENV_DICT['MAIL_DEFAULT_SENDER'] # render HTML from template page_html = render_template(template_path, **template_vars) msg = Message(subject=subject, sender=from_email, recipients=[to_email], html=page_html) if attachment_path: with open(attachment_path, 'r') as f: msg.attach("osf-results.json", "text/plain", f.read()) mail.send(msg)
def send_error_mail(exception): """Sends an error mail to the admin containing the traceback and configuration. After that, a custom HTTP 500 page is shown. :param exception: the exception raised :type exception: ``Exception`` :return: the HTML to send back in the response, and the HTTP code 500. :rtype: str, int """ # Inspired from <https://github.com/jasonwyatt/Flask-ErrorMail>. message = Message("Join2 ORCID exception: %s" % exception, sender=CFG_SITE_ADMIN_EMAIL, recipients=[CFG_SITE_ADMIN_EMAIL]) message_contents = ["Traceback:", "=" * 80, traceback.format_exc(), "\n", "Request Information:", "=" * 80] environ = request.environ for key in sorted(environ.keys()): message_contents.append("%s: %s" % (key, environ.get(key))) message.body = "\n".join(message_contents) + "\n" mailer.send(message) return render_template("500.html"), 500
def send_email(device_data, enrolled=True): subject = 'Mobile Device Enrolled' if enrolled else 'Mobile Device Un-Enrolled' email_data = { 'enrolled': enrolled, 'name': device_data['deviceName'], 'model': device_data['model'], 'serial': device_data['serialNumber'], 'url': os.path.join('{}/mobileDevices.html?id={}'.format(JSS_ADDRESS, device_data['jssID'])), 'time': datetime.datetime.utcnow() } txt, html = build_email_body(email_data) msg = Message( subject, recipients=[DESTINATION_EMAIL] ) msg.body = txt msg.html = html thr = Thread(target=send_async_email, args=[msg]) thr.start()
def sendEmail(httpOrigin, email, activationId, mailer): ''' Create a message and send it from our email to the passed in email. The message should contain a link built with the activationId ''' if activationId is None: return #Change the URL to the appropriate environment host = getProperEnvironment(httpOrigin) url = host + '/activation/{0}'.format(activationId) message = render_template('registrationEmail.html', url=url) #Build Message msg = Message('Mentii: Thank You for Creating an Account!', recipients=[email], extra_headers={'Content-Transfer-Encoding': 'quoted-printable'}, html=message) #Send Email mailer.send(msg)
def sendClassRemovalEmail(dynamoDBInstance, mailer, jsonData): ''' Create a message to send it from our email to the passed in email. The message should notify the user they were removed from a class ''' email = jsonData.get('email') classCode = jsonData.get('classCode') cl = getClassByCode(classCode, dynamoDBInstance) classTitle = cl['title'] message = render_template('removedEmail.html', classTitle=classTitle) #Build Message msg = Message('You have been removed from a class', recipients=[email], extra_headers={'Content-Transfer-Encoding': 'quoted-printable'}, html=message) mailer.send(msg)
def send_confirm_email(self): """ Send an confirm email to user. contains a link to confirm the email confirm link is not provide by this app, a client must implement this endpoint to complete the confirmation. """ from server import app, mail token = self.generate_confirm_email_token() confirm_url = '{0}://{1}/email-confirm?token={2}'.format(app.config['SITE_PROTOCOL'], app.config['SITE_HOST'], token) subject = '[{0}] Email Address Confirmation'.format(app.config['SITE_NAME']) email_content = render_template('email-confirm.html', info={ 'confirm_title': subject, 'confirm_url': confirm_url, 'site_name': app.config['SITE_NAME'], 'user_name': self.name }) msg = Message(subject, recipients=[self.email], html=email_content) try: mail.send(msg) except SMTPAuthenticationError: raise ServerError('SMTP authentication failed', 500)
def sendUpsellEmail(data): name = Utils.getUserName(data['user']) with webapp.app_context(): consumer_mail = render_template('mailers/extend_order.html', name = name, book_name = data['book_name'], order_id = data['order_id'], items = data['items'], curated_items = data['curated_items'], quote = data['quote'], quote_author = data['quote_author']) pre = Premailer(consumer_mail, remove_classes=False, strip_important=False) consumer_mail = pre.transform() email = Message('Enjoying the book?', recipients=[data['user'].email]) email.html = consumer_mail mail.send(email) return True
def request_password_reset(username): """ Emails a user a link to reset their password. Checks that a username was submitted to the function and grabs the relevant team info from the db. Generates a secure token and inserts it into the team's document as 'password_reset_token'. A link is emailed to the registered email address with the random token in the url. The user can go to this link to submit a new password, if the token submitted with the new password matches the db token the password is hashed and updated in the db. Args: username: the username of the account """ validate(password_reset_request_schema, {"username":username}) user = safe_fail(api.user.get_user, name=username) if user is None: raise WebException("No registration found for '{}'.".format(username)) token_value = api.token.set_token({"uid": user['uid']}, "password_reset") body = """We recently received a request to reset the password for the following {0} account:\n\n\t{2}\n\nOur records show that this is the email address used to register the above account. If you did not request to reset the password for the above account then you need not take any further steps. If you did request the password reset please follow the link below to set your new password. \n\n {1}/reset#{3} \n\n Best of luck! \n The {0} Team""".format(api.config.competition_name, api.config.competition_urls[0], username, token_value) subject = "{} Password Reset".format(api.config.competition_name) message = Message(body=body, recipients=[user['email']], subject=subject) mail.send(message)
def send_email_invite(gid, email, teacher=False): """ Sends an email registration link that will automatically join into a group. This link will bypass the email filter. """ group = api.group.get_group(gid=gid) token_value = api.token.set_token({"gid": group["gid"], "email": email, "teacher": teacher}, "registration_token") registration_link = "{}/#g={}&r={}".\ format(api.config.competition_urls[0], group["gid"], token_value) body = """ You have been invited by the staff of the {1} organization to compete in {0}. You will need to follow the registration link below to finish the account creation process. If you believe this to be a mistake you can safely ignore this email. Registration link: {2} Good luck! The {0} Team. """.format(api.config.competition_name, group["name"], registration_link) subject = "{} Registration".format(api.config.competition_name) message = Message(body=body, recipients=[email], subject=subject) mail.send(message)
def send_email(organisation, recipients): sender = config.ADMINS[0] subject = 'Results have been generated for ' + organisation text_body = """ Greetings from Zense, We have completed your task in regards to %s Please visit us back Regards, Team Stalker, Zense """ % str(organisation) msg = Message(subject, sender = sender, recipients = recipients) msg.body = text_body with app.app_context(): mail.send(msg)
def internal_server_error(e): message = repr(e) trace = traceback.format_exc() trace = string.split(trace, '\n') timestamp = (datetime.fromtimestamp(time.time()) .strftime('%Y-%m-%d %H:%M:%S')) if current_user.is_authenticated: user = current_user.username else: user = 'anonymous' gathered_data = ('message: {}\n\n\n' 'timestamp: {}\n' 'ip: {}\n' 'method: {}\n' 'request.scheme: {}\n' 'request.full_path: {}\n' 'user: {}\n\n\n' 'trace: {}'.format(message, timestamp, request.remote_addr, request.method, request.scheme, request.full_path, user, '\n'.join(trace))) # send email to admin if app.config['TESTING']: print(gathered_data) else: mail_message = gathered_data msg = Message('Error: ' + message[:40], body=mail_message, recipients=[app.config['ADMIN_MAIL']]) mail.send(msg) flash(_('A message has been sent to the administrator'), 'info') bookcloud_before_request() return render_template('500.html', message=gathered_data), 500
def send_email(to, subject, template, **kwargs): app = current_app._get_current_object() msg = Message(app.config['CIRCULATE_MAIL_SUBJECT_PREFIX'] + ' ' + subject, sender=app.config['CIRCULATE_MAIL_SENDER'], recipients=[to]) msg.body = render_template(template + '.txt', **kwargs) msg.html = render_template(template + '.html', **kwargs) thr = Thread(target=send_async_email, args=[app, msg]) thr.start() return thr
def send_email(sender, recipients, subject, text_body, html_body=None, attach=None): msg = Message(subject, reply_to=sender, recipients=recipients) msg.body = text_body msg.html = html_body if attach: mimes = MimeTypes() for file in attach: path_ = os.path.join(current_app.config['APP_UPLOADS'], file) with current_app.open_resource(path_) as fp: mime = mimes.guess_type(fp.name) msg.attach(file, mime[0], fp.read()) mail.send(msg)
def _do_on_login(sender, user, **extra): """Notify the user of a new login. This is only done if the ``notify_login`` attribute is set to ``True``. """ if user.notify_login: notification = Message( 'akamatsu - New session started', recipients=[user.email], body=render_template('mail/login.txt', user=user), html=render_template('mail/login.html', user=user) ) mail.send(notification)
def send_email(to, subject, template, **kwargs): """Send email using multible threads. """ app = current_app._get_current_object() msg = Message(app.config['MAIL_SUBJECT_PREFIX'] + ' ' + subject, sender=app.config['MAIL_SENDER'], recipients=[to]) msg.body = render_template(template + '.txt', **kwargs) msg.html = render_template(template + '.html', **kwargs) thread = Thread(target=send_async_email, args=[app, msg]) thread.start() return thread
def send_email(subject, recipients, text_body, html_body, sender=None): msg = Message(subject, recipients=recipients, sender=sender) msg.body = text_body msg.html = html_body mail.send(msg)
def send_mails(recipients, cc, mail_title, mail_body): msg = Message(mail_title) msg.body = mail_body msg.sender = current_app._get_current_object().config['MAIL_USERNAME'] msg.recipients = recipients msg.cc = cc mail.send(msg)
def send_email_thread(to, subject, template, **kwargs): """Send async email using threading. """ app = current_app._get_current_object() msg = Message(app.config['APP_MAIL_SUBJECT_PREFIX'] + ' ' + subject, sender=app.config['APP_MAIL_SENDER'], recipients=[to]) msg.body = render_template(template + '.txt', **kwargs) msg.html = render_template(template + '.html', **kwargs) thr = Thread(target=send_async_email, args=[app, msg]) thr.start() return thr
def send_email_celery(to, subject, template, countdown=None, **kwargs): """Send async email using Celery. """ app = current_app._get_current_object() msg = Message(app.config['APP_MAIL_SUBJECT_PREFIX'] + ' ' + subject, sender=app.config['APP_MAIL_SENDER'], recipients=[to]) msg.body = render_template(template + '.txt', **kwargs) msg.html = render_template(template + '.html', **kwargs) send_celery_async_email.apply_async(args=[msg], countdown=countdown)
def send_reset_password_email(user: models.User) -> None: token = user.get_reset_token() html_body = current_app.config['EMAIL_TEMPLATE'].replace( '\n\n', '<br><br>' ).format( site_url=current_app.config["EXTERNAL_URL"], url=f'{psef.app.config["EXTERNAL_URL"]}/reset_' f'password/?user={user.id}&token={token}', user_id=user.id, token=token, user_name=html.escape(user.name), user_email=html.escape(user.email), ) text_maker = html2text.HTML2Text(bodywidth=78) text_maker.inline_links = False text_maker.wrap_links = False message = Message( subject=f'Reset password on {psef.app.config["EXTERNAL_URL"]}', body=text_maker.handle(html_body), html=html_body, recipients=[user.email], ) try: mail.send(message) except Exception: raise APIException( 'Something went wrong sending the email, ' 'please contact your site admin', f'Sending email to {user.id} went wrong.', APICodes.UNKOWN_ERROR, 500, )
def send_notif_message(self, notif): msg = Message('Update on ' + notif.source_type + ' ' + str(notif.source_id), recipients=[notif.user.email]) msg.body = notif.to_text() + '\n\nsee it here: ' + notif.link options = { 'notif': notif } msg.html = render_template('email/notif.html', **options) self.messages.append(msg)
def send_invite_message(self, email, sender): msg = Message('Invitation to Tingsmen', recipients=[email]) msg.body = sender + ' invited you to join Tingsmen\n\nTingsmen is a social network where you can propose any ' \ 'change to the platform and, supposing the community is ok with it, it will be developed' \ '\n\nWanna know more? visit: https://tingsmen.herokuapp.com' options = { 'sender': sender } msg.html = render_template('email/invite.html', **options) self.messages.append(msg)
def send_email(to, subject, template, **kwargs): app = current_app._get_current_object() msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + ' ' + subject, sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to]) msg.body = render_template(template + '.txt', **kwargs) msg.html = render_template(template + '.html', **kwargs) thr = Thread(target=send_async_email, args=[app, msg]) thr.start() return thr
def _send_email(app, email_addr): with app.app_context(): msg = Message("[FreeTest] ??????", recipients=[email_addr]) msg.body = "This is body333." mail.send(msg)
def _broadcast_email(app, email_addrs): with app.app_context(): for email_addr in email_addrs: msg = Message("[FreeTest] ??????", recipients=[email_addr]) msg.body = "This is body333." mail.send(msg)
def send_email( app, to, subject, template ): msg = Message( subject, recipients=[to], html=template, sender=app.config['MAIL_DEFAULT_SENDER']) # with app.app_context(): # mail.send(msg) thr = Thread( target = send_async_email, args = [app, msg]) thr.start()
