时间戳

首页 > 技术文章 > Python Flask框架

Python Flask框架

hp-mzx 2021-03-09 16:08 原文

Flask框架

Flask与Django的区别

  • Django是一个大而全的框架，Flask是一个轻量级的框架。
  • Django内部提供给非常多的组件：orm、session、cookie、admin、form、modelform、路由、视图、模板、中间件、分页、auth、contenttype、缓存、信号、多数据库连接
  • Flask框架本身没有太多功能：路由、视图、模板（jinja2）、session、中间件，第三方组件常齐全。
  • Django的请求处理是逐步封装传递；Flask的请求是利用上下文管理来实现的。

 

Flask快速使用

安装Flask

pip install flask

依赖wsgi werkzeug

from werkzeug.serving import run_simple

def fun(environ, start_response):
    print("请求")

if __name__ == '__main__':
    run_simple('127.0.0.1',5000,fun)

from werkzeug.serving import run_simple

class Flask(object):
    
    def __call__(self,environ, start_response):
        return "xx"

app = Flask()

if __name__ == '__main__':
    run_simple('127.0.0.1',5000,app)

from werkzeug.serving import run_simple

class Flask(object):
    
    def __call__(self,environ, start_response):
        return "xx"

    def run(self):
    	run_simple('127.0.0.1',5000,self)
        
app = Flask()

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

 

快速使用Flask

from flask import Flask

app = Flask(__name__)


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


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

总结：

  • Flask框架是基于werkzeug的wsgi实现，flask没有自己的wsgi
  • 用户请求一旦到来，就会执行app.__call__方法
  • 写flask标准流程

 

用户登录实例

from flask import Flask,render_template,jsonify,request,redirect

app = Flask(__name__)

@app.route('/login',methods=['GET','POST'])
def login():
    if request.method == "GET":
        return render_template('login.html')
        # return jsonify({'code':100,'data':[1,2,3]})
    user = request.form.get('user')
    pwd = request.form.get('pwd')
    if user == 'hp' and pwd == '123':
        return redirect('/index')
    error = "用户名或密码失败"
    return render_template('login.html',error=error)

@app.route('/index')
def index():
    return '首页'


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

 

用户管理实例

from flask import Flask, render_template, jsonify, request, redirect, url_for

app = Flask(__name__)

DATA_DICT = {
    1: {'name': 'hp', 'age': 18},
    2: {'name': 'mzx', 'age': 19},
}	

@app.route('/index', endpoint='inde')
def index():
    data_dict = DATA_DICT
    return render_template('index.html', data_dict=data_dict)


@app.route('/edit', methods=['GET', 'POST'])
def edit():
    nid = request.args.get('nid')
    nid = int(nid)

    if request.method == 'GET':
        info = DATA_DICT[nid]
        return render_template('edit.html', info=info)
    
    user = request.form.get('user')
    age = request.form.get('age')
    DATA_DICT[nid]['name'] = user
    DATA_DICT[nid]['age'] = age
    return redirect(url_for('inde'))



@app.route('/del/<int:nid>')
def delete(nid):
    del DATA_DICT[nid]
    # return redirect('/index')
    return redirect(url_for("inde"))


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

 

总结

1.flask路由
@app.route('/index', methods=['GET', 'POST'])
def index():
    pass
2.路由参数
@app.route('/index', methods=['GET', 'POST'], endpoint='inde')

'/index'：地址
methods：支持发方法，不写默认为GET
endpoint：别名（不能重名）
注意：endpoint不写，函数名就是别名
3.动态路由
@app.route('/index/<nid>')
def index(nid):
    pass
@app.route('/index/<int:nid>')
def index(nid):
    pass
4.获取提交的数据
from flask import request

@app.route('/index')
def index(nid):
    request.args.get()	#GET形式提交的参数
    request.form.get()	#POST形式提交的参数
5.返回数据
from flask import request,render_template,jsonify,redirect

@app.route('/index')
def index(nid):
    return render_template('模板文件')	# 返回模板文件
	return jsonify()	# 返回json格式
	return redirect('/index')	# 返回重定向
	return ""	# 返回字符串
6.模板处理
{{ xx }}

{% for item in list %}
	{{ item }}
{% endfor %}

 

保存用户会话信息

 

蓝图（blue print）

构建业务功能可拆分的目录结构。

 

 

数据库连接池

安装

pip install dbutils
pip install pymysql

使用

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,    # 使用链接数据库模块
    maxconnections=6,   # 连接池允许的最大连接数，0和None表示不限制链接数
    mincached=2,        # 初始化时，链接池中知识创建的链接，0表示不创建
    blocking=True,      # 连接池中如果没有可用链接，是否阻塞等待。True，等待；False，不等待并报错
    ping=0,             # ping mysql服务端，检测是否服务可用

    host='127.0.0.1',
    port=3306,
    user='root',
    password='',
    database='cov',
    charset='utf8'
)

# 去连接池获取一个链接
conn = POOL.connection()

cursor = conn.cursor()
cursor.execute('select * from history')
result = cursor.fetchall()
cursor.close()

# 将连接 放回到链接池
conn.close()

开启多线程

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,    # 使用链接数据库模块
    maxconnections=6,   # 连接池允许的最大连接数，0和None表示不限制链接数
    mincached=2,        # 初始化时，链接池中知识创建的链接，0表示不创建
    blocking=True,      # 连接池中如果没有可用链接，是否阻塞等待。True，等待；False，不等待并报错
    ping=0,             # ping mysql服务端，检测是否服务可用

    host='127.0.0.1',
    port=3306,
    user='root',
    password='',
    database='cov',
    charset='utf8'
)

def task(num):
    # 去连接池获取一个链接
    conn = POOL.connection()

    cursor = conn.cursor()
    cursor.execute('select sleep(3)')	# 睡3秒
    result = cursor.fetchall()
    cursor.close()

    print(num,'------->',result)
    # 将连接 放回到链接池
    conn.close()


from threading import Thread
for i in range(10):
    t = Thread(target=task,args=(i,))
    t.start()

 

基于函数实现sqlhelper

import pymysql
from DBUtils.PooledDB import PooledDB

POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库模块
    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制链接数
    mincached=2,  # 初始化时，链接池中知识创建的链接，0表示不创建
    blocking=True,  # 连接池中如果没有可用链接，是否阻塞等待。True，等待；False，不等待并报错
    ping=0,  # ping mysql服务端，检测是否服务可用

    host='127.0.0.1',
    port=3306,
    user='root',
    password='',
    database='cov',
    charset='utf8'
)


def fetchall(sql, *args):
    """获取所以数据"""

    # 去连接池获取一个链接
    conn = POOL.connection()

    cursor = conn.cursor()
    cursor.execute(sql, args)
    result = cursor.fetchall()
    cursor.close()
    # 将连接 放回到链接池
    conn.close()

    return result


def fetchone(sql, *args):
    """获取一条数据"""

    # 去连接池获取一个链接
    conn = POOL.connection()

    cursor = conn.cursor()
    cursor.execute(sql, args)
    result = cursor.fetchone()
    cursor.close()
    # 将连接 放回到链接池
    conn.close()

    return result

 

基于类实现sqlhelper

import pymysql
from DBUtils.PooledDB import PooledDB


class SqlHelper(object):
    def __init__(self):
        self.pool = PooledDB(
            creator=pymysql,  # 使用链接数据库模块
            maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制链接数
            mincached=2,  # 初始化时，链接池中知识创建的链接，0表示不创建
            blocking=True,  # 连接池中如果没有可用链接，是否阻塞等待。True，等待；False，不等待并报错
            ping=0,  # ping mysql服务端，检测是否服务可用

            host='127.0.0.1',
            port=3306,
            user='root',
            password='',
            database='cov',
            charset='utf8'
        )

    def open(self):
        # 去连接池获取一个链接
        conn = self.pool.connection()
        cursor = conn.cursor()
        return conn, cursor

    def close(self, cursor, conn):
        cursor.close()
        # 将连接 放回到链接池
        conn.close()

    def fetchall(self, sql, *args):
        """获取所以数据"""
        # 去连接池获取一个链接
        conn, cursor = self.open()
        cursor.execute(sql, args)
        result = cursor.fetchall()
        self.close(cursor, conn)
        return result

    def fetchone(self, sql, *args):
        """获取一条数据"""
        # 去连接池获取一个链接
        conn, cursor = self.open()
        cursor.execute(sql, args)
        result = cursor.fetchone()
        self.close(cursor, conn)

        return result


db = SqlHelper()

 

补充：上下文管理机制

class Foo(object):
    def __enter__(self):
        return 123
    def __exit__(self, exc_type, exc_val, exc_tb):
        pass
obj = Foo()
with obj as f:
    print(f)
    
with机制运行开始会触动对象的__enter__属性，运行结束后会触发__exit__属性
所以可以使用该机制触发数据库连接，与关闭

 

静态文件处理

from flask import Flask
app = Flask(__name__,static_folder='static',static_url_path='xx')	
# static_folder设定静态文件目录，去static目录找文件
# static_url_path引入静态文件的值,不写默认为static

静态文件写法

<img src='/xx/1.jpg'>	<!-- 设定static_url_path -->
<img src='/static/1.jpg'>
<img src='{{ url_for('static',filname='1.jpg') }}'>

 

配置文件

from flask import Flask, render_template

app = Flask(__name__)

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

 

定义全局模板方法

from flask import Flask, render_template

app = Flask(__name__)

@app.template_global()	# {{ func("xx") }}
def func(arg):
    return 'name' + arg

@app.template_filter()	# {{ 'name'|func2('xx') }}
def func2(arg,args):
    return 'name' + arg + args

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

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

 

推荐阅读