Flask
Posted xu1686318405
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Flask相关的知识，希望对你有一定的参考价值。
一、Flask
安装：pip3 install flask
flask：短小精悍，可扩展的一个web框架（上下文管理机制）
依赖：wsgi:werkzurg
学习werkzurg:
示例一： from werkzeug.wrappers import Request, Response from werkzeug.serving import run_simple def run(environ,start_response): return [b"asdfasdf"] if __name__ == ‘__main__‘: run_simple(‘localhost‘, 4000, run) 示例二： from werkzeug.wrappers import Request, Response @Request.application def hello(request): return Response(‘Hello World!‘) if __name__ == ‘__main__‘: from werkzeug.serving import run_simple run_simple(‘localhost‘, 4000, hello)
对django和flask的认识？
django：是一个大二全的一个框架，里面有很多的组件供开发者使用，让开发者可以很快的快速开发。
flask：是一个很小的web框架，可以写一个小的项目。没有组件供开发者使用，它可以通过第三方组件。
django和flask最大的不同点：django的request是通过传参的方式获得，flask是通过导入request获得。
flask知识点：
给你一个路径 “settings.Foo”，可以找到类并获取去其中的大写的静态字段。
settings.py
class Foo:
DEBUG = True
TEST = True
xx.py
import importlib
path = "settings.Foo"
p,c = path.rsplit(‘.‘,maxsplit=1)
m = importlib.import_module(p)
cls = getattr(m,c)
# 如果找到这个类？
for key in dir(cls):
if key.isupper():
print(key,getattr(cls,key))
1.配置文件：
所有配置都在app.config中
app.config["xx"] = 123
app.config.from_object("类的路径")
应用：importlib、getattr
django中间件
rest framework全局配置
app.config.from_object("settings.DevelopmentConfig") class Config(object): DEBUG = False TESTING = False DATABASE_URI = ‘sqlite://:memory:‘ class ProductionConfig(Config): DATABASE_URI = ‘mysql://[email protected]/foo‘ class DevelopmentConfig(Config): DEBUG = True class TestingConfig(Config): TESTING = True
2.路由系统
endpoint,反向生成URL，默认函数名 - url_for(‘endpoint‘) / url_for("index",nid=777) - 动态路由： @app.route(‘/index/<int:nid>‘,methods=[‘GET‘,‘POST‘]) def index(nid): print(nid) return "Index"
3.FBV
4.请求相关
# 请求相关信息 # request.method # request.args # request.form # 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/‘ + secure_filename(f.filename)) - 请求 request.form request.args request.method - 响应 render redirect - session session[‘xx‘] = 123 session.get(‘xx‘)
5.响应
响应体： return “asdf” return jsonify({‘k1‘:‘v1‘}) return render_template(‘xxx.html‘) return redirect() 定制响应头： obj = make_response("asdf") obj.headers[‘xxxxxxx‘] = ‘123‘ obj.set_cookie(‘key‘, ‘value‘) return obj
示例程序：学生管理
版本一： @app.route(‘/index‘) def index(): if not session.get(‘user‘): return redirect(url_for(‘login‘)) returnrender_template(‘index.html‘,stu_dic=STUDENT_DICT) 版本二： import functools def auth(func): @functools.wraps(func) def inner(*args,**kwargs): if not session.get(‘user‘): return redirect(url_for(‘login‘)) ret = func(*args,**kwargs) return ret return inner @app.route(‘/index‘) @auth def index(): return render_template(‘index.html‘,stu_dic=STUDENT_DICT) 应用场景：比较少的函数中需要额外添加功能。 版本三：before_request @app.before_request def xxxxxx(): if request.path == ‘/login‘: return None if session.get(‘user‘): return None return redirect(‘/login‘)
6.模板渲染
基本数据类型：可以执行python语法，如：dict.get() list[‘xxx‘]
传入函数
--django,自动执行
--flask,不自动执行
（1）全局定义函数
@app.template_global() def sb(a1, a2): # {{sb(1,9)}} return a1 + a2 @app.template_filter() def db(a1, a2, a3): # {{ 1|db(2,3) }} return a1 + a2 + a3
（2）模板继承
layout.html <!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <title>Title</title> <meta name="viewport" content="width=device-width, initial-scale=1"> </head> <body> <h1>模板</h1> {% block content %}{% endblock %} </body> </html> tpl.html {% extends "layout.html"%} {% block content %} {{users.0}} {% endblock %} - include {% include "form.html" %} form.html <form> asdfasdf asdfasdf </form>
- 宏 {% macro ccccc(name, type=‘text‘, value=‘‘) %} <h1>宏</h1> <input type="{{ type }}" name="{{ name }}" value="{{ value }}"> <input type="submit" value="提交"> {% endmacro %} {{ ccccc(‘n1‘) }} {{ ccccc(‘n2‘) }}
（3）安全
前端：{{u|safe}}
后端：MarkUp(‘asdef‘)
7.session
加密后放置在用户浏览器的cookie中。
流程：
请求到来
视图函数
请求结束
配置文件
当请求刚到来：flask读取cookie中session对应的值：adw,将该值解密并反序列化成字典，放入内存以便视图函数使用。
视图函数：
@app.route(‘/ses‘) def ses(): session[‘k1‘] = 123 session[‘k2‘] = 456 del session[‘k1‘] return "Session" session[‘xxx‘] = 123 session[‘xxx‘]
当请求结束时，flask会读取内存中字典的值，进行序列化+加密，写入到用户cookie中
8.闪现，在session中存储一个数据，读取时通过pop将数据移除
from flask import Flask,flash,get_flashed_messages @app.route(‘/page1‘) def page1(): flash(‘临时数据存储‘,‘error‘) flash(‘sdfsdf234234‘,‘error‘) flash(‘adasdfasdf‘,‘info‘) return "Session" @app.route(‘/page2‘) def page2(): print(get_flashed_messages(category_filter=[‘error‘])) return "Session"
9.中间件
call方法什么时候出发？
用户发起请求是才执行
任务：在执行call方法之前，做一个操作，call方法执行之后做一个操作
class Middleware(object): def __init__(self,old): self.old = old def __call__(self, *args, **kwargs): ret = self.old(*args, **kwargs) return ret if __name__ == ‘__main__‘: app.wsgi_app = Middleware(app.wsgi_app) app.run()
10.特殊装饰器
1. before_request 2. after_request 示例： from flask import Flask app = Flask(__name__) @app.before_request def x1(): print(‘before:x1‘) return ‘滚‘ @app.before_request def xx1(): print(‘before:xx1‘) @app.after_request def x2(response): print(‘after:x2‘) return response @app.after_request def xx2(response): print(‘after:xx2‘) return response @app.route(‘/index‘) def index(): print(‘index‘) return "Index" @app.route(‘/order‘) def order(): print(‘order‘) return "order" if __name__ == ‘__main__‘: app.run() 3. before_first_request from flask import Flask app = Flask(__name__) @app.before_first_request def x1(): print(‘123123‘) @app.route(‘/index‘) def index(): print(‘index‘) return "Index" @app.route(‘/order‘) def order(): print(‘order‘) return "order" if __name__ == ‘__main__‘: app.run() 4. template_global 5. template_filter 6. errorhandler @app.errorhandler(404) def not_found(arg): print(arg) return "没找到"
Flask源码入口：
from werkzeug.wrappers import Response from werkzeug.serving import run_simple class Flask(object): def __call__(self,environ, start_response): response = Response(‘hello‘) return response(environ, start_response) def run(self): run_simple(‘127.0.0.1‘, 8000, self) app = Flask() if __name__ == ‘__main__‘: app.run()
二、Flask进阶版
1.路由+视图
a、路由设置的两种方式
@app.route(‘/xxx‘) def index(): return "index" def index(): return "index" app.add_url_rule("/xxx",None,index)
注意事项：1.不用让endpoint重名 2.如果重名函数也要一定要相同
b、参数
rule, URL规则 view_func, 视图函数名称 endpoint=None, 名称，用于反向生成URL，即： url_for(‘名称‘) methods=None, 允许的请求方式，如：["GET","POST"] strict_slashes=None, 对URL最后的 / 符号是否严格要求， redirect_to=None, 重定向到指定地址 defaults=None, 默认值,当URL中无参数，函数需要参数时，使用defaults={‘k‘:‘v‘}为函数提供参数 subdomain=None, 子域名访问
c、CBV
import functools from flask import Flask,views app = Flask(__name__) def wrapper(func): @functools.wraps(func) def inner(*args,**kwargs): return func(*args,**kwargs) return inner class UserView(views.MethodView): methods = [‘GET‘] decorators = [wrapper,] def get(self,*args,**kwargs): return ‘GET‘ def post(self,*args,**kwargs): return ‘POST‘ app.add_url_rule(‘/user‘,None,UserView.as_view(‘uuuu‘)) if __name__ == ‘__main__‘: app.run()
d、自定义正则
from flask import Flask,url_for app = Flask(__name__) # 步骤一：定制类 from werkzeug.routing import BaseConverter 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 # 步骤二：添加到转换器 app.url_map.converters[‘reg‘] = RegexConverter """ 1. 用户发送请求 2. flask内部进行正则匹配 3. 调用to_python(正则匹配的结果)方法 4. to_python方法的返回值会交给视图函数的参数 """ # 步骤三：使用自定义正则 @app.route(‘/index/<reg("d+"):nid>‘) def index(nid): print(nid,type(nid)) print(url_for(‘index‘,nid=987)) return "index" if __name__ == ‘__main__‘: app.run()
2.session实现原理（源码）
3.蓝图
目标：给开发者提供目录结构
其他：
自定义模板、静态文件
某一类url添加前缀
给一类url添加before_request
4.threading.local（和flask无任何关系）
作用：为每个线程创建一个独立的空间，使得线程对自己的空间中的数据进行操作（数据隔离）
import threading from threading import local import time obj = local() def task(i): obj.xxxxx = i time.sleep(2) print(obj.xxxxx,i) for i in range(10): t = threading.Thread(target=task,args=(i,)) t.start()
问题：如何获取一个线程的唯一标记？threading.get_ident()
根据字典自定义一个类似于threading.local功能？
import time import threading DIC = {} def task(i): ident = threading.get_ident() if ident in DIC: DIC[ident][‘xxxxx‘] = i else: IC[ident] = {‘xxxxx‘:i } time.sleep(2) print(DIC[ident][‘xxxxx‘],i) for i in range(10): t = threading.Thread(target=task,args=(i,)) t.start()
根据字典自定义一个为每个协程开辟空间进行存储数据
import time import threading import greenlet DIC = {} def task(i): # ident = threading.get_ident() ident = greenlet.getcurrent() if ident in DIC: DIC[ident][‘xxxxx‘] = i else: DIC[ident] = {‘xxxxx‘:i } time.sleep(2) print(DIC[ident][‘xxxxx‘],i) for i in range(10): t = threading.Thread(target=task,args=(i,)) t.start()
通过getattr/setattr 构造出来 threading.local的加强版（协程）
import time import threading try: import greenlet get_ident = greenlet.getcurrent except Exception as e: get_ident = threading.get_ident class Local(object): DIC = {} def __getattr__(self, item): ident = get_ident() if ident in self.DIC: return self.DIC[ident].get(item) return None def __setattr__(self, key, value): ident = get_ident() if ident in self.DIC: self.DIC[ident][key] = value else: self.DIC[ident] = {key:value} obj = Local() def task(i): obj.xxxxx = i time.sleep(2) print(obj.xxxxx,i) for i in range(10): t = threading.Thread(target=task,args=(i,)) t.start()
5.上下文管理（第一次）
请求到来时候：
ctx = RequestContext(self, environ) # self是app对象，environ请求相关的原始数据
ctx.request = Request(environ)
ctx.session = None
将包含了request/session的ctx对象放到“空调”
{
1232：{ctx:ctx对象}
1231：{ctx:ctx对象}
1211：{ctx:ctx对象}
1111：{ctx:ctx对象}
1261：{ctx:ctx对象}
}
视图函数：
from flask import request,session
request.method
请求结束：
根据当前线程的唯一标记，将“空调”上的数据移除。
以上是关于Flask的主要内容，如果未能解决你的问题，请参考以下文章