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
python类_get_current_object()的实例源码
def send_mail(to, subject, template, **kwargs):
"""
????
:param to: ???
:param subject: ??
:param template: ????
:param kwargs: ????
:return:
"""
app = current_app._get_current_object()
message = Message(app.config['MAIL_SUBJECT_PREFIX'] + '' + subject, sender=app.config['MAIL_SENDER'],
recipients=[to])
message.body = render_template(template + '.txt', **kwargs)
message.html = render_template(template + '.html', **kwargs)
thr = Thread(target=send_async_email, args=[app, message])
thr.start()
return thr
def save(self):
"""
????????
:return:
"""
try:
db_session.add(self)
db_session.commit()
except:
db_session.rollback()
raise
finally:
db_session.close()
model_saved.send(app._get_current_object())
return self
def info():
app = current_app._get_current_object()
key = app.config['IP_INFO_DB_KEY']
ip = request.args.get('ip', None)
if not ip:
if request.headers.getlist("X-Forwarded-For"):
ip = request.headers.getlist("X-Forwarded-For")[0]
else:
ip = request.remote_addr
user_agent = {'browser': request.user_agent.browser,
'language': request.user_agent.language,
'platform': request.user_agent.platform,
'string': request.user_agent.string,
'version': request.user_agent.version,
'status': 'success',
'message': 'localhost user agent information'}
else:
user_agent = {'status': 'error',
'message': 'not query localhost information'}
ip_info = requests.get('http://api.ipinfodb.com/v3/ip-city/?key={0}&ip={1}&format=json'.format(key, ip)).json()
return jsonify(status='success', data={'ip': ip, 'ip_information': ip_info, 'user_agent': user_agent})
def nslookup():
dom = request.args.get('domain', None)
if dom:
try:
result = socket.gethostbyname(dom)
except:
result = 'domain error, check your input'
if result.startswith('domain'):
ip_info = None
else:
app = current_app._get_current_object()
key = app.config['IP_INFO_DB_KEY']
ip_info = requests.get('http://api.ipinfodb.com/v3/ip-city/?key={0}&ip={1}&format=json'.format(key, result)).json()
return jsonify(status='success', data={'DNS record': result, 'IP infomation': ip_info})
else:
return jsonify(status='error', data='needs domain parameter'), 400
def register_user(**kwargs):
confirmation_link, token = None, None
kwargs['password'] = encrypt_password(kwargs['password'])
user = _datastore.create_user(**kwargs)
_datastore.commit()
if _security.confirmable:
confirmation_link, token = generate_confirmation_link(user)
do_flash(*get_message('CONFIRM_REGISTRATION', email=user.email))
user_registered.send(app._get_current_object(),
user=user, confirm_token=token)
if config_value('SEND_REGISTER_EMAIL'):
send_mail(config_value('EMAIL_SUBJECT_REGISTER'), user.email, 'welcome',
user=user, confirmation_link=confirmation_link)
return user
def _check_token():
header_key = _security.token_authentication_header
args_key = _security.token_authentication_key
header_token = request.headers.get(header_key, None)
token = request.args.get(args_key, header_token)
if request.get_json(silent=True):
if not isinstance(request.json, list):
token = request.json.get(args_key, token)
user = _security.login_manager.token_callback(token)
if user and user.is_authenticated:
app = current_app._get_current_object()
_request_ctx_stack.top.user = user
identity_changed.send(app, identity=Identity(user.id))
return True
return False
def nslookup():
dom = request.args.get('domain', None)
if dom:
try:
result = socket.gethostbyname(dom)
except:
result = 'domain error, check your input'
if result.startswith('domain'):
ip_info = None
else:
app = current_app._get_current_object()
key = app.config['IP_INFO_DB_KEY']
time.sleep(3)
ip_info = requests.get(
'http://api.ipinfodb.com/v3/ip-city/?key={0}&ip={1}&format=json'.format(key, result)).json()
return jsonify(status='success', data={'DNS record': result, 'IP infomation': ip_info})
else:
return jsonify(status='error', data='needs domain parameter'), 400
def send(self, message, envelope_from=None):
"""Verifies and sends message.
:param message: Message instance.
:param envelope_from: Email address to be used in MAIL FROM command.
"""
assert message.send_to, "No recipients have been added"
assert message.sender, (
"The message does not specify a sender and a default sender "
"has not been configured")
if message.has_bad_headers():
raise BadHeaderError
if message.date is None:
message.date = time.time()
if self.host:
self.host.sendmail(sanitize_address(envelope_from or message.sender),
list(sanitize_addresses(message.send_to)),
message.as_bytes() if PY3 else message.as_string(),
message.mail_options,
message.rcpt_options)
email_dispatched.send(message, app=current_app._get_current_object())
self.num_emails += 1
if self.num_emails == self.mail.max_emails:
self.num_emails = 0
if self.host:
self.host.quit()
self.host = self.configure_host()
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(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 __init__(self, thread_ID):
self.thread_ID = thread_ID
threading.Thread.__init__(self)
self.app = current_app._get_current_object()
def set_status(msg, status):
message = {
'msg': msg,
'to_do': 'I do not know what to do, the signal will tell me'
}
status_changed.send(current_app._get_current_object(), status="go_to_work")
status = get_status_callback()
return message, status
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(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(self, message):
"""Verifies and sends message.
:param message: Message instance.
"""
assert message.recipients, "No recipients have been added"
assert message.sender, (
"The message does not specify a sender and a default sender "
"has not been configured")
if message.has_bad_headers():
raise BadHeaderError
if message.date is None:
message.date = time.time()
if self.host:
self.host.sendmail(message.sender,
message.send_to,
message.as_string())
email_dispatched.send(message, app=current_app._get_current_object())
self.num_emails += 1
if self.num_emails == self.mail.max_emails:
self.num_emails = 0
if self.host:
self.host.quit()
self.host = self.configure_host()
def config(self):
return app._get_current_object().config
def super_user_role(self):
return app._get_current_object().config['SUPER_USER_ROLE']
def send(self, message, envelope_from=None):
"""Verifies and sends message.
:param message: Message instance.
:param envelope_from: Email address to be used in MAIL FROM command.
"""
assert message.send_to, "No recipients have been added"
assert message.sender, (
"The message does not specify a sender and a default sender "
"has not been configured")
if message.has_bad_headers():
raise BadHeaderError
if message.date is None:
message.date = time.time()
if self.host:
self.host.sendmail(sanitize_address(envelope_from or message.sender),
list(sanitize_addresses(message.send_to)),
message.as_bytes() if PY3 else message.as_string(),
message.mail_options,
message.rcpt_options)
email_dispatched.send(message, app=current_app._get_current_object())
self.num_emails += 1
if self.num_emails == self.mail.max_emails:
self.num_emails = 0
if self.host:
self.host.quit()
self.host = self.configure_host()
def _dispatching_log(message, level=None):
app = current_app._get_current_object()
if app is None or app.debug:
return _original_log(message, level)
def send_email(to, subject, template, **kwargs):
"""Send email using either Celery, or Thread.
Selection depends on CELERY_INSTEAD_THREADING config variable.
"""
app = current_app._get_current_object()
if app.config['CELERY_INSTEAD_THREADING']:
send_email_celery(to, subject, template, countdown=None, **kwargs)
else:
send_email_thread(to, subject, template, **kwargs)
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_mail(to, subject, template, **kwargs):
app = current_app._get_current_object()
msg = Message(subject, sender=os.environ.get('MAIL_USERNAME'), recipients=[to])
msg.body = render_template(template + '.txt', **kwargs)
msg.html = render_template(template + '.html', **kwargs)
thr = Thread(target=send_mail_async, args=[app, msg])
thr.start()
return thr
def send(self, message, envelope_from=None):
"""Verifies and sends message.
:param message: Message instance.
:param envelope_from: Email address to be used in MAIL FROM command.
"""
assert message.send_to, "No recipients have been added"
assert message.sender, (
"The message does not specify a sender and a default sender "
"has not been configured")
if message.has_bad_headers():
raise BadHeaderError
if message.date is None:
message.date = time.time()
if self.host:
self.host.sendmail(sanitize_address(envelope_from or message.sender),
list(sanitize_addresses(message.send_to)),
message.as_bytes() if PY3 else message.as_string(),
message.mail_options,
message.rcpt_options)
email_dispatched.send(message, app=current_app._get_current_object())
self.num_emails += 1
if self.num_emails == self.mail.max_emails:
self.num_emails = 0
if self.host:
self.host.quit()
self.host = self.configure_host()
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(self, message):
"""Verifies and sends message.
:param message: Message instance.
"""
assert message.recipients, "No recipients have been added"
assert message.sender, (
"The message does not specify a sender and a default sender "
"has not been configured")
if message.has_bad_headers():
raise BadHeaderError
if message.date is None:
message.date = time.time()
if self.host:
self.host.sendmail(message.sender,
message.send_to,
message.as_string())
email_dispatched.send(message, app=current_app._get_current_object())
self.num_emails += 1
if self.num_emails == self.mail.max_emails:
self.num_emails = 0
if self.host:
self.host.quit()
self.host = self.configure_host()
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 get_engine():
app = current_app._get_current_object()
if not hasattr(app, '_sa_engine'):
app._sa_engine = create_engine(get_config('DATABASE_URL'))
return app._sa_engine
def get_metadata():
app = current_app._get_current_object()
if not hasattr(app, '_sa_meta'):
app._sa_meta = MetaData()
app._sa_meta.bind = get_engine()
return app._sa_meta
