Flask

框架学习
发布时间 :
字数:4.6k 阅读 :
  1. Flask
    1. 安装依赖
    2. 响应体
    3. 列表
    4. Request和Session
    5. flask装饰器使用
    6. 蓝图 BluePrint
    7. 数据库连接池
    8. 静态文件目录
    9. 加载配置文件
    10. 路由系统
    11. 请求钩子
    12. Flask工作流程

Flask

Flask 是一个用 Python 编写的轻量级 Web 应用框架。

Flask相比Django更加轻量级，具有齐全的第三方组件库。

flask的模板引擎使用jinja2，web服务使用wsgi

安装依赖

pip install flask

一个简单的Flask Web程序

from flask import Flask

app = Flask(__name__)

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

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

WSGI

本质上Flask底层使用的还是werkzeug这个wsgi工具包来处理网络请求

from werkzeug.serving import run_simple

def service(environ, start_response):
    print('请求来了')
    pass

if __name__ == '__main__':
    run_simple('127.0.0.1',8181,service)

简单内部实现

from werkzeug.serving import run_simple

class Flask(object):
    def __call__(self, environ, start_response):
        return "hello world"
    
    def run(self):
        run_simple('127.0.0.1', 5000, self)

app = Flask()

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

用户请求进入，就会执行Flask的__call__()方法

响应体

视图渲染

需要将启动类和templates文件夹放在同一级目录下，将我们的html文件放在templates文件夹下

from flask import Flask,render_template

app = Flask(__name__)

@app.route('/login')
def login():
    return render_template("login.html")

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

Flask使用jinja2模板引擎来渲染前端视图界面

自定义模板文件夹名称，默认情况下使用templates

app = Flask(__name__,template_folder='templates')

JSON响应体

from flask import Flask,jsonify

app = Flask(__name__)

@app.route('/login')
def login():
    return jsonify({
        "code":200,
        "data":[1,2,3,4]
    })

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

使用jsonify 返回json格式数据

支持POST请求

默认情况下路由方法只支持GET请求

@app.route('/login' , methods=['GET','POST'])

一个简单的登录程序

<!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/html">
<head>
    <meta charset="UTF-8">
    <title>登录页面</title>
</head>
<body>
<h1>登录页面</h1>
<form method="post">
    <table>
        <tr>
            <td>用户名</td>
            <td><input type="text" name="user"></td>
        </tr>
        <tr>
            <td>密码</td>
            <td><input type="password" name="pwd"></td>
        </tr>
        <tr>
            <td><input type="submit" value="提交"/></td>
        </tr>
    </table>
    <br/>
    {{error}}
</form>
</body>
</html>

from flask import Flask,render_template,request,redirect

app = Flask(__name__,template_folder='templates')

@app.route('/login', methods=['GET','POST'])
def login():
    if request.method == 'GET':
        return render_template("login.html")
    if request.method == 'POST':
        if request.form['user'] == 'admin' and request.form['pwd'] == 'admin':
            return redirect("/main")
        else:
            return render_template("login.html",error="用户名或密码错误")

@app.route('/main', methods=['GET','POST'])
def main():
    return render_template('main.html')

if __name__ == '__main__':
    app.run()
  • 通过request.form获取表单提交内容
  • 通过request.method判断请求模式
  • 通过redirect("/url")来重定向页面
  • 通过传递字典信息来渲染登录失败提示

列表

{% %}是jinja2提供的一种模板渲染语法，可以查阅相关文档

@app.route("/index")
def index():
    return render_template('index.html',data_dict = {
        '1':{'name':'bob','age':23},
        '2':{'name':'kate','age':19},
        '3':{'name':'david','age':22},
    })

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>表单</title>
</head>
<body>
<table border="1px">
    <thead>
        <th>序号</th>
        <th>名字</th>
        <th>年龄</th>
    </thead>
    <tbody>
        {% for key,value in data_dict.items() %}
            
                {{key}}
                {{value.name}}
                {{value.age}}
            
        {% endfor %}
    </tbody>
</table>

</body>
</html>

添加编辑、删除功能

两种接收页面参数形式

通过request.args获取?后的参数

通过/url/<flag>的形式获取路径参数

@app.route('/edit', methods=['GET','POST'])
def edit():
    if request.method == 'GET':
        id = request.args.get('id')
        data = DATA_DICT.get(id)
        return render_template('edit.html', name=data['name'], age=data['age'], id=id)
    if request.method == 'POST':
        data_dict = DATA_DICT
        name = request.form.get('name')
        age = int(request.form.get('age'))
        id = request.form['id']
        data = data_dict[id]
        data['name'] = name
        data['age'] = age
        return render_template('index.html',data_dict = data_dict)

@app.route('/delete/<id>')
def delete(id):
    del DATA_DICT[id]
    return redirect("/index")

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>修改界面</title>
</head>
<body>
<form method="post">
    <table>
        <thead>
    <th>名字</th>
    <th>年龄</th>
    </thead>
    <tbody>
    <tr>
        <td><input value="{{name}}" name="name" type="text"/></td>
        <td><input value="{{age}}" name="age" type="text"/></td>
    </tr>
    </tbody>
    </table>
    <input value="{{id}}" name="id" hidden="true">
    <input type="submit" value="提交">
</form>

</body>
</html>

Request和Session

request

在 Flask 中由全局的 request 对象来提供请求信息

常用属性

属性名 介绍
form POST和PUT请求提交的表单信息（MultiDict）
args URL携带参数 ?key=value
cookies 请求的cookies，类型是dict
stream 在可知的mimetype下，如果进来的表单数据无法解码，会没有任何改动的保存到这个 stream 以供使用。很多时候，当请求的数据转换为string时，使用data是最好的方式。这个stream只返回数据一次
headers 请求头，字典类型
data 包含了请求的数据，并转换为字符串，除非是一个Flask无法处理的mimetype
files MultiDict，带有通过POST或PUT请求上传的文件
method 请求方法，比如POST、GET
url 获取全部url
blueprint 蓝图名字
json 如果mimetype是application/json，这个参数将会解析JSON数据，如果不是则返回None

Session

session使得服务器可以识别客户端状态，flask中使用session需要提供secret_key

flask的session信息实际上保存在客户端，但是是通过加密后存储在cookie中的形式存储的

#提供session生成令牌使用的密钥，后端存储
app.secret_key = 'd1niond12f0139jif10edj3d3'

@app.route('/login', methods=['GET','POST'])
def login():
    username = session.get('username')
    if request.method == 'GET':
        if not username:
            return render_template("login.html")
        return redirect("/main")
    if request.method == 'POST':
        if request.form['user'] == 'admin' and request.form['pwd'] == 'admin':
            session['username'] = 'admin'
            return redirect("/main")
        else:
            return render_template("login.html",error="用户名或密码错误")

flask装饰器使用

由于flask中路由是根据函数名称和url来建立路由关系的，默认endpoint使用的是函数的名称，如果不同的路由被同一个装饰器装饰，在没有包装的前提下都会被修改为内函数名称，因此会发送重名错误

在flask中使用装饰器时，我们需要使用functiontools.warp功能为函数进行包装，由于装饰器本身修改了函数名称使其成为内函数，因此我们需要使用该功能使得函数保持原来函数名称

def auth(func):
    @functools.wraps(func) #包装函数
    def wrapper(*args, **kwargs):
        print("auth")
        return func(*args, **kwargs)
    return wrapper

@app.route('/login', methods=['GET','POST'])
@auth
def index():
    #代码逻辑

以上正确的装饰器使用方式，必须要注意使用@functools.warp，导入functools模块即可

蓝图 BluePrint

构建业务功能可拆分的目录结构，帮助我们拆分业务功能构建多个py文件，把各个功能放置到各个蓝图，最后将蓝图注册到flask对象中

使用蓝图的步骤

  1. 创建蓝图：在独立的模块中定义蓝图，并指定路由和视图函数。
  2. 注册蓝图：在主应用中注册蓝图，并设置路由前缀。
  3. 使用蓝图中的模板和静态文件：将模板和静态文件放在蓝图的 templates 和 static 文件夹中。
  4. 使用请求钩子和错误处理：在蓝图中定义请求钩子和错误处理函数。

蓝图项目目录结构

  • app.py
  • auth/
    • __init__.py
    • routes.py
    • static/
      • login.js
      • login.css
    • templates/
      • login.html
      • register.html
  • blog/
    • __init__.py
    • routes.py
    • static/
      • index.js
      • index.css
    • templates/
      • intdex.html
      • post.html

app.py

通常我们在app.py文件中导入每个模块的蓝图，同时注册蓝图

from flask import Flask

app = Flask(__name__)

#导入蓝图
from auth.route import auth
from blog.route import blog

#注册蓝图
app.register_blueprint(auth, url_prefix='/auth')
app.register_blueprint(blog, url_prefix='/blog')

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

在注册蓝图时，可以给路由加上统一前缀

route.py

创建蓝图

from flask import Blueprint,render_template,request,redirect,url_for

auth = Blueprint('auth',__name__,template_folder='templates')

@auth.route('/login')
def login():
    return render_template('index.html')

@auth.route('/logout')
def logout():
    return redirect(url_for('auth.login'))

在我们的具体业务模块中创建route.py文件，在这个文件中，我们就开始创建我们的蓝图对象，同时创建不同路由对应的业务函数

数据库连接池

依赖

pip install dbutils

pip install pymysql

连接池使用

import pymysql
from dbutils.pooled_db import PooledDB

pool = PooledDB(
    creator=pymysql,
    maxconnections=6, #最大连接数量 0和None表示不限制数量
    blocking=True, #连接池中如果没有可用连接后，是否阻塞等待，不等待抛出异常
    ping=0, #检查Mysql服务器是否可用，0表示不检查, 1 = default 请求都会检查, 2 创建cursor时检查, 4 执行sql时检查, 7 全部检查
    host='localhost',
    port=3306,
    user='root',
    passwd='root',
    database='web_test',
    charset='utf8'
)

#去连接池中选择一个连接
con = pool.connection()

cur = con.cursor()
cur.execute("select * from user")
result = cur.fetchall()
print(result)
#将连接放回连接池
con.close()

使用数据库连接池可以支持并发请求的数据库查询，提高了系统效率

多线程下使用单例管理数据库连接池方案

import pymysql
from dbutils.pooled_db import PooledDB
import threading

class SqlHelper(object):


    def __init__(self):
        self.threadLocalMap = {}
        self.pool = PooledDB(
            creator=pymysql,
            maxconnections=6,  # 最大连接数量 0和None表示不限制数量
            blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待，不等待抛出异常
            ping=0,  # 检查Mysql服务器是否可用，0表示不检查, 1 = default 请求都会检查 #2 创建cursor时检查 #4 执行sql时检查 #7 全部检查
            host='localhost',
            port=3306,
            user='root',
            passwd='root',
            database='web_test',
            charset='utf8'
        )

    def open(self):
        connection = self.pool.connection()
        cursor = connection.cursor()
        return connection, cursor

    def close(self, connection, cursor):
        cursor.close()
        connection.close()

    def fetchall(self,sql):
        connection, cursor = self.getCurrentContext()
        cursor.execute(sql)
        return cursor.fetchall()

    def getCurrentContext(self):
        return self.threadLocalMap.get(threading.currentThread().ident)

    def __enter__(self):
        connection, cursor = self.open()
        self.threadLocalMap[threading.currentThread().ident] = (connection, cursor)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        connection, cursor = self.getCurrentContext()
        cursor.close()
        connection.close()

db = SqlHelper()

将数据库连接池交由Sqlhelper这个单例管理，由于python模块单例特性，这边要使用时直接引入db对象即可

在实现时需要考虑到多线程问题，多个线程进入使用db对象，要各自操作对应线程上绑定的连接和游标，因此这边引入一个字典对象来存储各个线程的连接和游标信息，使用各个线程名称作为key值

这种设计以后就可以使用with语法来自由开关闭使用sqlHelper

多线程下测试使用

from threading import Thread

def func():
    from sqlHelper import db
    with db as db:
        result = db.fetchall("select * from user")
        print("result1====>",result)

if __name__ == '__main__':
    for i in range(30):
        t1 = Thread(name=f"测试{i}", target=func)
        t1.start()

但这种设计不支持嵌套，因此下面介绍一种支持嵌套的多线程sqlHelper

使用threadinglocal

def func(num):
    local.num = num
    print(local.num)

if __name__ == '__main__':
    global local
    local = local()
    for i in range(10):
        t = Thread(target=func, args=(i,))
        t.start()

threadlocal为每个线程开辟了一个空间，当不同线程进入使用local时，所操作的将是当前线程的值，不和其它线程共享

使用threading.local的SqlHelper

import pymysql
from dbutils.pooled_db import PooledDB
import threading

class SqlHelper(object):


    def __init__(self):
        self.threadLocalMap = {}
        self.pool = PooledDB(
            creator=pymysql,
            maxconnections=6,  # 最大连接数量 0和None表示不限制数量
            blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待，不等待抛出异常
            ping=0,  # 检查Mysql服务器是否可用，0表示不检查, 1 = default 请求都会检查 #2 创建cursor时检查 #4 执行sql时检查 #7 全部检查
            host='localhost',
            port=3306,
            user='root',
            passwd='root',
            database='web_test',
            charset='utf8'
        )
        self.local = threading.local()

    def open(self):
        connection = self.pool.connection()
        cursor = connection.cursor()
        return connection, cursor

    def close(self, connection, cursor):
        cursor.close()
        connection.close()

    def __enter__(self):
        con, cur = self.open()
        rv = getattr(self.local,'stack',None)
        if rv is None:
            self.local.stack = [(con,cur)]
        else:
            rv.append(cur)
        return cur

    def __exit__(self, exc_type, exc_val, exc_tb):
        rv = getattr(self.local, 'stack', None)
        if not rv:
            del rv
            return
        con,cur = rv.pop()
        self.close(con,cur)

db = SqlHelper()

测试

from threading import Thread

def func():
    from sqlHelper2 import db
    with db as db:
        db.execute("select * from user")
        result = db.fetchall()
        print("result1====>",result)

if __name__ == '__main__':
    for i in range(30):
        t1 = Thread(name=f"测试{i}", target=func)
        t1.start()

在这个 SqlHelper 类中，使用 stack（栈）是为了管理数据库连接和游标对象的生命周期，特别是在使用上下文管理器（__enter__ 和 __exit__ 方法）时。这样做有几个好处：

  1. 维护数据库连接和游标的配对：每次进入 __enter__ 方法时，都会打开一个新的数据库连接和游标，并将它们作为一个元组压入栈中。这样，每个连接都有一个对应的游标，并且它们可以被正确地配对。
  2. 支持嵌套的数据库操作：在某些情况下，你可能需要在已经打开的数据库连接中执行更深层次的操作，这可能涉及到再次进入 __enter__ 和 __exit__ 方法。使用栈可以确保即使在嵌套的情况下，每个 __enter__ 调用都有一个对应的 __exit__ 调用来关闭连接和游标。
  3. 确保资源正确释放：在 __exit__ 方法中，通过弹出栈顶的元素（即最后打开的连接和游标）来关闭它们。这样可以确保即使在发生异常时，资源也能被正确释放，避免资源泄露。
  4. 线程安全：由于每个线程都有自己的 threading.local() 对象，因此每个线程都有自己的栈，这确保了线程安全，不同线程的数据库操作不会互相干扰。

静态文件目录

app = Flask(__name__,static_url_path='/static',static_folder='static')

static_url_path是用于识别静态文件引入的url标识

static_folder则是静态文件目录位置

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>主页面</title>
</head>
<body>
 主页面
<img src="{{url_for('static', filename='test.png')}}" />
</body>
</html>

推荐使用url_for来引入静态文件，以避免静态文件名修改需要重新修改引入路径名

全局模板函数

使用app.template_global装饰器直接将函数注册为模板全局函数

template_global

定制在所有模板中可以使用的函数

@app.template_global
def bar():
    return "I am bar."

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>index</title>
</head>
<body>
    <hr />
    <p>调用bar()的结果为：{{ bar() }}</p>
 
</body>
</html>

加载配置文件

基于全局变量配置

app = Flask(__name__)

#加载配置文件
app.config.from_object('config.settings')

print(app.config['DB_HOST'])

到config包下的settings.py文件中加载配置

配置文件

DB_HOST = '192.168.16.88'

#导入本地配置文件（如测试环境，会使用以下操作），如果要上传生产环境不上传localsettings
try:
    from .localsettings import *
except ImportError:
    pass

通常在本地使用测试文件时会将配置放在localsettings里，去覆盖生产配置文件

路由系统

路由的两种写法

@app.route('/index')
def index():
    return render_template('index.html')

def index():
    return render_template('index.html')
app.add_url_rule('/index','index',index)

两种写法本质上没有区别

动态路由基本过程

  • 将url和函数打包成rule对象
  • 将rule对象加到Map对象中
  • app.url_map = Map对象

请求钩子

请求钩子允许你在处理请求的不同阶段插入代码，Flask 提供了几种钩子来处理请求生命周期的不同阶段：

  • **before_request**：在每个请求处理之前执行。
  • **after_request**：在每个请求处理之后执行。
  • **teardown_request**：请求处理结束后，无论是否发生异常都会执行。
  • **before_first_request**：仅在应用第一次处理请求之前执行。
app = Flask(__name__)

@app.before_request
def func():
    print(request.headers)

@app.after_request
def after_request(response):
    return response

两个常用装饰器执行顺序

@app.before_request
def func1():
    print("func1")

@app.before_request
def func2():
    print("func2")

@app.after_request
def after_request1(response):
    print("after1")
    return response

@app.after_request
def after_request2(response):
    print("after2")
    return response

一个请求进入，打印func1 func2 after2 after1

before_request按照谁在前谁先生效的原则

after_request则反之

  • 对于钩子函数，是可以在蓝图中定义的，但是作用域也只会在定义的蓝图内生效

Flask工作流程

当有用户请求到来时

  1. 创建ctx = RequestContext对象，其内部封装了Request对象，g对象
  2. 创建app_ctx = AppContext对象，内部封装了App对象，g对象
  3. ctx.push函数触发ctx和app_ctx通过自己的LocalStack对象将其放入Local中，Local的本质是一个以线程ID为key，以栈为value的字典
  4. 执行所有before_request函数
  5. 执行视图函数
  6. 执行所有after_request函数（session加密放到cookie中）
  7. 销毁ctx和app_ctx

g保存的是当前请求的全局变量，仅在当前这一个请求内，从一个函数到另一个函数共享数据，不同的请求会有不同的全局变量，通过不同的thread id区别

补充：在Flask中request，session，g都使用LocalProxy和偏函数的方式进行代理本质上都是调用ctx.操作对象的方式

转载请注明来源，欢迎对文章中的引用来源进行考证，欢迎指出任何有错误或不够清晰的表达。可以邮件至 1300452403@qq.com

赏

文章标题:Flask

字数:4.6k

本文作者:Os467

发布时间:2024-12-18, 11:39:13

最后更新:2024-12-18, 11:40:20

原始链接:https://os467.github.io/2024/12/18/flask/

版权声明: "署名-非商用-相同方式共享 4.0" 转载请保留原文链接及作者。

©2016-2020 os467

Built with Hexo and 3-hexo theme

×

喜欢就点赞,疼爱就打赏