Flask

一：web框架Django和Flask本质

socket服务端

?
wsgi： Web服务网关接口
    - wsgiref           # Django内部内置模块
    - werkzeug          # Flask安装完成后，内部默认已经安装好werkzeug

from werkzeug.wrappers import Request, Response
from werkzeug.serving import run_simple@Request.application
def hello(request):return Response('Hello World!')if __name__ == '__main__':run_simple('localhost', 4000, hello)        # hello是回调方法

werkzeug在Flask中完成web框架本质原理

from wsgiref.simple_server import make_serverdef run_server(environ, start_response):start_response('200 OK', [('Content-Type', 'text/html')])return [bytes('<h1>Hello, web!</h1>', encoding='utf-8'), ]if __name__ == '__main__':httpd = make_server('', 8000, run_server)httpd.serve_forever()

wsgiref在Django中完成web框架本质原理

二：简单的Flask

创建Flask s1，生成最简单的代码。运行s1.py文件，flask运行成功。

from flask import Flaskapp = Flask(__name__)# 路由映射关系
@app.route('/')
def hello_world():return 'Hello World!'if __name__ == '__main__':app.run()

s1.py

三：配置

以下配置参数为app = Flask(__name__)的参数，查看源码类Flask __init__中可传的参数

import_name,                         # 就是Flask(__name__)中的__name__，一般写__name__
static_path=None,                    # 静态文件路径，这个即将被废弃了
static_url_path=None,                  # 静态前缀：static_url_path = '/sssss'。创建flask时目录被默认创建为/static,未配置该参数时，访问127.0.0.1:5000/static/1.jpg就可访问/static目录下的图片但是修改配置后直接访问127.0.0.1:5000/sssss/1.jpg就可访问/static目录下的图片
static_folder='static',                # 静态文件目录，创建Flask时目录/static被默认创建
template_folder='templates',           # 模板路径，创建Flask时目录/templates被默认创建。from flask import Flask,render_template    return render_template('hello.html')
instance_path=None,                    # C:\Users\Administrator\PycharmProjects\s133\instance，用的少，默认是路径，当前目录 + \instance
instance_relative_config=False,        # 当为True，会默认去C:\Users\Administrator\PycharmProjects\s133\instance找配置文件。如果为Flase时，不管它。
root_path=None                         # C:\Users\Administrator\PycharmProjects\s133，当前目录。默认在当前目录找配置文件instance_relative_config=True时，默认去C:\Users\Administrator\PycharmProjects\s133\instance找配置文件

View Code

以下配置为flask.config.Config对象（继承字典）的默认参数

{'DEBUG':                                get_debug_flag(default=False),  # 是否开启Debug模式'TESTING':                              False,                          # 是否开启测试模式'PROPAGATE_EXCEPTIONS':                 None,                          'PRESERVE_CONTEXT_ON_EXCEPTION':        None,'SECRET_KEY':                           None,'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),                # session的超时时间'USE_X_SENDFILE':                       False,'LOGGER_NAME':                          None,'LOGGER_HANDLER_POLICY':               'always','SERVER_NAME':                          None,'APPLICATION_ROOT':                     None,'SESSION_COOKIE_NAME':                  'session','SESSION_COOKIE_DOMAIN':                None,'SESSION_COOKIE_PATH':                  None,'SESSION_COOKIE_HTTPONLY':              True,'SESSION_COOKIE_SECURE':                False,'SESSION_REFRESH_EACH_REQUEST':         True,'MAX_CONTENT_LENGTH':                   None,'SEND_FILE_MAX_AGE_DEFAULT':            timedelta(hours=12),'TRAP_BAD_REQUEST_ERRORS':              False,'TRAP_HTTP_EXCEPTIONS':                 False,'EXPLAIN_TEMPLATE_LOADING':             False,'PREFERRED_URL_SCHEME':                 'http','JSON_AS_ASCII':                        True,'JSON_SORT_KEYS':                       True,'JSONIFY_PRETTYPRINT_REGULAR':          True,'JSONIFY_MIMETYPE':                     'application/json','TEMPLATES_AUTO_RELOAD':                None,
}

View Code

app.config['DEBUG'] = True    # 进入调试模式
app.debug = True            # 进入调试模式
app.session_interface        # session的接口
app.config.updata({})

配置方式一：(s1.py中通过操作字典的方式)

第一种：去一个.py文件中导入配置,例如flask目录下创建一个settings.py，与staic目录同一级别s133.py:app.config.from_pyfile("settings.py")settings.py:DEBUG = True第二种：环境变量中取app.config.from_envvar("环境变量名称"),内部调用from_pyfile方法使用：test.py:import osos.environ['xxxxx'] = "settings"    # 或者os.environ['xxxxx'] = "settings.py"，settings加入环境变量
        s133.py:app.config.from_envvar("xxxxx")        # 找到settings对象，然后执行第一种app.config.from_pyfile("settings.py")
第三种：同第一种方式，创建json.py文件,s133.py中调用from_json方法
·    app.config.from_json("json文件名称")JSON文件名称，必须是json格式，因为内部会执行json.loads第四种：字典的格式app.config.from_mapping({'DEBUG':True})第五种：比较推荐使用的，注意要写大写，小写是导入不成功的。app.config.from_object("settings.TestingConfig")settings.py：class Config(object):DEBUG = FalseTESTING = FalseDATABASE_URI = 'sqlite://:memory:'class ProductionConfig(Config):DATABASE_URI = 'mysql://user@localhost/foo'class DevelopmentConfig(Config):DEBUG = Trueclass TestingConfig(Config):TESTING = True

配置方式二

四：路由

路由使用：

@app.route('/')
def hello_world():return 'Hello World!'

方法一：装饰器方式

def hello_world():# 反向生成urlfrom flask import url_forurl = url_for('xxx')        # url此时为 /             return 'Hello World!'
app.add_url_rule('/',view_func=hello_world,endpoint='xxx',methods=["GET","POST"])    # view_func视图函数；endpoint和django中的name一样，反向生成url，不加endpoint，endpoint默认值为视图函数名

方式二：

url正则匹配：

@app.route('/edit/<int:nid>')
def hello_world(nid):return 'Hello World!'

示例

@app.route('/user/<username>')
@app.route('/post/<int:post_id>')
@app.route('/post/<float:post_id>')
@app.route('/post/<path:path>')
@app.route('/login', methods=['GET', 'POST'])

常用的路由系统,django支持自己写正则表达式,flask不支持

DEFAULT_CONVERTERS = {'default':          UnicodeConverter,'string':           UnicodeConverter,'any':              AnyConverter,'path':             PathConverter,'int':              IntegerConverter,'float':            FloatConverter,'uuid':             UUIDConverter,
}

所有的路由系统都是基于对应关系来处理

from flask import Flask, views, url_for
from werkzeug.routing import BaseConverterapp = Flask(import_name=__name__)class RegexConverter(BaseConverter):"""自定义URL匹配正则表达式"""def __init__(self, map, regex):super(RegexConverter, self).__init__(map)self.regex = regexdef to_python(self, value):"""路由匹配时，匹配成功后传递给视图函数中参数的值:param value: :return: """return int(value)def to_url(self, value):"""使用url_for反向生成URL时，传递的参数经过该方法处理，返回的值用于生成URL中的参数:param value: :return: """val = super(RegexConverter, self).to_url(value)return val+'666'# 添加到flask中
app.url_map.converters['regex'] = RegexConverter# 自定义的url正则的使用
@app.route('/index/<regex("\d+"):nid>')
def index(nid):print(url_for('index', nid='888'))        # 反向生成url /index/888666/ ,反向生成url之前会先执行to_url方法return 'Index'if __name__ == '__main__':app.run()

自定制url正则匹配

方法一：def auth(func):def inner(*args, **kwargs):print('before')result = func(*args, **kwargs)print('after')return resultreturn inner@app.route('/index.html',methods=['GET','POST'],endpoint='index')@authdef index():return 'Index'方法二：def auth(func):def inner(*args, **kwargs):print('before')result = func(*args, **kwargs)print('after')return resultreturn inner        class IndexView(views.MethodView):methods = ['GET']decorators = [auth, ]            # 执行的装饰器def get(self):return 'Index.GET'def post(self):return 'Index.POST'app.add_url_rule('/index', view_func=IndexView.as_view(name='index'))  # name=endpoint

flask中装饰器的使用

rule,                       URL规则
view_func,                  视图函数名称
defaults=None,              默认值,当URL中无参数，函数需要参数时，使用defaults={'nid':9}为函数提供参数
endpoint=None,              名称，用于反向生成URL，即： url_for('名称')
methods=None,               允许的请求方式，如：["GET","POST"]strict_slashes=None,       对URL最后的 / 符号是否严格要求，如：@app.route('/index',strict_slashes=False)，访问 http://www.xx.com/index/ 或 http://www.xx.com/index均可@app.route('/index',strict_slashes=True)仅访问 http://www.xx.com/index redirect_to=None,           重定向到指定地址如：@app.route('/index/<int:nid>', redirect_to='/home/<nid>')    # 请求到来不执行/index/<int:nid>代码，直接重定向到/home/<nid>
        或def func(adapter, nid):return "/home/888"@app.route('/index/<int:nid>', redirect_to=func)subdomain=None,             子域名访问如：from flask import Flask, views, url_forapp = Flask(import_name=__name__)app.config['SERVER_NAME'] = 'xuyaping.com:5000'        # 必须写，才能支持子域名
@app.route("/index/", subdomain="admin")                    # 访问http://admin/xuyaping.com:5000/index/        def static_index():"""Flask supports static subdomainsThis is available at static.your-domain.tld"""return "static.your-domain.tld"if __name__ == '__main__':app.run()

@app.route和app.add_url_rule参数

五：模板

模板的使用
　　Flask使用的是Jinja2模板，所以其语法和Django无差别　

　　不过在django模板中执行函数或方法时，不用加括号就会自己执行，而Flask必须自己加括号才会执行。

　　flask中的Markup等价django的mark_safe

自定义模板方法
　　创建一个函数并通过参数的形式传入render_template，如：

<!DOCTYPE html>
<html>
<head lang="en"><meta charset="UTF-8"><title></title>
</head>
<body><h1>自定义函数</h1>{{xyp()|safe}}</body>
</html>

html

from flask import Flask,render_template
app = Flask(__name__)def index():return '<h1>index</h1>'@app.route('/login', methods=['GET', 'POST'])
def login():return render_template('login.html', ss=index)app.run()

run.py:

六：请求和响应

from flask import Flask
from flask import request
from flask import render_template
from flask import redirect
from flask import make_responseapp = Flask(__name__)@app.route('/login.html', methods=['GET', "POST"])
def login():# 请求相关信息# request.method# request.args                # GET传的参数# request.form                # 表单，POST传的参数# request.values# request.cookies# request.headers# request.path# request.full_path# request.script_root# request.url# request.base_url# request.url_root# request.host_url# request.host# request.files                # 文件# obj = request.files['the_file_name']# obj.save('/var/www/uploads/')                # save直接把文件存储到/var/www/uploads/目录中了# 响应相关信息# return "字符串"                                # 相当于django中的Httpresponse# return render_template('html模板路径',**{})    # 相当于django中的render# return redirect('/index.html')                # 相当于django中的redirect# response = make_response(render_template('index.html'))        # make_response把返回的数据封装起来，然后就有了delete_cookie、set_cookie、headers方法了# response是flask.wrappers.Response类型# response.delete_cookie('key')# response.set_cookie('key', 'value')# response.headers['X-Something'] = 'A value'# return responsereturn "内容"if __name__ == '__main__':app.run()

View Code

七：session

?
flask内置session默认放在加密Cookie中，依赖于session.secret_key   
设置：session['username'] ＝ 'xxx'
删除：session.pop('username', None)

自定义session及使用

import uuid
import json
from flask.sessions import SessionInterface
from flask.sessions import SessionMixin
from itsdangerous import Signer, BadSignature, want_bytesclass MySession(dict, SessionMixin):def __init__(self, initial=None, sid=None):self.sid = sidself.initial = initialsuper(MySession, self).__init__(initial or ())def __setitem__(self, key, value):super(MySession, self).__setitem__(key, value)def __getitem__(self, item):return super(MySession, self).__getitem__(item)def __delitem__(self, key):super(MySession, self).__delitem__(key)class MySessionInterface(SessionInterface):session_class = MySessioncontainer = {}def __init__(self):import redisself.redis = redis.Redis()def _generate_sid(self):return str(uuid.uuid4())def _get_signer(self, app):if not app.secret_key:return Nonereturn Signer(app.secret_key, salt='flask-session',key_derivation='hmac')def open_session(self, app, request):"""程序刚启动时执行，需要返回一个session对象"""sid = request.cookies.get(app.session_cookie_name)if not sid:sid = self._generate_sid()return self.session_class(sid=sid)signer = self._get_signer(app)try:sid_as_bytes = signer.unsign(sid)sid = sid_as_bytes.decode()except BadSignature:sid = self._generate_sid()return self.session_class(sid=sid)# session保存在redis中# val = self.redis.get(sid)# session保存在内存中val = self.container.get(sid)if val is not None:try:data = json.loads(val)return self.session_class(data, sid=sid)except:return self.session_class(sid=sid)return self.session_class(sid=sid)def save_session(self, app, session, response):"""程序结束前执行，可以保存session中所有的值如：保存到resit写入到用户cookie"""domain = self.get_cookie_domain(app)path = self.get_cookie_path(app)httponly = self.get_cookie_httponly(app)secure = self.get_cookie_secure(app)expires = self.get_expiration_time(app, session)val = json.dumps(dict(session))# session保存在redis中# self.redis.setex(name=session.sid, value=val, time=app.permanent_session_lifetime)# session保存在内存中
        self.container.setdefault(session.sid, val)session_id = self._get_signer(app).sign(want_bytes(session.sid))response.set_cookie(app.session_cookie_name, session_id,expires=expires, httponly=httponly,domain=domain, path=path, secure=secure)

sessions.py

from sessions import MySessionInterface
app.session_interface = MySessionInterface()

使用

或者使用flask-session模块，配置文件中设置

from flask import Flask
from flask import session
from pro_flask.utils.session import MySessionInterface
app = Flask(__name__)app.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
app.session_interface = MySessionInterface()@app.route('/login.html', methods=['GET', "POST"])
def login():print(session)session['user1'] = 'alex'session['user2'] = 'alex'del session['user2']return "内容"if __name__ == '__main__':app.run()

使用flask-session

八：message

message是一个基于Session实现的用于保存数据的集合，其特点是：使用一次就删除。

from flask import Flask, flash, redirect, render_template, request, get_flashed_messagesapp = Flask(__name__)
app.secret_key = 'some_secret'@app.route('/')
def index1():messages = get_flashed_messages()        # 从session中取，取到就删掉print(messages)return "Index1"@app.route('/set')
def index2():v = request.args.get('p')flash(v)            # 存储在session中return 'ok'if __name__ == "__main__":app.run()

View Code

九：扩展：伪中间件

from flask import Flask, flash, requestapp = Flask(__name__)
app.secret_key = 'some_secret'@app.route('/index')
def index():return 'index.html'# 中间件
class MiddleWare:def __init__(self,wsgi_app):self.wsgi_app = wsgi_appdef __call__(self, environ, start_response):        #  environ, start_response是wsgi socket传的参数print('before')response = self.wsgi_app(environ, start_response)print('after')return responseif __name__ == "__main__":app.wsgi_app = MiddleWare(app.wsgi_app)app.run(port=9999)

View Code

十：Flask插件

?
WTForms    　　    form组件，做form表单验证的组件
SQLAchemy　　    ORM操作
Flask-Session　　 session插件

十一：蓝图

蓝图的功能就是将不同功能放在不同的py文件中

eg：

order.py

+ View Code?
from flask import Blueprint
order = Blueprint('order',__name__)
@order.route('/order')
def order():
    return 'Order'

account.py

+ View Code?
from flask import Blueprint,render_template
account = Blueprint('account',__name__)
@account.route('/login')
def login():
    return render_template('login.html')

_init_.py

+ View Code?
from flask import Flask
from .views import account
from .views import order
app = Flask(__name__)
app.register_blueprint(account.account)
app.register_blueprint(order.order)

十二：数据库连接池

"""
为每个线程创建一个连接，thread.local实现。"""from DBUtils.PersistentDB import PersistentDB
import pymysqlPOOL = PersistentDB(creator=pymysql,  # 使用链接数据库的模块maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayscloseable=False,# 如果为False时， conn.close() 实际上被忽略，供下次使用，再线程关闭时，才会自动关闭链接。如果为True时， conn.close()则关闭链接，那么再次调用pool.connection时就会报错，因为已经真的关闭了连接（pool.steady_connection()可以获取一个新的链接）threadlocal=None,  # 本线程独享值得对象，用于保存链接对象，如果链接对象被重置host='127.0.0.1',port=3306,user='root',password='123',database='pooldb',charset='utf8'
)def func():# conn = SteadyDBConnection()conn = POOL.connection()cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()cursor.close()conn.close() # 不是真的关闭，而是假的关闭。 conn = pymysql.connect()   conn.close()
conn = POOL.connection()cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()cursor.close()conn.close()import threadingfor i in range(10):t = threading.Thread(target=func)t.start()

模式一

import time
import pymysql
import threading
from DBUtils.PooledDB import PooledDB, SharedDBConnection
POOL = PooledDB(creator=pymysql,  # 使用链接数据库的模块maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
maxcached=5,  # 链接池中最多闲置的链接，0和None不限制maxshared=3,  # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]ping=0,# ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = alwayshost='127.0.0.1',port=3306,user='root',password='123',database='pooldb',charset='utf8'
)def func():# 检测当前正在运行连接数的是否小于最大链接数，如果不小于则：等待或报raise TooManyConnections异常# 否则# 则优先去初始化时创建的链接中获取链接 SteadyDBConnection。# 然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。# 如果最开始创建的链接没有链接，则去创建一个SteadyDBConnection对象，再封装到PooledDedicatedDBConnection中并返回。# 一旦关闭链接后，连接就返回到连接池让后续线程继续使用。# PooledDedicatedDBConnectionconn = POOL.connection()# print(th, '链接被拿走了', conn1._con)# print(th, '池子里目前有', pool._idle_cache, '\r\n')
cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()conn.close()conn = POOL.connection()cursor = conn.cursor()cursor.execute('select * from tb1')result = cursor.fetchall()conn.close()func()