Werkzeug是一个全面的WSGI Web应用程序库。它最初是WSGI实用程序各种工具的简单集合，现已成为最高级的WSGI实用程序库之一，是Flask背后的项目。Werkzeug 是一个德语单词，工具的意思。这个单词发音对我来说，有点困难（可能也是它知名度不高的重要因素之一），刚好官方logo是个锤子，我就简称“德国锤子”。文章已经完成上下两篇，上篇介绍:
下篇介绍:
“德国锤子” 还有3个比较重要的功能，不要放过，我们继续学习:
reloader是调试程序时非常实用的功能，开发的时候不用手动重启服务，修改代码后会自动重启服务，提高研发效率。运行示例中的Shorty服务:
# python3 shortly.py
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
* Restarting with stat
* Debugger is active!
* Debugger PIN: 722-230-382程序启动后实际上有2个进程， 10527 的主进程和 10529 的子进程:
501 10527 7144 0 8:36上午 ttys008 0:00.23 .../Python.app/Contents/MacOS/Python shortly.py
501 10529 10527 0 8:36上午 ttys008 0:00.34 .../Python.app/Contents/MacOS/Python /Users/yoo/work/yuanmahui/python/ch20-werkzeug/shortly.py随便修改一下 shortly.py 代码，比如增加一个日志输出。可以发现启动日志有下面reload的信息:
...
* Detected change in '/Users/yoo/work/yuanmahui/python/ch20-werkzeug/shortly.py', reloading
* Restarting with stat
...再观查进程信息可以发现 10529 子进程已经退出，新增了 10634 子进程:
501 10527 7144 0 8:36上午 ttys008 0:00.24 .../Python.app/Contents/MacOS/Python shortly.py
501 10634 10527 0 8:38上午 ttys008 0:00.77 .../Python.app/Contents/MacOS/Python /Users/yoo/work/yuanmahui/python/ch20-werkzeug/shortly.py可以推测，主/子进程检测代码变动，然后子进程关闭/退出，再由主进程重新创建一个子进程。那么到底子进程是自动退出还是被主进程关闭？是主进程监听代码变化还是子进程监听的呢？我们带着这2个问题，一起看看代码实现。
启动服务的时候，需要使用 use_reloader=True 参数启动reloader
run_simple("127.0.0.1", 5000, app, use_debugger=True, use_reloader=True)
服务启动时候判断是独立启动，还是由reload启动:
# serving.py
def run_simple(...):
if not is_running_from_reloader():
...
from ._reloader import run_with_reloader as _rwr
_rwr(
inner,
extra_files=extra_files,
exclude_patterns=exclude_patterns,
interval=reloader_interval,
reloader_type=reloader_type,
)主进程和子进程的判断是通过 WERKZEUG_RUN_MAIN 的环境变量进行判断，默认情况下是没有这个环境变量:
def is_running_from_reloader() -> bool:
return os.environ.get("WERKZEUG_RUN_MAIN") == "true"使用reloader启动的代码如下:
# _reloader.py
def run_with_reloader(
main_func: t.Callable[[], None],
extra_files: t.Optional[t.Iterable[str]] = None,
exclude_patterns: t.Optional[t.Iterable[str]] = None,
interval: t.Union[int, float] = 1,
reloader_type: str = "auto",
) -> None:
"""Run the given function in an independent Python interpreter."""
import signal
signal.signal(signal.SIGTERM, lambda *args: sys.exit(0))
reloader = reloader_loops[reloader_type](
extra_files=extra_files, exclude_patterns=exclude_patterns, interval=interval
)
try:
if os.environ.get("WERKZEUG_RUN_MAIN") == "true":
ensure_echo_on()
t = threading.Thread(target=main_func, args=())
t.daemon = True
# Enter the reloader to set up initial state, then start
# the app thread and reloader update loop.
with reloader:
t.start()
reloader.run()
else:
sys.exit(reloader.restart_with_reloader())
except KeyboardInterrupt:
pass代码主要功能:
stat 的实现，还有一种是 watchdog 的实现。后者需要额外安装，但是效率会高一些关于stat和watchdog的区别，请看下面的官方文档:
默认stat后端只是mtime定期检查所有文件的 。这对于大多数情况来说已经足够了，但是众所周知，它会耗尽笔记本电脑的电池。
在watchdog后端使用文件系统事件，而且比stat快, 但是它需要 安装看门狗模块。实现此目的的推荐方法是添加 Werkzeug[watchdog]到您的需求文件中。
ReloaderLoop是reloader实现的基类:
class ReloaderLoop:
name = ""
def __init__(
self,
extra_files: t.Optional[t.Iterable[str]] = None,
exclude_patterns: t.Optional[t.Iterable[str]] = None,
# 默认1s的间隔
interval: t.Union[int, float] = 1,
) -> None:
self.extra_files: t.Set[str] = {os.path.abspath(x) for x in extra_files or ()}
self.exclude_patterns: t.Set[str] = set(exclude_patterns or ())
self.interval = interval
def __enter__(self) -> "ReloaderLoop":
"""Do any setup, then run one step of the watch to populate the
initial filesystem state.
"""
self.run_step()
return self
def __exit__(self, exc_type, exc_val, exc_tb): # type: ignore
"""Clean up any resources associated with the reloader."""
pass
def run(self) -> None:
"""Continually run the watch step, sleeping for the configured
interval after each step.
"""
while True:
self.run_step()
time.sleep(self.interval)
def run_step(self) -> None:
"""Run one step for watching the filesystem. Called once to set
up initial state, then repeatedly to update it.
"""
pass
def restart_with_reloader(self) -> int:
"""Spawn a new Python interpreter with the same arguments as the
current one, but running the reloader thread.
"""
while True:
_log("info", f" * Restarting with {self.name}")
args = _get_args_for_reloading()
new_environ = os.environ.copy()
new_environ["WERKZEUG_RUN_MAIN"] = "true"
exit_code = subprocess.call(args, env=new_environ, close_fds=False)
if exit_code != 3:
return exit_code
def trigger_reload(self, filename: str) -> None:
self.log_reload(filename)
sys.exit(3)
def log_reload(self, filename: str) -> None:
filename = os.path.abspath(filename)
_log("info", f" * Detected change in {filename!r}, reloading")subprocess.call 创建一个子进程，并监听子进程的退出状态。如果退出状态为3则可以无限循环；如果不为3则会退出循环，结束主进程。StatReloaderLoop的实现比较简单，代码如下:
class StatReloaderLoop(ReloaderLoop):
name = "stat"
def __enter__(self) -> ReloaderLoop:
self.mtimes: t.Dict[str, float] = {}
return super().__enter__()
def run_step(self) -> None:
for name in chain(_find_stat_paths(self.extra_files, self.exclude_patterns)):
try:
mtime = os.stat(name).st_mtime
except OSError:
continue
old_time = self.mtimes.get(name)
if old_time is None:
self.mtimes[name] = mtime
continue
if mtime > old_time:
self.trigger_reload(name)所以主进程只是负责持续创建子进程，子进程自己检测代码变化和自动退出。
在Shortly中增加一个echo的view:
Rule("/echo", endpoint="echo"),
def on_echo(self, request):
raise访问这个view可以看到下面的异常信息, 带有完整的业务堆栈:
点击右侧的console图标，会提示输入PIN:
在console中可以进行调试:
2.0.1版本的dubug可能有bug，需要使用os.environ["WERKZEUG_DEBUG_PIN"] = "off"关闭pin-auth的认证，才能够正常工作
python3自带REPL的模块 code, 看起来和python的命令行差不多，仔细对比会发现多了 (InteractiveConsole) 的输出。自己的应用程序中嵌入code，就可以实现交互式的debug功能。
# python3 -m code
Python 3.8.5 (v3.8.5:580fbb018f, Jul 20 2020, 12:11:27)
[Clang 6.0 (clang-600.0.57)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole) # <- 提示信息
>>>
# python3
Python 3.8.5 (v3.8.5:580fbb018f, Jul 20 2020, 12:11:27)
[Clang 6.0 (clang-600.0.57)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>>启动服务的入口，如果有debug参数则使用DebuggedApplication来包裹业务app:
# serving.py
def run_simple(...):
if use_debugger:
from .debug import DebuggedApplication
application = DebuggedApplication(application, use_evalex)DebuggedApplication主要的call函数:
# debug
def __call__(
self, environ: "WSGIEnvironment", start_response: "StartResponse"
) -> t.Iterable[bytes]:
"""Dispatch the requests."""
request = Request(environ)
response = self.debug_application
if request.args.get("__debugger__") == "yes":
# 处理debug请求
cmd = request.args.get("cmd")
arg = request.args.get("f")
secret = request.args.get("s")
frame = self.frames.get(request.args.get("frm", type=int))
if cmd == "resource" and arg:
response = self.get_resource(request, arg) # type: ignore
elif cmd == "pinauth" and secret == self.secret:
response = self.pin_auth(request) # type: ignore
elif cmd == "printpin" and secret == self.secret:
response = self.log_pin_request() # type: ignore
elif (
self.evalex
and cmd is not None
and frame is not None
and self.secret == secret
and self.check_pin_trust(environ)
):
response = self.execute_command(request, cmd, frame) # type: ignore
elif (
self.evalex
and self.console_path is not None
and request.path == self.console_path
):
response = self.display_console(request) # type: ignore
return response(environ, start_response)debug的重点在debug_application函数:
def debug_application(
self, environ: "WSGIEnvironment", start_response: "StartResponse"
) -> t.Iterator[bytes]:
"""Run the application and conserve the traceback frames."""
app_iter = None
try:
# 正常业务
app_iter = self.app(environ, start_response)
yield from app_iter
if hasattr(app_iter, "close"):
app_iter.close() # type: ignore
except Exception:
# 异常调试
if hasattr(app_iter, "close"):
app_iter.close() # type: ignore
traceback = get_current_traceback(
skip=1,
show_hidden_frames=self.show_hidden_frames,
ignore_system_exceptions=True,
)
for frame in traceback.frames:
self.frames[frame.id] = frame
self.tracebacks[traceback.id] = traceback
try:
start_response(
"500 INTERNAL SERVER ERROR",
[
("Content-Type", "text/html; charset=utf-8"),
("X-XSS-Protection", "0"),
],
)
except Exception:
environ["wsgi.errors"].write(
"Debugging middleware caught exception in streamed "
"response at a point where response headers were already "
"sent.\n"
)
else:
is_trusted = bool(self.check_pin_trust(environ))
yield traceback.render_full(
evalex=self.evalex, evalex_trusted=is_trusted, secret=self.secret
).encode("utf-8", "replace")
traceback.log(environ["wsgi.errors"])app_iter = self.app(environ, start_response) 执行业务功能debug可以使用web界面调试程序，这会产生安全问题。所以werkzeug的debug中引入了PIN机制，需要输入PIN验证码才可以进行调试，PIN在服务端的命令行中输出。
前端页面输入PIN后会提交PIN码:
# debugger.js
function initPinBox() {
document.querySelector(".pin-prompt form").addEventListener(
"submit",
function (event) {
....
fetch(
`${document.location.pathname}?__debugger__=yes&cmd=pinauth&pin=${pin}&s=${encodedSecret}`
)
.then((res) => res.json())
.then(({auth, exhausted}) => {
if (auth) {
EVALEX_TRUSTED = true;
fadeOut(document.getElementsByClassName("pin-prompt")[0]);
} else {
....
}
})
...
},
false
);
}这段JS代码重点是 __debugger__=yes&cmd=pinauth&pin=${pin}&s=${encodedSecret} 的URL，这个请求会直接被DebuggedApplication处理:
if request.args.get("__debugger__") == "yes":
cmd = request.args.get("cmd")
arg = request.args.get("f")
secret = request.args.get("s")
frame = self.frames.get(request.args.get("frm", type=int)) # type: ignore
...
elif cmd == "pinauth" and secret == self.secret:
response = self.pin_auth(request) # type: ignore
...
elif (
self.evalex
and cmd is not None
and frame is not None
and self.secret == secret
and self.check_pin_trust(environ)
):
response = self.execute_command(request, cmd, frame) # type: ignoreauth认证函数:
def pin_auth(self, request: Request) -> Response:
"""Authenticates with the pin."""
exhausted = False
auth = False
trust = self.check_pin_trust(request.environ)
...
# Otherwise go through pin based authentication
else:
entered_pin = request.args["pin"]
# 对比PIN
if entered_pin.strip().replace("-", "") == pin.replace("-", ""):
self._failed_pin_auth = 0
auth = True
else:
self._fail_pin_auth()
rv = Response(
json.dumps({"auth": auth, "exhausted": exhausted}),
mimetype="application/json",
)
if auth:
# 设置cookie
rv.set_cookie(
self.pin_cookie_name,
f"{int(time.time())}|{hash_pin(pin)}",
httponly=True,
samesite="None",
)
elif bad_cookie:
rv.delete_cookie(self.pin_cookie_name)
return rv查看http请求详情，会发现认证成功后会设置一个Response-Cookie：__wzd10d9760bb71ac5d1b21e ，这样后续的debug调试都使用这个cookie。
开启PIN验证后，无法调试就是因为这个cookie在调试的时候没有附带上
debug调试主要使用_InteractiveConsole的runsource实现:
class _InteractiveConsole(code.InteractiveInterpreter):
locals: t.Dict[str, t.Any]
...
def runsource(self, source: str, **kwargs: t.Any) -> str: # type: ignore
source = f"{source.rstrip()}\n"
ThreadedStream.push()
prompt = "... " if self.more else ">>> "
try:
source_to_eval = "".join(self.buffer + [source])
if super().runsource(source_to_eval, "<debugger>", "single"):
self.more = True
self.buffer.append(source)
else:
self.more = False
del self.buffer[:]
finally:
output = ThreadedStream.fetch()
return prompt + escape(source) + output
...runsource函数比较复杂，但是核心就是对前端网页提交的字符串信息进行编译执行，并把执行的输出捕获后反馈给前端，和之前的介绍CGI实现类似。
开发的时候，可以利用debug功能协助进行调试，提高研发效率。需要注意的是，debug功能不可以用于线上环境。
数据库操作是Web程序非常重要的一环，werkzeug中操作数据可以使用sqlalchemy。
在正式介绍Werkzeug配合SQLAlchemy操作数据库操作数据库之前，我们先简单回顾一下SQLAlchemy的ORM使用:
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite:///:memory:', echo=True)
Model = declarative_base()
class User(Model):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
fullname = Column(String)
nickname = Column(String)
def __repr__(self):
return "<User(name='%s', fullname='%s', nickname='%s')>" % (
self.name, self.fullname, self.nickname)
Model.metadata.create_all(engine)
print("=" * 10)
Session = sessionmaker(bind=engine)
session = Session()
ed_user = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
session.add(ed_user)
session.commit()
print(ed_user.id)
result = engine.execute("select * from users")
for row in result:
print(row)示例shorty中演示了如何使用SQLAlchemy操作SQLite数据库。使用方法是先使用 initdb 初始化sqlite数据库，然后再使用 runserver 启动服务:
python3 manage-shorty.py initdb
python3 manage-shorty.py runserver先看看View中如何通过ORM操作数据，这是插入数据:
uid = URL(url, "private" not in request.form, alias).uid
session.commit()这是查询数据:
url = URL.query.get(uid)
URL的数据模型是这样定义的:
url_table = Table(
"urls",
metadata,
Column("uid", String(140), primary_key=True),
Column("target", String(500)),
Column("added", DateTime),
Column("public", Boolean),
)
class URL:
query = session.query_property()
def __init__(self, target, public=True, uid=None, added=None):
self.target = target
self.public = public
self.added = added or datetime.utcnow()
if not uid:
while 1:
uid = get_random_uid()
if not URL.query.get(uid):
break
self.uid = uid
session.add(self)
@property
def short_url(self):
return url_for("link", uid=self.uid, _external=True)
def __repr__(self):
return f"<URL {self.uid!r}>"
mapper(URL, url_table)database_engine在创建创建App时候创建:
class Shorty:
def __init__(self, db_uri):
local.application = self
self.database_engine = create_engine(db_uri, convert_unicode=True)
self.dispatch = SharedDataMiddleware(self.dispatch, {"/static": STATIC_PATH})
def init_database(self):
# 等同 Model.metadata.create_all(engine)
metadata.create_all(self.database_engine)最关键的地方是session:
local = Local()
local_manager = LocalManager([local])
# local.application = self
application = local("application")
...
session = scoped_session(
lambda: create_session(
application.database_engine, autocommit=False, autoflush=False
)
)简单的说这里scoped_session是绑定到线程的，跟随请求的生命周期。这样在请求中可以使用session访问数据。
本章我们学习了web框架如何使用reload和debug协助研发，提高研发效率。reload主要是使用了subprocess开启多个python进程，debug则是使用code的REPL功能。
werkzeug也提供了使用sqlalchemy操作数据库的示例Shorty，使用ORM功能可以快速编写适配多种数据存储引擎的程序。
本文由哈喽比特于3年以前收录，如有侵权请联系我们。
文章来源：https://mp.weixin.qq.com/s/GNZM9krOc3v8DLI684RPzw
京东创始人刘强东和其妻子章泽天最近成为了互联网舆论关注的焦点。有关他们“移民美国”和在美国购买豪宅的传言在互联网上广泛传播。然而，京东官方通过微博发言人发布的消息澄清了这些传言，称这些言论纯属虚假信息和蓄意捏造。
日前，据博主“@超能数码君老周”爆料，国内三大运营商中国移动、中国电信和中国联通预计将集体采购百万台规模的华为Mate60系列手机。
据报道，荷兰半导体设备公司ASML正看到美国对华遏制政策的负面影响。阿斯麦（ASML）CEO彼得·温宁克在一档电视节目中分享了他对中国大陆问题以及该公司面临的出口管制和保护主义的看法。彼得曾在多个场合表达了他对出口管制以及中荷经济关系的担忧。
今年早些时候，抖音悄然上线了一款名为“青桃”的 App，Slogan 为“看见你的热爱”，根据应用介绍可知，“青桃”是一个属于年轻人的兴趣知识视频平台，由抖音官方出品的中长视频关联版本，整体风格有些类似B站。
日前，威马汽车首席数据官梅松林转发了一份“世界各国地区拥车率排行榜”，同时，他发文表示：中国汽车普及率低于非洲国家尼日利亚，每百户家庭仅17户有车。意大利世界排名第一，每十户中九户有车。
近日，一项新的研究发现，维生素 C 和 E 等抗氧化剂会激活一种机制，刺激癌症肿瘤中新血管的生长，帮助它们生长和扩散。
据媒体援引消息人士报道，苹果公司正在测试使用3D打印技术来生产其智能手表的钢质底盘。消息传出后，3D系统一度大涨超10%，不过截至周三收盘，该股涨幅回落至2%以内。
9月2日，坐拥千万粉丝的网红主播“秀才”账号被封禁，在社交媒体平台上引发热议。平台相关负责人表示，“秀才”账号违反平台相关规定，已封禁。据知情人士透露，秀才近期被举报存在违法行为，这可能是他被封禁的部分原因。据悉，“秀才”年龄39岁，是安徽省亳州市蒙城县人，抖音网红，粉丝数量超1200万。他曾被称为“中老年...
9月3日消息，亚马逊的一些股东，包括持有该公司股票的一家养老基金，日前对亚马逊、其创始人贝索斯和其董事会提起诉讼，指控他们在为 Project Kuiper 卫星星座项目购买发射服务时“违反了信义义务”。
据消息，为推广自家应用，苹果现推出了一个名为“Apps by Apple”的网站，展示了苹果为旗下产品（如 iPhone、iPad、Apple Watch、Mac 和 Apple TV）开发的各种应用程序。
特斯拉本周在美国大幅下调Model S和X售价，引发了该公司一些最坚定支持者的不满。知名特斯拉多头、未来基金（Future Fund）管理合伙人加里·布莱克发帖称，降价是一种“短期麻醉剂”，会让潜在客户等待进一步降价。
据外媒9月2日报道，荷兰半导体设备制造商阿斯麦称，尽管荷兰政府颁布的半导体设备出口管制新规9月正式生效，但该公司已获得在2023年底以前向中国运送受限制芯片制造机器的许可。
近日，根据美国证券交易委员会的文件显示，苹果卫星服务提供商 Globalstar 近期向马斯克旗下的 SpaceX 支付 6400 万美元（约 4.65 亿元人民币）。用于在 2023-2025 年期间，发射卫星，进一步扩展苹果 iPhone 系列的 SOS 卫星服务。
据报道，马斯克旗下社交平台𝕏（推特）日前调整了隐私政策，允许 𝕏 使用用户发布的信息来训练其人工智能（AI）模型。新的隐私政策将于 9 月 29 日生效。新政策规定，𝕏可能会使用所收集到的平台信息和公开可用的信息，来帮助训练 𝕏 的机器学习或人工智能模型。
9月2日，荣耀CEO赵明在采访中谈及华为手机回归时表示，替老同事们高兴，觉得手机行业，由于华为的回归，让竞争充满了更多的可能性和更多的魅力，对行业来说也是件好事。
《自然》30日发表的一篇论文报道了一个名为Swift的人工智能（AI）系统，该系统驾驶无人机的能力可在真实世界中一对一冠军赛里战胜人类对手。
近日，非营利组织纽约真菌学会（NYMS）发出警告，表示亚马逊为代表的电商平台上，充斥着各种AI生成的蘑菇觅食科普书籍，其中存在诸多错误。
社交媒体平台𝕏（原推特）新隐私政策提到：“在您同意的情况下，我们可能出于安全、安保和身份识别目的收集和使用您的生物识别信息。”
2023年德国柏林消费电子展上，各大企业都带来了最新的理念和产品，而高端化、本土化的中国产品正在不断吸引欧洲等国际市场的目光。
罗永浩日前在直播中吐槽苹果即将推出的 iPhone 新品，具体内容为：“以我对我‘子公司’的了解，我认为 iPhone 15 跟 iPhone 14 不会有什么区别的，除了序（列）号变了，这个‘不要脸’的东西，这个‘臭厨子’。
