Flask 快速使用 进阶——

Posted 2021-01-03 crazymagic

案例：可配置发送信息的系统

假如我们有这样的一个系统，可以发送短息，邮件和微信，后期可能还会增加一些平台，怎么才可以做到快速切换的去使用某种功能呢，在这里我会通过在配置文件中简单的配置就可以实现

在项目的根目录下创建  utils/message/base.py  创建继承的基类，强制其子类实现send() 方法

该系统可适用于 各种缓存系统

class Base(object):

    def send(self,msg):
        raise NotImplementedError(‘...‘)

 

在项目的根目录下创建  utils/message/email.py 实现发送邮件的类

from .base import Base

class Email(Base):
    """
    发送邮件提醒相关
    """
    def __init__(self):
        """
        邮箱相关数据初始化
        """
        self.username = ‘asdf‘
        self.pwd = ‘asdf‘

    def send(self,msg):
        print(‘email‘)

在项目的根目录下创建  utils/message/msg.py 实现发送短息的类 

from .base import Base

class Msg(Base):
    """
    发送短信提醒相关
    """

    def __init__(self):
        """
        短信相关数据初始化
        """
        self.username = ‘asdf‘
        self.pwd = ‘asdf‘

    def send(self, msg):
        print(‘msg‘)

 

在项目的根目录下创建  utils/message/wechat.py 实现发送微信的类 

from .base import Base


class WeChat(Base):

    def send(self,msg):
        print(‘weixin‘)

 

 在项目的根目录下创建  utils/message/__init__.py 通过读取配置文件配置信息，动态的导入和执行send()方法

import settings
import importlib

def send_msgs(msg):

    for path in settings.MSG_LIST:
        m,c = path.rsplit(‘.‘,maxsplit=1) # 从右边开始分割  utils.message.wechat WeChat
        md = importlib.import_module(m)
        obj = getattr(md,c)()
        obj.send(msg)

 在项目的根目录下创建  settings.py  配置文件信息

MSG_LIST = [
    "utils.message.email.Email",
    "utils.message.msg.Msg",
    "utils.message.wechat.WeChat",
]

 在项目文件下创建主程序app.py

from flask import Flask,request

from utils.message import send_msgs

app = Flask(__name__)

@app.route(‘/‘)
def index():
    send_msgs(‘.....‘)

    return ‘Hello World!‘


if __name__ == ‘__main__‘:
    app.run()

 

 访问

http://127.0.0.1:5000、　　

 后台打印数据如下

 闪现

 在服务端可以对数据进行临时的存储，本质是基于session做的，取的时候清除

 

from flask import Flask,flash,get_flashed_messages

app = Flask(__name__)
app.secret_key = ‘asdfasdf‘
@app.route(‘/get‘)
def get():
    # 从某个地方获取设置过的所有值，并清除。
    data = get_flashed_messages()
    print(data)
    return ‘Hello World!‘


@app.route(‘/set‘)
def set():
    # 向某个地方设置一个值
    flash(‘阿斯蒂芬‘)

    return ‘Hello World!‘


if __name__ == ‘__main__‘:
    app.run()

 

 简单的小应用

from flask import Flask,flash,get_flashed_messages,request,redirect

app = Flask(__name__)
app.secret_key = ‘asdfasdf‘


@app.route(‘/index‘)
def index():
    # 从某个地方获取设置过的所有值，并清除。
    val = request.args.get(‘v‘)
    if val == ‘oldboy‘:
        return ‘Hello World!‘
    flash(‘超时错误‘,category="x1")
    return "ssdsdsdfsd"

@app.route(‘/error‘)
def error():
    """
    展示错误信息
    :return:
    """
    data = get_flashed_messages(category_filter=[‘x1‘])
    if data:
        msg = data[0]
    else:
        msg = "..."
    return "错误信息：%s" %(msg,)

if __name__ == ‘__main__‘:
    app.run()

请求扩展

1 before_request 在每次请求前执行 2 after_request在每次请求后执行，必须有返回值

可以定义多个，before_request 按代码的顺序一次执行，after_request按代码的顺序逆序执行

注意请求假如没通过，被拦截，所有的response还会被执行

from flask import Flask,render_template,request,redirect,session,url_for
app = Flask(__name__)
app.debug = True
app.secret_key = ‘siuljskdjfs‘

@app.before_request
def process_request1(*args,**kwargs):
    print(‘process_request1 进来了‘)

@app.before_request
def process_request2(*args,**kwargs):
    print(‘process_request2 进来了‘)
    
@app.after_request
def process_response1(response):
    print(‘process_response1 走了‘)
    return response

@app.after_request
def process_response2(response):
    print(‘process_response2 走了‘)
    return response

@app.errorhandler(404)
def error_404(arg):
    return "404错误了"

@app.before_first_request
def first(*args,**kwargs):
    pass

@app.route(‘/index‘,methods=[‘GET‘])
def index():
    print(‘index函数‘)
    return "Index"

if __name__ == ‘__main__‘:
    app.run()

 

访问

http://127.0.0.1:5000/index

　　

后台打印如下

 

基于请求做的登陆的认证

from flask import Flask,render_template,request,redirect,session,url_for
app = Flask(__name__)
app.secret_key = ‘siuljskdjfs‘

@app.before_request
def process_request(*args,**kwargs):
    if request.path == ‘/login‘:
        return None
    user = session.get(‘user_info‘)
    if user:
        return None
    return redirect(‘/login‘)

@app.route(‘/detail/‘)
def detail():
    return ‘详情‘

@app.route(‘/index‘)
def index():
    return ‘主页‘

@app.route(‘/login‘,methods=[‘GET‘,‘POST‘])
def login():
    session[‘user_info‘] = ‘zhang‘
    return ‘登陆成功‘

if __name__ == ‘__main__‘:
    app.run()

 

中间键 

flask 每次请求进来都会执行app.__call__方法中的 wsgi_app方法，我们可以对其中的 wsgi_app 进行封装

 

from flask import Flask

app = Flask(__name__)

@app.route(‘/‘)
def index():
    return ‘Hello World!‘

class Md(object):
    def __init__(self,old_wsgi_app):
        self.old_wsgi_app = old_wsgi_app

    def __call__(self,  environ, start_response):
        print(‘开始之前‘)
        ret = self.old_wsgi_app(environ, start_response)
        print(‘结束之后‘)
        return ret

if __name__ == ‘__main__‘:
    app.wsgi_app = Md(app.wsgi_app)
    app.run()

 

访问

http://127.0.0.1:5000/ 

 测试结果如下

简单的定制错误信息

@app.errorhandler(404)
def error_404(arg):
    return "404错误了"

 模板中的定制方法

from flask import Flask,render_template,request,redirect,session,url_for
app = Flask(__name__)
app.secret_key = ‘siuljskdjfs‘

@app.template_global()
def sb(a1, a2):
    return a1 + a2

@app.template_filter()
def db(a1, a2, a3):
    return a1 + a2 + a3

@app.route(‘/login‘)
def login():
    return render_template(‘login.html‘)
if __name__ == ‘__main__‘:
    app.run()

 

templates/login.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
    <h1>{{ 1|db(2,3)}}</h1>
    <h1>{{sb(1,2)}}</h1>
</body>
</html>

 输入以下地址

http://127.0.0.1:5000/login

　

测试结果如下