flask框架之多app应用、flask-scrip、导出依赖、偏函数、threading.local、请求上下文执行流程

多app应用

以前我们在写flask都是实例化得到一个app对象，其实flask是可以使用多个app

from flask import Flask
from werkzeug.middleware.dispatcher import DispatcherMiddleware
from werkzeug.serving import run_simple

app01 = Flask('app01')
app02 = Flask('app02')


@app01.route('/app01')
def index():
    return 'app01'


dm = DispatcherMiddleware(app01, {'/app02': app02})


@app02.route('/app02')
def home():
    return 'app02'


if __name__ == '__main__':
    run_simple('localhost', 5000, dm)

ps：内部执行

1.请求来了会执行dm(environ,start_response)

2.dm的__call__根据请求的地址拿到不同的app，执行app(environ,start_response)

flask_script使用

通过这个第三方模块我们可以定制命令，完成像django的 python manage.py runserver的命令

安装模块

pip3 install flask-script

'''
注意依赖版本：
Flask==2.2.2
Flask_Script==2.0.3
'''

快速使用

from flask import Flask
from flask_script import Manager

app = Flask(__name__)

manager = Manager(app)

if __name__ == '__main__':
    manager.run()

# 启动项目
# python 项目文件  runserver 

自定制命令

from flask import Flask
from flask_script import Manager

app = Flask(__name__)
manager = Manager(app)


@manager.command
def custom(arg):
    """
    自定义命令
    python 项目文件.py custom 123
    :param arg:
    :return:
    """
    print(arg)


@manager.option('-n', '--name', dest='name')
@manager.option('-u', '--url', dest='url')
def cmd(name, url):
    """
    自定义命令（-n也可以写成--name）
    执行： python 项目文件.py  cmd -n lqz -u https://www.baidu.com
    执行： python 项目文件.py  cmd --name lqz --url https://www.jd.com
    :param name:
    :param url:
    :return:
    """
    print(name, url)


if __name__ == '__main__':
    manager.run()

django中自定义命令步骤

1.在app中新建包：managenent

2.在management下新建包：commands

3.commands新建py文件，py文件名就是命令名 init.py

4.init.py写入

from django.core.management.base import BaseCommand, CommandError
class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument('--name', type=str) # 指令接受的参数列表，参数名左边有两个减号，可以添加类型限制

    def handle(self, *args, **options):
        name = options['name'] 

# python manage.py init --name=abc

导出项目依赖

第一种：将当前虚拟环境中所有的模块都保存下来但不包括 pip、wheel、setuptools 等自带包，配合虚拟环境virtualenv使用比较好

pip freeze 生成>>>> requirements.txt

第二种：使用真实环境导出依赖，会把所有模块都导出在下载依赖的时候可能会出现问题

- 安装：pip3 install pipreqs
- 生成依赖文件：pipreqs ./ --encoding=utf8
- 安装依赖文件：pip3 install -r requirements.txt 

偏函数

python内置给我们的一个偏函数，将函数包裹一下提前传参数

# 导入一下模块
from functools import partial


# 定义函数
def add(a,b,c):
    return a+b+c


# 正常情况调用
res = add(1,2,3)
print(res)

# 使用偏函数
res = partial(add,1,2)
print(res(3))

threading.local

多线程操作同一个变量时如果不加锁会出现数据的错乱问题，threading.local作用是内部填充的变量属于当前线程该变量对其他线程而言是隔离的，也就是说该变量是当前线程独有的变量。threading.local为变量在每个线程中都创建了一个副本，那么每个线程可以访问自己内部的副本变量

自定义local对象解决数据错乱

flask的request和session都是全局的但是我们在不同的视图函数中使用的是当次请求的对象它的底层就是基于local写的

flask部署支持多进程线程架构，也支持协程架构，内部通过重写local来完成

# 通过捕获异常兼容线程和协程
try:
    from greenlet import getcurrent as get_ident
except Exception as e:
    from threading import get_ident
from threading import Thread
import time


class Local(object):
    def __init__(self):
        # self.storage = {}   # 只要self.属性，就会调用 __setattr__，内部又掉了self.storage--->递归了
        # 类来调用对象的绑定方法__setattr__，这个方法就会变成函数，有几个值就要传几个值
        # 本质就是完成 self.storage = {} 要完成的事，但是不会触发递归调用
        object.__setattr__(self, 'storage', {})
        # setattr(self,'storage', {})  # 反射的方式设置值，也会触发递归

    def __setattr__(self, k, v):
        ident = get_ident()
        if ident in self.storage:
            self.storage[ident][k] = v
        else:
            self.storage[ident] = {k: v}

    def __getattr__(self, k):
        ident = get_ident()  # 在协程中，gevent中是获取协程id号，如果在线程中，获取的是线程id号
        return self.storage[ident][k]


obj = Local()  # 每个local对象，用自己的字典


def task(arg):
    obj.val = arg
    v = obj.val
    time.sleep(0.01)
    print(v)


for i in range(10):
    t = Thread(target=task, args=(i,))
    t.start()

local的本质实现

设置值：
    - 线程1：local.val='abc'   ------> {'线程1ID号':{val:abc},}
    - 线程2：local.val='cba'   ------> {'线程1ID号':{val:abc},'线程2ID号':{val:cba},}

取值：
    - 线程1：print(local.val)  ---->l={'线程1id号':{val:abc},'线程2id号':{val:cba},}--》先当前线程的id号：get_ident()   l[get_ident(线程1)]['val']
    - 线程2：print(local.val)  ---->l={'线程1id号':{val:abc},'线程2id号':{val:cba},}--》先当前线程的id号：get_ident()   l[get_ident(线程2)]['val']

flask 自定义的local，支持线程和携程 

class Local(object):

    def __init__(self):
        object.__setattr__(self, "__storage__", {})
        object.__setattr__(self, "__ident_func__", get_ident)


    def __getattr__(self, name):
        try:
            return self.__storage__[self.__ident_func__()][name]
        except KeyError:
            raise AttributeError(name)

    def __setattr__(self, name, value):
        ident = self.__ident_func__()
        storage = self.__storage__
        try:
            storage[ident][name] = value
        except KeyError:
            storage[ident] = {name: value}

flask请求上下文分析

请求来了先执行app()，调用Flask类的__call__方法

# self是app对象
def __call__(self, environ, start_response):
    return self.wsgi_app(environ, start_response)

__call__方法中执行了self.wsgi_app

def wsgi_app(self, environ, start_response):
		# 执行self.request_context(environ)
		# 返回的是RequestContext类的对象
        # environ:http请求拆成了字典
        # ctx对象：RequestContext类的对象，对象里有：当次的requets对象，app对象，session对象
        ctx = self.request_context(environ)
        error = None
        try:
            try:
                #ctx RequestContext类 push方法
                ctx.push()
                # 匹配成路由后，执行视图函数
                response = self.full_dispatch_request()
            except Exception as e:
                error = e
                response = self.handle_exception(e)
            except:  # noqa: B001
                error = sys.exc_info()[1]
                raise
            return response(environ, start_response)
        finally:
            if self.should_ignore_error(error):
                error = None
            ctx.auto_pop(error)

self.request_context(environ)执行

def request_context(self, environ):
    return RequestContext(self, environ)
# 返回RequestContext类的对象

RequestContext类返回的ctx对象

class RequestContext:
	def __init__(self, app, environ, request=None, session=None):
	    self.app = app
	    if request is None:
	    	#app.request_class就是flask/wrappers里的Request
	        request = app.request_class(environ)
	    self.request = request
	    self.url_adapter = None
	    try:
	        self.url_adapter = app.create_url_adapter(self.request)
	    except HTTPException as e:
	        self.request.routing_exception = e
	    self.flashes = None
	    self.session = session

分析ctx.push的执行

# RequestContext---->ctx.push
def push(self):
    # _request_ctx_stack = LocalStack() ---》push(ctx对象)--》ctx:request，session，app
    _request_ctx_stack.push(self)
    #session相关的
    if self.session is None:
        session_interface = self.app.session_interface
        self.session = session_interface.open_session(self.app, self.request)

        if self.session is None:
            self.session = session_interface.make_null_session(self.app)
		# 路由匹配相关的
    if self.url_adapter is not None:
        self.match_request()


# LocalStack()---->push
def push(self, obj):
        #self._local  _local 就是咱们刚刚自己写的Local的对象---》LocalStack的init初始化的_local---》self._local = Local()---》Local对象可以根据线程区分数据 
        rv = getattr(self._local, "stack", None)
        if rv is None:
            rv = []
            self._local.stack = rv  # self._local.stack 根据不同线程用的是自己的数据
        rv.append(obj)  # self._local.stack.append(obj)
        # {'线程id号'：{stack：[ctx]},'线程id号2'：{stack：[ctx]}}
        return rv

Local类的__getattr__方法

def __getattr__(self, name):
    try:
    	# 根据线程id取stack属性
        return self.__storage__[self.__ident_func__()][name]
    except KeyError:
        raise AttributeError(name)

Local类的__setattr__方法

def __setattr__(self, name, value):
	# 获取线程id
    ident = self.__ident_func__()
    # 获取存储字典，现在是一个空字典
    storage = self.__storage__
    try:
    	# 线程存在就修改
        storage[ident][name] = value
    except KeyError:
    	# 线程不存在就新增
    	# storage = {线程id:{'stack':[]}}
        storage[ident] = {name: value}