Flask基础
2022/6/20 23:20:28
本文主要是介绍Flask基础，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
1.什么是wsgi
wsgi
wsgi是一种规范.全称Python Web Server Gateway Interface,指定了web服务器和Python web应用或web框架之间的标准接口.WSGI规定，Web程序必须有一个可调用对象，且该可调用对象接收两个参数，返回一个可迭代对象：
environ：字典，包含请求的所有信息
start_response：在可调用对象中调用的函数，用来发起响应，参数包括状态码，headers等
uwsgi
uwsgi是uWSGI服务器的独占协议.
uWSGI
uWSGI：是一个web服务器，实现了WSGI协议、uwsgi协议、http协议等.
符合wsgi的application
1 def application(environ,srart_response):
2 start_response('200 OK', [('Content-Type', 'text/plain')])
3 return ['This is a python application!']
1 class ApplicationClass(object):
2 def __init__(self, environ, start_response):
3 self.environ = environ
4 self.start_response = start_response
5
6 def __iter__(self):
7 self.start_response('200 OK', [('Content-type', 'text/plain')])
8 yield "Hello world!n"
2.什么是werkzug
werkzeug
werkzeug是一个WSGI工具包,提供了Request和Response.
flask中的werkzeug
Flask中的程序实例app就是一个可调用对象，我们创建app实例时所调用的Flask类实现了__call方法，__call方法调用了wsgi_app()方法，该方法完成了请求和响应的处理，WSGI服务器通过调用该方法传入请求数据，获取返回数据
1 def wsgi_app(self, environ, start_response): 2 ctx = self.request_context(environ) 3 error = None 4 try: 5 try: 6 ctx.push() 7 response = self.full_dispatch_request() 8 except Exception as e: 9 error = e 10 response = self.handle_exception(e) 11 except: # noqa: B001 12 error = sys.exc_info()[1] 13 raise 14 return response(environ, start_response) 15 finally: 16 if self.should_ignore_error(error): 17 error = None 18 ctx.auto_pop(error) 19 20 def __call__(self, environ, start_response): 21 return self.wsgi_app(environ, start_response)
本质
1 from werkzeug.serving import run_simple
2 from werkzeug.wrappers import BaseResponse
3
4 def func(environ,start_response):
5 print('请求来了')
6 response = BaseResponse('你好')
7 return response(environ,start_response)
8
9 if __name__ == '__main__':
10 run_simple('127.0.0.1',5000,func)
3.flask的快速使用
from flask import Flask, render_template, jsonify, request, redirect, url_for, session
# template_folder模板目录 static_folder 静态目录 static_url_path 改变url的path
app = Flask(__name__, template_folder='templates', static_folder='static', static_url_path='/static')
# app密钥
app.secret_key = 's56h5h45h54s5sfs545agd6'
# 登录认证
def auth(func):
@functools.wraps(func)
def inner(*args, **kwargs):
username = session.get('xxx')
if not username:
# redirect 重定向跳转 url_for 反向生成 等价于redirect('/login')
return redirect(url_for('login'))
return func(*args, **kwargs)
return inner
# 请求之前中间件
@app.before_request
def f_before():
pass
# 请求之后中间件
@app.after_request
def f_after(response):
return response
# 在对应用程序实例的第一个请求之前注册要运行的函数, 只会执行一次
@app.before_first_request
def bfr():
pass
# 主页 endpoint起别名
@app.route('/', methods=['GET'], endpoint='index')
@auth
def index():
DATA_DICT = {}
return render_template('index.html', data_dict=DATA_DICT)
# 登录 endpoint默认等于'/login'
@app.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'GET':
# return 'hello' # 返回HttpResponse响应
# return jsonify({'code':1000,'data':[1,2,3]}) # 返回json数据
return render_template('login.html')
user = request.form.get('user', '0')
if user == '1':
session['xxx'] = 'yyy'
return redirect('/index')
return render_template('login.html', error='error')
# 路径传参
@app.route('/delete/<nid>', methods=['GET'])
@auth
def delete(nid):
print(nid)
return redirect(url_for('idx'))
# 启动flask
if __name__ == '__main__':
app.run()
4.蓝图及其他
1 from flask import Blueprint, render_template, request, redirect, session, url_for
2
3 loginroute = Blueprint('loginroute', __name__)
4
5 @loginroute.route('/login', methods=['GET', 'POST'], endpoint='login')
6 def login():
7 return render_template('login.html', error=error)
8
9 # 主app中注册蓝图
10 def create_app():
11 app = Flask(__name__, template_folder='templates', static_folder='static', static_url_path='/static')
12 app.secret_key = 'df2hfdghjdf5gj'
13 app.register_blueprint(login.loginroute, url_prefix='/blog')
14 return app
15 class UserView(views.MethodView):
16 methods = ['GET', 'POST']
17 decorators = [test, ]
18
19 def get(self):
20 return 'get'
21
22 def post(self):
23 return 'POST'
24 app.add_url_rule('/user',view_func=UserView.as_view('user'))
25 # 如果不用方法视图实现需要在普通视图内部调用request.method判断是否为GET，POST进行判断
5.threading.local
1 # 自定义threading_local()
2 # storage = {
3 # 1001:{'x1':1},
4 # 1002:{'x1':2},
5 # 1003:{'x1':3},
6 # 1004:{'x1':4},
7 # }
8 class Local(object):
9 def __init__(self):
10 # storage = {}
11 object.__setattr__(self, 'storage', {})
12
13 def __setattr__(self, key, value):
14 ident = threading.get_ident()
15 if ident in self.storage:
16 self.storage[ident][key] = value
17 else:
18 self.storage[ident] = {key: value}
19
20 def __getattr__(self, item):
21 ident = threading.get_ident()
22 if ident not in self.storage:
23 return
24 return self.storage[ident].get(item)
25 # 自定义高级threading_local()
26 # storage = {
27 # 1001:{'x1':[1,2]},
28 # 1002:{'x1':[]},
29 # 1003:{'x1':[3,]},
30 # 1004:{'x1':[4,5,6]},
31 # }
32 class Local(object):
33 def __init__(self):
34 # storage = {}
35 object.__setattr__(self, 'storage', {})
36
37 def __setattr__(self, key, value):
38 ident = threading.get_ident()
39 if ident in self.storage:
40 self.storage[ident][key].append(value)
41 else:
42 self.storage[ident] = {key: [value,]}
43
44 def __getattr__(self, item):
45 ident = threading.get_ident()
46 if ident not in self.storage:
47 return
48 return self.storage[ident][item][-1]
6.LocalStack和Local对象实现栈
1 """
2 __storage__={
3 1111:{"stack":[xx,yy]}
4 }
5 """
6
7
8 class LocalStack(object):
9 def __iter__(self):
10 self._local = Local()
11
12 def push(self, obj):
13 rv = getattr(self._local, 'stack', None)
14 if rv is None:
15 self._local.stack = rv = []
16 rv.append(obj)
17 return rv
18
19 def pop(self):
20 stack = getattr(self._local, 'stack', None)
21 if stack is None:
22 return None
23 elif len(stack) == 1:
24 return stack[-1]
25 else:
26 return stack.pop()
27
28 def top(self):
29 try:
30 return self._local.stack[-1]
31 except (AttributeError,IndexError):
32 return None
7.Flask全过程
1 app = Flask(__name__,static_url_path='/xx')
2
3 # 在源码种生成如下数据
4 url_rule_class = Rule
5 url_map_class = Map
6 session interface = SecureCookieSessionInterface()
7 # Flask中的__init__
8 self.static_url_path = static_url_path
9 self.config = Config
10 self.view_functions = {
11 "index":index
12 }
13 self.before_request_funcs = {}
14 self.after_request_funcs = {}
15 self.before_first_request = []
16 self.url_map = self.url_map_class()
17 self.add_url_rule("static")
18 # 从字典中读取配置信息
19 app.config.from_object("xx.xx")
20 @app.before_request
21 def f_before():
22 pass
23 # 将f_before添加到self.before_request_funcs
24 self.before_request_funcs['None'].append(f_before)
25 @app.after_request
26 def f_after(response):
27 return response
28 # 将f_after添加到self.after_request_funcs
29 self.after_request_funcs['None'].append(f_after)
30 # self.after_request_funcs中的中间件逆序执行
31 @app.before_first_request
32 def bfr():
33 pass
34 # bfr.before_first_request_funcs
35 self.before_first_request_funcs['None'].append(bfr)
36 @app.route('/', methods=['GET'], endpoint='index')
37 @auth
38 def index():
39 DATA_DICT = {}
40 return render_template('index.html', data_dict=DATA_DICT)
41 #
42 self.add_url_rule
43 # 将url和对应的视图函数名绑定到rule,再将rule和endpoint放到Map,通过as.view方法实现view_func函数的传入
44 # Rule
45 endpoint和url
46 # Map
47 __init__
48 self.rules = [rule,rule]
49 # 启动flask
50 if __name__ == '__main__':
51 app.run()
52 # 请求到来之后,执行__call__方法
53 # run_simple(*,*,self,*)
54 # 执行第三个函数(self)
55 def wsgi_app(self, environ, start_response):
56 """
57 创建ctx,包含request=Request(environ) session=None
58 详见8.1
59 """
60 ctx = self.request_context(environ)
61 error = None
62 try:
63 try:
64 """
65 app_ctx = AppContext(app,g)
66 将app_ctx放入local中
67 将ctx放入local
68 session赋值
69 路由匹配
70 详见8.1
71 """
72 ctx.push()
73 """
74 详见8.4
75 """
76 response = self.full_dispatch_request()
77 except Exception as e:
78 error = e
79 response = self.handle_exception(e)
80 except: # noqa: B001
81 error = sys.exc_info()[1]
82 raise
83 return response(environ, start_response)
84 finally:
85 if self.should_ignore_error(error):
86 error = None
87 ctx.auto_pop(error)
88
89 def __call__(self, environ, start_response):
90 return self.wsgi_app(environ, start_response)
91
92 def full_dispatch_request(self) -> Response:
93 # 触发所有first_request函数
94 self.try_trigger_before_first_request_functions()
95 try:
96 # 信号 暂留
97 request_started.send(self)
98 # 执行所有before_request
99 rv = self.preprocess_request()
100 if rv is None:
101 # 执行视图函数
102 rv = self.dispatch_request()
103 except Exception as e:
104 rv = self.handle_user_exception(e)
105 return self.finalize_request(rv)
106
107 def try_trigger_before_first_request_functions(self) -> None:
108 if self._got_first_request:
109 return
110 with self._before_request_lock:
111 if self._got_first_request:
112 return
113 for func in self.before_first_request_funcs:
114 self.ensure_sync(func)()
115 self._got_first_request = True
116 def preprocess_request(self) -> t.Optional[ResponseReturnValue]:
117 names = (None, *reversed(request.blueprints))
118 for name in names:
119 if name in self.url_value_preprocessors:
120 for url_func in self.url_value_preprocessors[name]:
121 url_func(request.endpoint, request.view_args)
122 for name in names:
123 if name in self.before_request_funcs:
124 for before_func in self.before_request_funcs[name]:
125 rv = self.ensure_sync(before_func)()
126 if rv is not None:
127 return rv
128 return None
129 class RequestContext:
130 def __init__(
131 self,
132 app: "Flask",
133 environ: dict,
134 request: t.Optional["Request"] = None,
135 session: t.Optional["SessionMixin"] = None,
136 ) -> None:
137 self.app = app
138 if request is None:
139 request = app.request_class(environ)
140 self.request = request
141 self.url_adapter = None
142 try:
143 self.url_adapter = app.create_url_adapter(self.request)
144 except HTTPException as e:
145 self.request.routing_exception = e
146 self.flashes = None
147 self.session = session
148 def push(self) -> None:
149 """
150 调用了如下函数,返回了AppContext
151 详见8.2
152 def app_context(self) -> AppContext:
153 return AppContext(self)
154 """
155 app_ctx = self.app.app_context()
156 app_ctx.push()
157 """
158 _request_ctx_stack = LocalStack()
159 _app_ctx_stack = LocalStack()
160 详见8.3
161 """
162 _request_ctx_stack.push(self)
163
164 if self.session is None:
165 session_interface = self.app.session_interface
166 self.session = session_interface.open_session(self.app, self.request)
167
168 if self.session is None:
169 self.session = session_interface.make_null_session(self.app)
170
171 if self.url_adapter is not None:
172 self.match_request()
173 class AppContext:
174 def __init__(self, app: "Flask") -> None:
175 self.app = app
176 self.url_adapter = app.create_url_adapter(None)
177 self.g = app.app_ctx_globals_class()
178
179 # Like request context, app contexts can be pushed multiple times
180 # but there a basic "refcount" is enough to track them.
181 self._refcnt = 0
182 def push(self) -> None:
183 """Binds the app context to the current context."""
184 self._refcnt += 1
185 """
186 _request_ctx_stack = LocalStack()
187 _app_ctx_stack = LocalStack()
188 详见8.3
189 """
190 _app_ctx_stack.push(self)
191 appcontext_pushed.send(self.app)
192 class LocalStack(object):
193 def __iter__(self):
194 self._local = Local()
195 def push(self, obj):
196 rv = getattr(self._local, 'stack', None)
197 if rv is None:
198 self._local.stack = rv = []
199 rv.append(obj)
200 return rv
201 def pop(self):
202 stack = getattr(self._local, 'stack', None)
203 if stack is None:
204 return None
205 elif len(stack) == 1:
206 return stack[-1]
207 else:
208 return stack.pop()
209 def top(self):
210 try:
211 return self._local.stack[-1]
212 except (AttributeError,IndexError):
213 return None
214 """
215 _request_ctx_stack = LocalStack()
216 request context，程序上下文，存储的是请求级别的信息，比如当前访问的url
217 _app_ctx_stack = LocalStack()
218 app context，应用上下文，存储的是应用级别的信息，比如数据库连接信息。
219 """
220 _request_ctx_stack = {111:{stack:[ctx,]}}
221 _app_ctx_stack = {111:{stack:[app_ctx,]}}
8.用户请求全过程
1 """
2 创建ctx=RequestContext对象,其内部封装了Request和session
3 创建app_ctx=AppContext对象,其内部封装了App和g
4 ctx.push触发ctx和app_ctx分别通过自己的LocalStack对象将其放入Local中,以线程ID为key,以{"stack":[]}为value的字典
5 {
6 1111:{"stack":[ctx,]}
7 }
8 {
9 1111:{"stack":[app_ctx,]}
10 }
11
12 注意:以后要获取request/session/app/g时,去local中获取
13 执行所有before_request
14 执行视图函数
15 执行所有after_request(session放到cookie,保存在浏览器中)
16 销毁ctx和app_ctx
17 """
这篇关于Flask基础的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
- 2024-05-15鸿蒙生态设备数量超8亿台
- 2024-05-13TiDB + ES：转转业财系统亿级数据存储优化实践
- 2024-05-09“2024鸿蒙零基础快速实战-仿抖音App开发（ArkTS版）”实战课程已上线
- 2024-05-09聊聊如何通过arthas-tunnel-server来远程管理所有需要arthas监控的应用
- 2024-05-09log4j2这么配就对了
- 2024-05-09nginx修改Content-Type
- 2024-05-09Redis多数据源，看这篇就够了
- 2024-05-09Google Chrome驱动程序 124.0.6367.62（正式版本）去哪下载？
- 2024-05-09有没有大佬知道这种数据应该怎么抓取呀？
- 2024-05-09这种运行结果里的10.100000001，怎么能最快改成10.1？
