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是回调方法

from wsgiref.simple_server import make_server

def 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()

二：简单的Flask

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

from flask import Flask

app = Flask(__name__)


# 路由映射关系
@app.route('/')
def hello_world():
    return 'Hello World!'


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

三：配置

 以下配置参数为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:UsersAdministratorPycharmProjectss133instance，用的少，默认是路径，当前目录 + instance
instance_relative_config=False,        # 当为True，会默认去C:UsersAdministratorPycharmProjectss133instance找配置文件。如果为Flase时，不管它。
root_path=None                         # C:UsersAdministratorPycharmProjectss133，当前目录。默认在当前目录找配置文件instance_relative_config=True时，
                                        默认去C:UsersAdministratorPycharmProjectss133instance找配置文件

 以下配置为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,
}

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

第一种：
    去一个.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 os
            os.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 = False
            TESTING = False
            DATABASE_URI = 'sqlite://:memory:'

        class ProductionConfig(Config):
            DATABASE_URI = 'mysql://user@localhost/foo'

        class DevelopmentConfig(Config):
            DEBUG = True

        class TestingConfig(Config):
            TESTING = True

四：路由

路由使用：

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

def hello_world():
    # 反向生成url
    from flask import url_for
    url = 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'])

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 BaseConverter

app = Flask(import_name=__name__)


class RegexConverter(BaseConverter):
    """
    自定义URL匹配正则表达式
    """
    def __init__(self, map, regex):
        super(RegexConverter, self).__init__(map)
        self.regex = regex

    def 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()

方法一：
    def auth(func):
        def inner(*args, **kwargs):
            print('before')
            result = func(*args, **kwargs)
            print('after')
            return result
    return inner

    @app.route('/index.html',methods=['GET','POST'],endpoint='index')
    @auth
    def index():
        return 'Index'

方法二：
    def auth(func):
        def inner(*args, **kwargs):
            print('before')
            result = func(*args, **kwargs)
            print('after')
            return result
    return 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

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_for

        app = 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 subdomains
            This is available at static.your-domain.tld"""
            return "static.your-domain.tld"

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

五：模板

模板的使用
　　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>

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()

六：请求和响应

from flask import Flask
from flask import request
from flask import render_template
from flask import redirect
from flask import make_response

app = 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 response


    return "内容"

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

七：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_bytes


class MySession(dict, SessionMixin):
    def __init__(self, initial=None, sid=None):
        self.sid = sid
        self.initial = initial
        super(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 = MySession
    container = {}

    def __init__(self):
        import redis
        self.redis = redis.Redis()

    def _generate_sid(self):
        return str(uuid.uuid4())

    def _get_signer(self, app):
        if not app.secret_key:
            return None
        return 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)

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()

八：message

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

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

app = 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()

九：扩展：伪中间件

from flask import Flask, flash, request

app = 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_app
 
    def __call__(self, environ, start_response):        #  environ, start_response是wsgi socket传的参数
        print('before')
        response = self.wsgi_app(environ, start_response)
        print('after')
        return response
 
if __name__ == "__main__":
    app.wsgi_app = MiddleWare(app.wsgi_app)
    app.run(port=9999)

十：Flask插件

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

　　

十一：蓝图

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

eg：

order.py

from flask import Blueprint

order = Blueprint('order',__name__)
@order.route('/order')
def order():
    return 'Order'

account.py

from flask import Blueprint,render_template

account = Blueprint('account',__name__)
@account.route('/login')
def login():
    return render_template('login.html')

_init_.py

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 pymysql

POOL = 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 = always
    closeable=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 threading

for 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 = always
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='pooldb',
    charset='utf8'
)


def func():
    # 检测当前正在运行连接数的是否小于最大链接数，如果不小于则：等待或报raise TooManyConnections异常
    # 否则
    # 则优先去初始化时创建的链接中获取链接 SteadyDBConnection。
    # 然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。
    # 如果最开始创建的链接没有链接，则去创建一个SteadyDBConnection对象，再封装到PooledDedicatedDBConnection中并返回。
    # 一旦关闭链接后，连接就返回到连接池让后续线程继续使用。

    # PooledDedicatedDBConnection
    conn = POOL.connection()

    # print(th, '链接被拿走了', conn1._con)
    # print(th, '池子里目前有', pool._idle_cache, '
')

    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()