Flask快速入门（路由、CBV、请求和响应、session）

Flask快速入门（路由、CBV、请求和响应、session）

安装

pip install flask

创建页面

from flask import Flask# 用当前脚本名称实例化Flask对象，方便flask从该脚本文件中获取需要的内容
app = Flask(__name__)# 配置路由和视图函数的对应关系(基于装饰器)
@app.route("/")
def index():return "Hello World!"# 启动一个本地开发服务器，激活该网页
app.run()
# 或者：
# if __name__ == '__main__':
#     app.run()

Debug模式

终端执行

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

debug模式下：

• 自动重载 - 根据代码内容变化自动重启项目
• 终端展示错误提示
• 日志记录

快速使用

用route接受请求方式

@app.route("/", methods=['GET', 'POST'])
def index():return '你好'

直接调用请求方法

@app.get("/")
def index():return '你好'

Werkzeug介绍

Werkzeug是一个WSGI工具包，他可以作为一个Web框架的底层库，它并没有和flask有直接联系，但是flask可以借助它执行各种Web操作，例如Request，Response

from werkzeug.wrappers import Request, Response@app.route("/")
def index():return Response('你好')

watchdog介绍

快速体验

当前目录下修改文件会被监控，并打印日志

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandlerif __name__ == "__main__":# 配置日志logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(message)s',datefmt='%Y-%m-%d %H:%M:%S')path = sys.argv[1] if len(sys.argv) > 1 else '.'# 初始化监控类(文件的创建 删除...)event_handler = LoggingEventHandler()# 初始化监控类observer = Observer()# 配置 observer 以使用 event_handler 来处理 path 路径下的事件，并递归地监控该路径下的所有子目录（由于 recursive=True）observer.schedule(event_handler, path, recursive=True)# 启动监控observer.start()try:while True:time.sleep(1)except KeyboardInterrupt:observer.stop()observer.join()

路由系统

源码分析

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

直接Ctrl左键进入route

@setupmethod
def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:# 将视图函数作为参数执行了add_url_ruledef decorator(f: T_route) -> T_route:endpoint = options.pop("endpoint", None)self.add_url_rule(rule, endpoint, f, **options)return freturn decorator

该注释的意思是，当有给定的url规则来注册该视图函数时，将会触发该装饰器，且装饰器会执行add_url_rule方法，因此我们只需要搞明白add_url_rule做了什么

@setupmethod
def add_url_rule(self,rule: str,endpoint: str | None = None,view_func: ft.RouteCallable | None = None,provide_automatic_options: bool | None = None,**options: t.Any,
) -> None:

• rule：定义路由的字符串，例如'/'、'/login'
• endpoint：当URL中无参数，函数需要参数时，使用defaults = {‘k’: ‘v’}-为函数提供参数，django中也有，叫kwargs，默认为None
• view_func：视图函数的名称，也就是源码中的f
• provide_automatic_options: 一个布尔值，用于控制是否应该自动为路由添加一个 OPTIONS 方法的处理器。如果为 None，则使用应用配置的 ADD_AUTOMATIC_OPTIONS 值
• options: 一个可变参数，用于传递额外的选项给路由。这些选项可以包括用于路由的各种配置，如 HTTP 方法的集合（methods）

手动配置路由

知晓了路由原理后我们就可以不用装饰器自定义路由

def login():return '我是login'app.add_url_rule('/login', endpoint=None, view_func=login, methods=['GET', 'POST'])

动态路由-转换器

大致与Django同理

# 转换器
app.add_url_rule('/index/<int:pk>')

pk值是一个int类型的任意参数，除此之外还有其他类型参数：

DEFAULT_CONVERTERS = {'default':          UnicodeConverter,'string':           UnicodeConverter,'any':              AnyConverter,'path':             PathConverter,'int':              IntegerConverter,'float':            FloatConverter,'uuid':             UUIDConverter,
}

Flask的CBV

CBV（Class-Based Views）指的是基于类的视图编写方式

快速使用

from flask import Flask, url_for
from flask.views import MethodViewapp = Flask(__name__)
app.debug = Trueclass IndexView(MethodView):def get(self):return 'get请求'def post(self):return 'post请求'# name可以理解为别名
app.add_url_rule('/index', view_func=IndexView.as_view(name='index'))if __name__ == '__main__':app.run()# CBV也可以使用路由装饰器：
# @route('/index')  
# class IndexView(MethodView):  

即传了endpoint，又传了name，以谁为准？

• 如果传了endpoint，以endpoint 为准
  如果没传，以函数名为准，函数名是view，但是被name改了，所以以name为准

cbv源码分析

class View:methods: t.ClassVar[t.Collection[str] | None] = Noneprovide_automatic_options: t.ClassVar[bool | None] = Nonedecorators: t.ClassVar[list[t.Callable[[F], F]]] = []init_every_request: t.ClassVar[bool] = Truedef dispatch_request(self) -> ft.ResponseReturnValue:raise NotImplementedError()@classmethoddef as_view(cls, name: str, *class_args: t.Any, **class_kwargs: t.Any) -> ft.RouteCallable:if cls.init_every_request:def view(**kwargs: t.Any) -> ft.ResponseReturnValue:self = view.view_class(  # type: ignore[attr-defined]*class_args, **class_kwargs)# 这里其实就是return self.dispatch_request(**kwargs)return current_app.ensure_sync(self.dispatch_request)(**kwargs)  # type: ignore[no-any-return]else:self = cls(*class_args, **class_kwargs)def view(**kwargs: t.Any) -> ft.ResponseReturnValue:return current_app.ensure_sync(self.dispatch_request)(**kwargs)  # type: ignore[no-any-return]if cls.decorators:# 将view的名字改为传入的name 否则一直叫viewview.__name__ = nameview.__module__ = cls.__module__for decorator in cls.decorators:view = decorator(view)view.view_class = cls  # type: ignoreview.__name__ = nameview.__doc__ = cls.__doc__view.__module__ = cls.__module__view.methods = cls.methods  # type: ignoreview.provide_automatic_options = cls.provide_automatic_options  # type: ignorereturn view

• 每当有请求过来时执行view()，执行view的本质就是执行dispatch_request()，self就是我们定义的视图类对象

请求和响应

请求对象(request)

from flask import Flask, request

响应对象

make_response生成的响应对象可以存放字符串、模板、重定向、json

from flask import Flask, render_template, make_responseclass ContactView(MethodView):def get(self):res = make_response('get请求')res.set_cookie('name', '<NAME>', path='/contact')return res

• 可以对响应对象添加cookie，path的意思是只有path指定的路径会存储cookie，浏览其他页面不会携带cookie

Session

基本使用

# 配置secret_key 自定义的字符串
app.secret_key = 'abcdefg'

• 增：session['name'] = '张三'
• 删：session.pop('name')
• 清：session.clear()
• 取：name = session.get('name')

原理解析

• 存储或修改session时
  • 对session进行加密(三段式)
  • 存储到cookie(obj.set_cookie)
• 校验session时
  • 根据session取出cookie
  • 将第二段反解放入session中

源码：

# open_session
def open_session(self, app: Flask, request: Request) -> SecureCookieSession | None:s = self.get_signing_serializer(app)if s is None:return Noneval = request.cookies.get(self.get_cookie_name(app))if not val:return self.session_class()max_age = int(app.permanent_session_lifetime.total_seconds())try:data = s.loads(val, max_age=max_age)return self.session_class(data)except BadSignature:return self.session_class()# save_session
def save_session(self, app: Flask, session: SessionMixin, response: Response) -> None:name = self.get_cookie_name(app)domain = self.get_cookie_domain(app)path = self.get_cookie_path(app)secure = self.get_cookie_secure(app)samesite = self.get_cookie_samesite(app)httponly = self.get_cookie_httponly(app)# Add a "Vary: Cookie" header if the session was accessed at all.if session.accessed:response.vary.add("Cookie")# If the session is modified to be empty, remove the cookie.# If the session is empty, return without setting the cookie.if not session:if session.modified:response.delete_cookie(name,domain=domain,path=path,secure=secure,samesite=samesite,httponly=httponly,)response.vary.add("Cookie")returnif not self.should_set_cookie(app, session):returnexpires = self.get_expiration_time(app, session)val = self.get_signing_serializer(app).dumps(dict(session))  # type: ignoreresponse.set_cookie(name,val,  # type: ignoreexpires=expires,httponly=httponly,domain=domain,path=path,secure=secure,samesite=samesite,)response.vary.add("Cookie")