谢乾坤 | Kingname

给时光以生命。

如果你在Flask中启动过子线程,然后在子线程中读写过g对象或者尝试从request对象中读取url参数,那么,你肯定对下面这个报错不陌生:RuntimeError: Working outside of request context..

例如下面这段Flask代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import threading
from flask import Flask, request

app = Flask(__name__)

def inner_func():
doc_id = request.args.get('doc_id', '')
print(f'用户ID为:{doc_id}')


@app.route('/start_thread')
def start_thread():
thread = threading.Thread(target=inner_func)
thread.start()
return {'success': True, 'msg': '获取用户ID成功!'}

请求/start_thread接口就会报错,如下图所示:

如果你在网上搜索flask thread RuntimeError: Working outside of request context. ,那么你可能会看到官方文档或者StackOverFlow上面提供了一个装饰器@copy_current_request_context。如下图所示:

照着它这样写,确实能解决问题,如下图所示:

但无论是官网还是StackOverFlow,它的例子都非常简单。但是我们知道,启动线程有很多种方法,例如:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 方法一,启动简单线程
import threading

job = threading.Thread(target=函数名, args=(参数1, 参数2), kwargs={'参数3': xxx, '参数4': yyy})
job.start()

# 方法2,使用类定义线程
import threading

class Job(threading.Thread):
def __init__(self, 参数):
super().__init__()

def run(self):
print('子线程开始运行')

job = Job(参数)
job.start()

# 方法3,使用线程池
from multiprocessing.dummy import Pool
pool = Pool(5) # 5个线程的线程池
pool.map(函数名, 参数列表)

网上的方法只能解决第一种写法的问题。如果想使用方法2和方法3启动子线程,代码应该怎么写呢?如果在子线程中又启动子线程,再用一次@copy_current_request_context还行吗?

相信我,你在网上搜索一下午,只有两种结果:一是找不到答案,二是找到的答案是晚于2023年1月14日的,因为是别人看了我这篇文章以后,再写的。

阅读全文 »

我们知道,在软件工程中,单元测试是保证软件质量的重要手段之一。一个优秀的代码,单元测试的代码量,经常会超过被测试的代码本身。一个理想化的开发团队,可能有三分之二的时间是在写测试,剩下的三分之一时间才是写业务代码。

阅读全文 »

在以前的文章中,在微信群中,我多次强调,写函数的时候,不要把所有参数放到一个大字典里面作为参数到处传,否则时间久了以后,根本不知道字典里面有哪些数据:

1
2
3
4
def parse(data):
name = data['name']
age = data['age']
xxx = data['xx']

上面这样写,对原作者来说确实简单,但是如果代码还有别人来维护,他就根本不知道这个字典里面有哪些数据。必须要一层一层查找调用链,费时费力。

阅读全文 »

如果你使用过嘀嗒清单或者Todoist,那你应该知道他们有一个很好用的功能,那就是自动识别任务中的时间,例如:

1
下周二下午三点给老板发邮件

它会自动识别为:

今天,公众号粉丝群里面,有一个叫做NowAnti的同学推荐了一个项目,叫做司南,它就可以让Python实现这样的功能。

阅读全文 »

我们知道,在Python里面,可以使用input获取用户的输入。例如:

但有一个问题,如果你什么都不输入,程序会永远卡在这里。有没有什么办法,可以给input设置超时时间呢?如果用户在一定时间内不输入,就自动使用默认值。

阅读全文 »
0%