Web开发之Flask框架从入门到精通3_核心对象和常用插件

Flask四大内置对象

Request : 使用request获取
Session : 使用session获取
G : 使用g获取
Config : 在模板中使用config获取 , 在python代码中使用app.config获取

四个内置对象可以在view和template中使用

Session 对象

db.session.add(obj) 添加对象
db.session.add_all([obj1,obj2,..]) 添加多个对象
db.session.delete(obj) 删除对象
db.session.commit() 提交会话
db.session.rollback() 回滚
db.session.remove() 移除会话

g对象

@user_router.before_request
def handler():
    g.msg = 'this is msg' #设置g
    print(f'Handling {request.url}')
    
@user_router.route('/',methods=["GET", "POST",])
def employee():
    print(g.msg) # 获取g
    EmployeeModel.query.paginate()
    emp_record = EmployeeModel.query.all()
    return render_template("user/user.html", emp_record=emp_record)

Request,Response对象

参数

url 完整请求地址
base_url 去掉GET参数的URL
host_url 只有主机和端口号的URL
path 路由中的路径
method 请求方法
remote_addr 请求的客户端地址
args GET请求参数 (它并不是get专属，所有请求都能获取这个参数)
form POST请求参数
files 文件上传
headers 请求头
cookies 请求中的cookie

request属性

属性	说明	类型
data	记录请求的数据，并转换为字符串	*
form	记录请求中的表单数据	MultiDict
args	记录请求中的查询参数	MultiDict
cookies	记录请求中的cookie信息	Dict
headers	记录请求中的报文头	EnvironHeaders
method	记录请求使用的HTTP方法	GET/POST
url	记录请求的URL地址	string
files	记录请求上传的文件	*

直接终止请求abort

1
2
3

@blue.route('/')
def index():
    abort(404)

abort源码 : 本质raise Execption

HttpExeception :子类指定两个属性即可实现

code

description

捕获异常errorhandler

1
2
3

@app.errorhandler(404)
def hello(e):
    return 'LOL'

注意 : 本蓝图的errorhandler只能捕获本蓝图的error

创建Response的三种方式

直接返回字符串
make_response
直接构建Response1

render_template : 帮助把模板变成html字符串

重定向 : redirect
反向解析 : url_for

Flask两大核心模块

Jinjia2 : 模板引擎
Werkzurg : WSGI工具集

Flask五个钩子函数

before_first_request
before_request
after_request
teardown_request
errorhandler

Flask的钩子函数有点像Django的中间件 , 本质就是面向切面编程

process_request(self,request)

process_view(self, request, callback, callback_args, callback_kwargs)

process_template_response(self,request,response)

process_exception(self, request, exception)

process_response(self, request, response)

Flask的钩子函数可以用在APP上,也可以用在蓝图上 (APP的优先级更高)

@user_router.before_request
def handler():
    print(f'Handling {request.url}')
# 和Django一样,不必像Scrapy一样必须返回Request

# middleware.py
from flask import request
def load_middleware(app):
    @app.before_request
    def before():
        print(f'Handling {request.url}')
        
    @app.after_request
    def after(response):  # after_request需要传入response参数
        print(f'Handled {request.url}')
        return response # 还需要返回response
    
# app.__init__.py
def create_app():
    load_middleware(app)

面向切面编程AOP :

简单来说 : 动态介入某个流程

可以通过预编译方式和运行期动态代理实现在不修改源代码的情况下给程序动态统一添加功能的一种技术。

利用AOP可以对业务逻辑的各个部分进行隔离，从而使得业务逻辑各部分之间的耦合度降低，提高程序的可重用性，同时提高了开发的效率。

主要功能 : 日志记录，性能统计，安全控制，事务处理，异常处理

上下文

上下文：相当于一个容器，保存了 Flask 程序运行过程中的一些信息。

Flask中有两种上下文

请求上下文
应用上下文

请求上下文(request context)

在视图函数中，如何取到当前请求的相关数据？比如：请求地址，请求方式，cookie等等

在 flask 中，可以直接在视图函数中使用 request 这个对象进行获取相关数据，而 request 就是请求上下文的对象，保存了当前本次请求的相关数据，请求上下文对象有：request、session

request
- 封装了HTTP请求的内容，针对的是http请求。举例：user = request.args.get(‘user’)，获取的是get请求的参数。
session
- 用来记录请求会话中的信息，针对的是用户信息。举例：session[‘name’] = user.id，可以记录用户信息。还可以通过session.get(‘name’)获取用户信息。

应用上下文(application context)

它的字面意思是应用上下文，但它不是一直存在的，它只是request context 中的一个对 app 的代理，所谓local proxy。它的作用主要是帮助 request 获取当前的应用，它是伴 request 而生，随 request 而灭的。

应用上下文对象有：current_app，g

两者区别

请求上下文：保存了客户端和服务器交互的数据
应用上下文：flask 应用程序运行过程中，保存的一些配置信息，比如程序名、数据库连接、应用信息等

Flask常用插件

flask_script

让flask支持命令行参数
使用方法 : 使用APP创建manager对象 , 启动Manager对象

flask_blueprint

可以扩展路由

flask_sqlalchemy

orm , 针对于flask进行优化和封装的Sqlalchemy
使用方法 :
1. 需要App构建SQLalchemy对象
2. 配置DATABASES_URI
3. 使用Models进行模型定制
4. 使用Column创建字段
5. 使用SQLAlchemy对象创建create_all
6. 删除:drop_all

flask_migrate

实现数据库迁移
使用方法 :
1. 初始化需要app和数据库SQLAlchemy
2. 可以和flask_migrate配合使用 : 在Manager上添加命令 MigrateCommend

flask_session

快速实现将session存储到其他位置(如数据库,文件,缓存等) . 并且提供一些设置参数

flask_bootstrap

Flask-Bootstrap把Bootstrap打包进一个扩展，这个扩展主要由一个叫“bootstrap”的蓝本（blueprint）组成。它也可以创建链接从一个CDN上引用Bootstrap资源。
集成了Bootstrap插件 , 为开发者准备了默认模板base.html

Flask-DebugToolbar

此扩展将工具栏覆盖添加到flask应用程序中，其中包含调试所需的有用信息。

Flask-cachinging

添加缓存

flask_mail

发送邮件

Flask_RESTful

实现RESTful接口

flask_migrate使用

Flask-Migrate是一个为Flask应用处理SQLAlchemy数据库迁移的扩展，使得可以通过Flask的命令行接口或者Flask-Scripts对数据库进行操作。

单独使用

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'

db = SQLAlchemy(app)
migrate = Migrate(app, db)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))

结合flask_script使用

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'

db = SQLAlchemy(app)
migrate = Migrate(app, db)

manager = Manager(app)
manager.add_command('db', MigrateCommand)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128))

if __name__ == '__main__':
    manager.run()

之后就可以使用命令:

python manage.py db init

初始化

这个命令将会新建一个名字为migrations的文件夹，并且记录一个数据库版本号，一份保留在migrations中，一份保存在数据库中(新建一张名字为alembic_version的表来保存)，值得注意的是新建了migrations文件夹后需要对数据库模型进行修改，然后使用flask-migrations进行迁移，这样才产生第一个版本号。
python manage.py db migrate

生成迁移文件
python manage.py db upgrade

迁移

每次数据库模型变化，需要重复使用migrate命令和upgrade命令（按顺序组合使用），使用成功后将修改版本号。
python manage.py db downgrade : 取消上一次迁移
python manage.py db --help

帮助

观察一下迁移文件 : 本质就是一段python脚本

"""add note timestamp

Revision ID: 7f3dae8cae4d
Revises: 
Create Date: 2019-04-01 21:56:32.469000

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '7f3dae8cae4d'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('draft')
    op.drop_table('post')
    op.drop_table('comment')
    op.add_column('note', sa.Column('timeStamp', sa.String(length=70), nullable=True))
    op.create_unique_constraint(None, 'note', ['timeStamp'])
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(None, 'note', type_='unique')
    op.drop_column('note', 'timeStamp')
    op.create_table('comment',
    sa.Column('id', sa.INTEGER(), nullable=False),
    sa.Column('body', sa.TEXT(), nullable=True),
    sa.Column('post_id', sa.INTEGER(), nullable=True),
    sa.ForeignKeyConstraint(['post_id'], [u'post.id'], ),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('post',
    sa.Column('id', sa.INTEGER(), nullable=False),
    sa.Column('title', sa.VARCHAR(length=50), nullable=True),
    sa.Column('body', sa.TEXT(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    op.create_table('draft',
    sa.Column('id', sa.INTEGER(), nullable=False),
    sa.Column('body', sa.TEXT(), nullable=True),
    sa.Column('edit_time', sa.INTEGER(), nullable=True),
    sa.PrimaryKeyConstraint('id')
    )
    # ### end Alembic commands ###

从上面的代码可以看出，迁移脚本主要包含了两个函数：

upgrate()函数用来将改动应用到数据库，函数中包含了向表中添加timestamp字段的命令，
而downgrade()函数用来撤消改动，包含了删除timestamp字段的命令。

所以我们执行python manage.py db upgrade 和 python manage.py db downgrade 其实就是执行这两个函数

使用flask-script和flask-migrate

使用flask-script管理数据库创立，此外flask-migrate也支持flask-script的命令行接口，所以可以用flask-script统一管理。

flask-migrate提供了一个ManagerCommand类，可以附加在flask-script的Manager类实例上。
使用add_command()添加一个shell命令。使用shell命令将进入python shell状态，由于将db加入了上下文，可以使用shell手动创建数据库

会话技术

跨请求共享数据
出现原因
1. web开发中http都是短连接
2. http请求是无状态的
3. 请求从request到response就结束了
Cookie
Session
Token

设置获取cookies

@blue.route('/test/')
def test():
    resp = Response('登录成功')
    resp.set_cookies('username',username)
    return resp


@blue.route('/test/')
def test():
    username = reques.cookies.get('username')

设置获取Session

@blue.route('/test/')
def test():
    resp = Response('登录成功')
    session['username'] = 'hyl'
    return resp

@blue.route('/test/')
def test():
    username = session.get('username')

flask中的cookie默认对中文进行了处理，直接可以使用中文

Django对session序列化到数据库中 , 但是Flask默认是存储到内存中
Django是将Session的Key作为SessionID传给cookies , flask将整个session序列化,然后将这整个序列化的session存到cookies中
也就是说:
1
2
3
4
5
@blue.route('/test/')
def test():
 session['username'] = 'hyl'
 session['password'] = '110'
 return Response('登录成功')
Django就是将username,password这两个键作为cookies的值

Flask是将{‘username’:’hyl’,’password’:’110’}序列化后作为cookies的值

这就造成这样的现象 :

用户A请求网页B,Flask为用户A设置了session.之后flask挂掉了重新启动(也就是说内存里面已经没有session了).但是用户A依旧能使用这个session

原因 : Flask的session保存了完整的信息,Flask只要在接收后解密就行了
实际Flask对session的操作:

将session数据序列化

生成签名,hash

base64编码

组装在一条数据上,存储在客户端

flask_session使用

实现了服务端session
将数据存储服务端，将数据对应的key存储在cookies中
flask_session是嵌入级的插件,不需要修改flask源码 , 只需要配置redis就可以了

from flask import Flask, session
from flask.ext.session import Session

app = Flask(__name__)
# 把session存在了redis里面
SESSION_TYPE = 'redis'
app.config.from_object(__name__)
# 初始化app
Session(app)

@app.route('/set/')
def set():
    session['key'] = 'value'
    return 'ok'

@app.route('/get/')
def get():
    return session.get('key', 'not set')

有两种使用模式。

一种是使用app初始化实例：

1 2	app = Flask(__name__) Session(app)

第二种是创建对象并稍后配置应用程序：(懒加载)

sess = Session()

def create_app():
    app = Flask(__name__)
    sess.init_app(app)
    return app

参数:

SESSION_COOKIE_NAME : 设置返回给客户端的cookie的名称，默认是“session”;放置在response的头部；
SESSION_COOKIE_DOMAIN : 设置会话的域，默认是当前的服务器，因为Session是一个全局的变量，可能应用在多个app中；
SESSION_COOKIE_PATH : 设置会话的路径，即哪些路由下应该设置cookie，如果不设置，那么默认为‘/’，所有的路由都会设置cookie；
SESSION_COOKIE_HTTPONLY : cookie应该和httponly标志一起设置，默认为True，这个一般采用默认。
SESSION_COOKIE_SECURE : cookie是否和安全标志一起设置，默认为false，这个一般采用默认。
PERMANENT_SESSION_LIFETIME : 设置session的有效期，即cookie的失效时间，单位是s。这个参数很重要，因为默认会话是永久性的。
SESSION_TYPE

SESSION_TYPE = ‘null’ : 采用flask默认的保存在cookie中；
SESSION_TYPE = ‘redis’ : 保存在redis中
SESSION_TYPE = ‘memcached’ : 保存在memcache
SESSION_TYPE = ‘filesystem’ : 保存在文件
SESSION_TYPE = ‘mongodb’ : 保存在MongoDB
SESSION_TYPE = ‘sqlalchemy’ : 保存在关系型数据库
SESSION_PERMANENT : 是否使用永久会话，默认True，但是如果设置了PERMANENT_SESSION_LIFETIME，则这个失效；
SESSION_USE_SIGNER : 是否为cookie设置签名来保护数据不被更改，默认是False；如果设置True,那么必须设置flask的secret_key参数；
SESSION_KEY_PREFIX : 在所有的会话键之前添加前缀，对于不同的应用程序可以使用不同的前缀；默认“session:”，即保存在redis中的键的名称前都是以“session:”开头SESSION_KEY_PREFIX = 'session:'；
SESSION_REDIS : 如果SESSION_TYPE = ‘redis’，那么设置该参数连接哪个redis，其是一个连接对象；如果不设置的话，默认连接127.0.0.1:6379/0

SESSION_REDIS = redis.StrictRedis(host=”127.0.0.1”, port=6390, db=4)

Flask-caching使用

使用

from flask import Flask
from flask_caching import Cache

cache = Cahce(config={'CACHE_TYPE':'simple'})
app = Flask(__name__)
cache.init_app(app)

# view.py
# 缓存时间为50秒
@app.route('/index/')
@cache.cached(timeout=50)
def index():
    return render_template('index.html')

这里的存储是存储整个request , 也就是说50秒内只会在第一次执行index函数 , 其余的都是执行从存缓中返回html代码

使用redis

from flask import Flask
from flask_caching import Cache

cache = Cahce(config={'CACHE_TYPE':'redis'})
app = Flask(__name__)
cache.init_app(app)

# view.py
@app.route('/index/')
def index():
   username =  cache.get('username') # 获取
    cache.set('username':session.get('username')) # 设置
    return render_template('index.html')

Flask-DebugToolbar使用

from flask import Flask
from flask_debugtoolbar import DebugToolbarExtension

app = Flask(__name__)

app.debug = True

# set a 'SECRET_KEY' to enable the Flask session cookies
app.config['SECRET_KEY'] = '<replace with a secret key>'

toolbar = DebugToolbarExtension()
app = create_app('the-config.cfg')
toolbar.init_app(app)

flask_mail使用

from flask import Flask
from flask_mail import Mail

mail = Mail()
app = Flask(__name__)
mail.init_app(app)

# config
MAIL_DEBUG = True
# 邮箱服务器
MAIL_SERVER = 'smtp.163.com'
# 端口号
MAIL_PORT = 465
# 邮箱账号
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
# 邮箱授权码
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
# 邮件发件人名字
MAIL_NAME = os.environ.get('MAIL_NAME')
# 使用SSL协议
MAIL_USE_SSL = True
MAIL_USE_TLS = False
# 邮件主题
SERVERS_EXPIRES_MAIL_SUBJECT = '服务器过期提醒'
DOMAINS_EXPIRES_MAIL_SUBJECT = '域名过期提醒'

def send_domains_email():
    from serve import app
        subject = app.config.get('DOMAINS_EXPIRES_MAIL_SUBJECT')
        sender = (app.config.get('MAIL_NAME'),app.config.get('MAIL_USERNAME'))
        template = 'domains/expires.html'
        for contact, domains in d.items():
            try:
                with app.app_context():
                    msg = Message(subject, sender=sender, recipients=[contact.email])
                    msg.html = render_template(template, domains=domains, contact=contact)
                    mail.send(msg)
            except ConnectionRefusedError as e:
                app.logger.error('domains: %s, contact: %s' % (domains, contact.name))
            else:
                app.logger.info('domains: %s, contact: %s' % (domains, contact.name))

Flask_RESTful使用

使用Resource

from flask import Flask
from flask_restful import Resource,Api

app = Flask(__name__)
api.init_app(app)

# 定义CBV
class HelloWorld(Resource):
    def get(self,id):
        return {'hello':'world'}
    
    def post(self,id):
        # 输出字典 , 后台会直接将其序列化
        return {'hello':'world'}
    
# 将CBV注册到url中
api.add_resource(HelloWorld,'/user/<int:id>')


if __name__ == '__main__':
    app.run(debug=True)

初始化 : 使用APP进行Api的初始化
创建资源:
- 继承Resource
- 和视图行为基本一致
- CBV
注册资源 : 在API上添加

RESTful接口返回的JSON格式:

// 单个对象
{
    'status':200,
    'msg':'ok',
    'data': {
        'property':'value',
        'property':'value',
    }
}

// 多个对象
{
    'status':200,
    'msg':'ok',
    'data': [
        {
            'property':'value',
            'property':'value',
        },
        {
             'property':'value',
            'property':'value',
        }
    ]
}

序列化

from flask_restful import Resource , marshal

goods_fields = {
    "g_name":fields.String,
    "g_price":fields.Float
}

class GoodListResource(Resource):
    def post(self):
        goods = GoodModel.query.first()
        data = {
            'msg':'ok',
            'data':marshal(goods,goods_fields)
        }

返回复杂JSON

@alarm.route('/show_alarm', methods=['GET'])
def show_alarm_person():
    data = AlarmPerson.get_alarm_person() or []
    return JsonpResp().success({"alarms": data})

import simplejson as json
from flask import Response

class JsonpResp(object):
    def __init__(self, *args):
        self.callback = args[0] if len(args) >= 1 else None

    def _build(self, jsonify_result):
        reponse = Response(jsonify_result)
        if self.callback:
            reponse = Response(
                "%s(%s)" % (self.callback, jsonify_result),
                mimetype="application/javascript"
            )
        return reponse

    def success(self, data, page_status=None):
        """处理成功状态下的jsonp请求"""
        d = {"state": 1, "code": "200", "msg": "success", "data": data}
        if page_status is not None:
            d.update({"page_status": page_status})
        return self._build(json.dumps(d, ensure_ascii=False))

    def error(self, code, msg):
        """处理错误状态下的jsonp请求"""
        return self._build(
            json.dumps(
                {"state": 0, "code": code, "msg": msg},
                ensure_ascii=False
            )
        )

使用use_args

from webargs.flaskparser import use_kwargs, use_args

@alarm.route('/delete_alarm', methods=["GET", "POST"])
@use_args({"id": fields.Int()})
def delete_alarm_person(args):
    employee = Employee.get_by_id(args["id"])

class DepartmentSchema(Schema):
    '''部门数据Schema'''
    id = Integer(allow_none=True)
    name = String(allow_none=True, validate=Length(
        max=Department.name.type.length))


class ListDepartmentSchema(DepartmentSchema, ListSchema):
    '''部门数据列表的参数Schema'''
    pass

#######################################

@department_router.route('/api/departments', methods=['GET'])
@login_required
@use_args(ListDepartmentSchema(), locations=('query',))
def list_departments(args):
    '''视图department.list_departments 部门列表'''
    departments, count = query_data(Department, args)
    return jsonify({
        'data': dump(departments, many=True),
        'itemsCount': count
    })

https://www.bilibili.com/video/av67507215/?p=51