Flask

予早 2024-06-10 21:56:35
Categories: Tags:

https://read.helloflask.com/

https://helloflask.com/book/1/

https://github.com/helloflask/flask-origin

https://github.com/helloflask/flask-extension-status

Flask

https://flask.palletsprojects.com/

https://helloflask.com/book/

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

https://read.helloflask.com/

安装

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
  1. source ~/virtualenv/yuzaoyah/bin/activate,激活虚拟环境

  2. FLASK_APP=app.py,设置临时环境变量,指定 APP

  3. flask run,运行 Flask 项目

  4. --port 8080,指定端口

  5. --host 0.0.0.0,指定服务要监听的IP,这样服务器就可以接受来自任何IP地址的请求

  6. --reload,使服务器在代码改变时自动重启,方便开发时实时更新

  7. --with-threads,允许服务器使用线程来处理请求,提高并发处理能力

  8. > /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
      

架构

一切从客户端发起请求开始

模板

Jiinja2

模板

模板上下文环境

提示 在 Python 脚本里,url_for() 函数需要从 flask 包中导入,而在模板中则可以直接使用,因为 Flask 把一些常用的函数和对象添加到了模板上下文(环境)里。

模板

  1. 模板上下文处理函数
  2. 模板继承

静态文件

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

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://www.sqlalchemy.org/

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_dataflush_product等常用名称不得不附加其他字段保证文件名称唯一,例如export_order_data_6573export_order_data_2024_06_10export_order_data_05_10_error

当然,定期删除临时脚本可以缓解该问题,不过在命名规范上作文章才是从源头解决问题。

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