https://helloflask.com/book/1/
https://github.com/helloflask/flask-origin
https://github.com/helloflask/flask-extension-status
Flask
https://flask.palletsprojects.com/
Flask 是典型的微框架,作为 Web 框架来说,它仅保留了核心功能:请求响应处理和模板渲染。这两类功能分别由 Werkzeug(WSGI 工具库)完成和 Jinja(模板渲染库)完成,因为 Flask 包装了这两个依赖,我们暂时不用深入了解它们。Werkzeug是一个WSGI工具包,他可以作为一个Web框架的底层库,而并非是一个web服务器,也不是一个web框架,封装好了很多 Web 框架的东西,例如 Request,Response 等等
Flask 框架就是一 Werkzeug 为基础开发的
flask也可以支持协程,安装方式要进行改变
资源
https://blog.csdn.net/shifengboy/article/details/114274271
安装
pip install flask
部署
使用 nohup 命令可以后台部署python服务:
nohup python app.py &
执行 flask run 命令时,Flask 会使用内置的开发服务器来运行程序。这个服务器默认监听本地机的 5000 端口。
启动 Flask 项目
source ~/virtualenv/yuzaoyah/bin/activate;FLASK_APP=app.py flask run --port 8080 --host 0.0.0.0 --reload --with-threads >/tmp/project-2024-06-10.log 2>&1
source ~/virtualenv/yuzaoyah/bin/activate,激活虚拟环境FLASK_APP=app.py,设置临时环境变量,指定 APPflask run,运行 Flask 项目--port 8080,指定端口--host 0.0.0.0,指定服务要监听的IP,这样服务器就可以接受来自任何IP地址的请求--reload,使服务器在代码改变时自动重启,方便开发时实时更新--with-threads,允许服务器使用线程来处理请求,提高并发处理能力> /tmp/project-2024-06-10.log 2>&1,重定向操作,此处2>&1保持紧凑,中间不要有空格0,标准输入文件描述符,1,标准输出文件描述符,2,标准错误文件描述符
通常
cmd 123 > xxx.log,含义是将命令cmd 123的标准输出重定向到文件xxx.log,此处省略了文件描述符1,相当于cmd 123 1 > xxx.log(但该写法本身不正确)2>&1表示将标准错误重定向到标准输出,此处&1表示标准输出文件描述符,2>1表示将标准错误重定向到普通文件1验证
# xxx.log 中会有内容 123 echo 123 > xxx.log # xxx.log 中会有内容 cat: 123: No such file or directory cat 123 > xxx.log 2>&1 # 普通文件 1 中会有内容 cat: 123: No such file or directory cat 123 > xxx.log 2>1
架构
一切从客户端发起请求开始
- 所有Flask程序都必须创建一个程序实例。
- 当客户端想要获取资源时,一般会通过浏览器发起HTTP请求。
- 此时,Web服务器使用一种名为WEB服务器网关接口的WSGI(Web Server Gateway Interface)协议,把来自客户端的请求都交给Flask程序实例。
- Flask使用Werkzeug来做路由分发(URL请求和视图函数之间的对应关系)。根据每个URL请求,找到具体的视图函数。
- 在Flask程序中,路由一般是通过程序实例的装饰器实现。通过调用视图函数,获取到数据后,把数据传入HTML模板文件中,模板引擎负责渲染HTTP响应数据,然后由Flask返回响应数据给浏览器,最后浏览器显示返回的结果。
模板
Jiinja2
模板
{{ ... }}用来标记变量。{% ... %}用来标记语句,比如 if 语句,for 语句等。`` 用来写注释。
为了方便对变量进行处理,Jinja2 提供了一些过滤器,语法形式如下:
{{ 变量|过滤器 }}左侧是变量,右侧是过滤器名。比如,上面的模板里使用
length过滤器来获取movies的长度,类似 Python 里的len()函数。提示 访问 https://jinja.palletsprojects.com/en/3.0.x/templates/#builtin-filters 查看所有可用的过滤器。
模板上下文环境
提示 在 Python 脚本里,url_for() 函数需要从 flask 包中导入,而在模板中则可以直接使用,因为 Flask 把一些常用的函数和对象添加到了模板上下文(环境)里。
模板
- 模板上下文处理函数
- 模板继承
- 因为基模板会被所有其他页面模板继承,如果你在基模板中使用了某个变量,那么这个变量也需要使用模板上下文处理函数注入到模板里。
静态文件
view funciton
视图函数
request
flask在接受请求后将请求信息封装于request对象,由于上下文机制,request只能在视图函数中使用
from flask import request
对于flask中的request有下面三种情况:
单进程、单线程,只需要基于全局变量实现即可
单进程、多线程,此时需要threading.local()对象来实现,为每一个请求的request单独开辟一个空间,这样不会造成request对象的混乱
单进程、单线程、多协程,这种情况threading.local()对象是无法实现的,因为线程中协程的资源都是共享的
那么,如果在flask中支持协程,应该怎么实现呢如果要支持线程,可以自定义类似threading.local()对象,那么它是在threading.local()对象的基础上进一步强化,可以支持协程。
import threading
local_values = threading.local()
def func(num):
local_values.num_ = num
print(local_values.num_, threading.current_thread().name)
for i in range(10):
t = threading.Thread(target=func, args=(i,), name='线程%s' % i)
t.start()
"""
输出:
0 线程0
1 线程1
2 线程2
3 线程3
4 线程4
5 线程5
6 线程6
7 线程7
8 线程8
9 线程9
"""
from sys import getsizeof
print(getsizeof(''))
print({1: "cc", 0: 'sfs'}[True])
from flask import request
# HTTP 请求中参数有两个位置,1)在URL中,2)在body中
request.args # URL 参数
request.data # 请求体 字节
request.form # 请求体 Content-Type 为 application/x-www-form-urlencoded
request.files #
request.values # args + form
request.json # 请求体 Content-Type 为 application/json
# 路径参数
request.view_args
response
cookie
session
from flask import session
配置
flask框架可以从.py文件、配置文件、环境变量中获取配置,最终保存于app.config字典中。
环境变量的设置在不同环境中方式不一:
在Linux Shell中:
export FLASK_APP=hello.py
在 Windows CMD 中使用 set 命令:
set FLASK_APP=hello.py
在 Windows PowerShell 中则使用下面的命令:
$env:FLASK_APP = "hello.py"
FLASK_APP
python -m flask --app hello run
FLASK_APP=hello.py python -m flask run
export FLASK_APP=hello.py
python -m flask run
FLASK_ENV
FLASK_ENV 用来设置程序运行的环境,默认为production,可以设置为development开启调试模式。
调试模式
- 程序出错,浏览器页面会显示错误信息
- 代码出现变动后,程序会自动重载
Flask 可以使用dotenv管理环境变量
pip install python-dotenv
命令行接口
内置命令行接口
flask shell
https://flask.palletsprojects.com/en/3.0.x/cli/
自定义命令行接口
import click
from flask import Flask
app = Flask(__name__)
@app.cli.command("create-user")
@click.argument("name")
def create_user(name):
...
flask create-user admin
扩展
https://wizardforcel.gitbooks.io/flask-extension-docs/content/flask-pymongo.html
flask-sqlalchemy
https://docs.sqlalchemy.org/en/14/
https://flask-sqlalchemy.palletsprojects.com/en/2.x/
flask-sqlalchemy对SQLAlchemy进行封装以适配Flask,例如可以直接从app.config中获取配置
pip install flask-sqlalchemy
# 使用扩展后可以直接在flask框架中加载配置
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test3'
flask-mysqldb
pip install flask-mysqldb
flask-restful
pip install flask-restful
获取 Flask 中所有路由
FLASK_APP=app.py flask routes
app.url_map
线下脚本命名规范
项目中会有一些用于解决某些问题、需要单独手动执行的常用脚本,这些脚本通常会不定时的手动执行,也会有一些仅为解决某些偶然问题、只需要在当时手动执行一次或几次的临时脚本,这些脚本通常在当时执行完后便没有价值或者基本没有价值。这些常用或者临时的脚本被称为线下脚本,表示非由服务调用,而是手动指定,例如导出数据、刷脏数据、补数据、清缓存等脚本。
随着时间的推移,这些脚本会越来越多,类似export_order_data、flush_product等常用名称不得不附加其他字段保证文件名称唯一,例如export_order_data_6573、export_order_data_2024_06_10、export_order_data_05_10_error。
当然,定期删除临时脚本可以缓解该问题,不过在命名规范上作文章才是从源头解决问题。
- 线下脚本统一管理在
offline_script,常用脚本和临时脚本分别管理 - 常用脚本统一管理在
offline_script/recycle目录下,同功能脚本优先扩充已存在的同类功能的脚本,例如已存在export_order_id_and_custom_name的脚本,现在要导出order_id和custom_id两个字段,不妨扩充export_order_id_and_custom_name的功能,一是新增可导出字段custom_id,二是增加导出字段可选的功能,然后考虑给脚本改个更合适的名字,例如export_order。该方式的改动会影响到原本的业务,自然不如新写一个方便,但可以保证脚本的简洁,对于常用脚本,这个投入是值得的。 - 每个开发者在该目录下面创建已该开发者代号(姓名全拼等)为名的目录,如
offline_script/yuzaoyah,每个开发者在该目录下各自管理临时脚本,命名统一格式为ts_2024_06_10_export_order_data,ts表示temporary script,2024_06_10表示脚本编写日期,export_order_data表示脚本功能,临时脚本随用随写,不涉及脚本复用。 - 一些情况下,有可能开始判断只需要写一个脚本临时解决一下问题,一段时间后发现可能不定期就需要执行一次,这时不妨将该脚本升级为常用脚本。反过来,一开始以为要不定期执行而实际后续只需要执行一次的脚本,可以放到
offline_script/unrecycle。
offline_script
├─recycle
│ clear_cache.py
│ flush_calendar.py
├─unrecycle
│ flush_order_data.py
└─yuzaoyah
ts_2024_06_10_export_custom_s.py
集成
flask-restful
from flask import Flask
from flask_restful import Api, Resource, reqparse, abort
# from flask.ext import restful,旧版本flask拓展使用该方法
app = Flask(__name__)
api = Api(app)
# http://www.pythondoc.com/Flask-RESTful/quickstart.html
TODOS = {
'todo1': {'task': 'build an API'},
'todo2': {'task': '?????'},
'todo3': {'task': 'profit!'},
}
def abort_if_todo_doesnt_exist(todo_id):
if todo_id not in TODOS:
abort(404, message="Todo {} doesn't exist".format(todo_id))
parser = reqparse.RequestParser()
parser.add_argument('task', type=str)
class Todo(Resource):
def get(self, todo_id):
abort_if_todo_doesnt_exist(todo_id)
return TODOS[todo_id]
def delete(self, todo_id):
abort_if_todo_doesnt_exist(todo_id)
del TODOS[todo_id]
return '', 204
def put(self, todo_id):
args = parser.parse_args()
task = {'task': args['task']}
TODOS[todo_id] = task
return task, 201
# shows a list of all todos, and lets you POST to add new tasks
class TodoList(Resource):
def get(self):
return TODOS
def post(self):
args = parser.parse_args()
todo_id = int(max(TODOS.keys()).lstrip('todo')) + 1
todo_id = 'todo%i' % todo_id
TODOS[todo_id] = {'task': args['task']}
return TODOS[todo_id], 201
api.add_resource(TodoList, '/todos')
api.add_resource(Todo, '/todos/<todo_id>')
if __name__ == '__main__':
app.run(debug=True)
flask-restx
flask-debugtoolbar
https://github.com/pallets-eco/flask-debugtoolbar
local-proxy
import threading
class Singleton(type):
__instance = None
def __call__(cls, *args, **kwargs):
if not cls.__instance:
cls.__instance = type.__call__(cls, *args, **kwargs)
return cls.__instance
class LocalProxy(metaclass=Singleton):
__local = threading.local()
__lock = threading.Lock()
__attribute_set = {
# 接口请求者相关信息(保证如下信息均为同一请求者)
'user_id', # 用户id
'user_name' # 用户名称
}
def __setattr__(self, key, value):
if key not in self.__attribute_set:
raise AttributeError("attribute does not exist")
setattr(self.__local, key, value)
def __getattr__(self, item):
if item not in self.__attribute_set:
raise AttributeError("attribute does not exist")
return getattr(self.__local, item, None)
def __delattr__(self, item):
if item not in self.__attribute_set:
raise AttributeError("attribute does not exist")
delattr(self.__local, item)
def release(self):
with self.__lock:
for attr in self.__attribute_set:
if attr in self.__local.__dict__:
delattr(self, attr)
import flask
from flask import g
class Singleton(type):
__instance = None
def __call__(cls, *args, **kwargs):
if not cls.__instance:
cls.__instance = type.__call__(cls, *args, **kwargs)
return cls.__instance
class GProxy(metaclass=Singleton):
"""
代理 flask g,避免直接使用 g 造成线程全局变量的管理混乱
在 __slots__ 中预先定义需要的线程全局变量
"""
__slots__ = (
# 接口请求者相关信息(保证如下信息均为同一请求者)
'user_id', # 用户id
'user_name' # 用户名称
)
def __setattr__(self, key, value):
if key not in self.__slots__:
raise AttributeError("attribute does not exist")
setattr(g, key, value)
def __getattr__(self, item):
if item not in self.__slots__:
raise AttributeError("attribute does not exist")
return getattr(g, item, None)
with flask.Flask("123").app_context():
print(GProxy().password)
Flask 上下文
from flask import request, has_request_context
if has_request_contest():
request... # c