
示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载导读本文围绕 python-docs-samples 仓库中的 Django 示例应用深入讲解如何在 App Engine 标准环境 Python 3 运行时下通过 Bundled Deferred API 将耗时任务异步化——在 HTTP 请求返回后由 Task Queue 在后台延迟执行。读完本文你将掌握wrap_wsgi_app(use_deferredTrue)的接入方式、deferred.defer()的三种调度形态立即执行、_countdown延时、自定义_url回调、无settings.py文件的最小 Django 配置技巧以及一套基于独立版本隔离的可并行端到端测试方案。一、示例背景Bundled Services 下的 Deferred 服务App Engine 标准环境早年基于 Python 2.7 的运行时提供了一批打包服务Bundled Services其中 Deferred 服务允许开发者把函数调用序列化后投递到任务队列由系统在请求上下文之外异步执行非常适合后台数据聚合、延迟通知、批量更新等场景。在迁移到 Python 3 标准运行时后这类能力并未消失而是通过appengine-python-standard提供的兼容层继续可用。仓库在 bundled-services 目录 下按服务类型deferred、mail、blobstore组织了示例而 deferred 子目录 内放置了三份功能完全一致、仅 Web 框架不同的应用用于演示同一套 Deferred API 的三套写法框架说明示例位置Flask通过app.wsgi_app wrap_wsgi_app(...)包装deferred/flaskDjango通过wrap_wsgi_app(get_wsgi_application(), ...)包装deferred/djangoApp Engine 原生 WSGI直接包装 WSGI 应用deferred/wsgi本文聚焦其中的Django 版本其完整源码位于 main.py。二、应用行为与 URL 路由设计示例应用实现了一个简单的计数器功能闭环清晰GET /counter/get返回当前计数值GET /counter/increment通过 Deferred 服务触发三次递增随后立即返回响应POST /custom/path自定义的 Deferred 任务回调端点。上层 deferred/README.md 描述了通用的行为约定计数器通过 Deferred 服务递增任务分为立即执行一次、延时执行一次、再延时执行一次三档。以本 Django 示例的实际源码为准main.py 中三次调度的间隔分别为0 秒、60 秒_countdown60与 120 秒_countdown120这一点由 main_test.py 中的断言链40s→10、100s→20、160s→30逐段验证。URL 路由在 main.py 中以 Django 的urlpatterns声明urlpatterns ( path(counter/get, view_counter, nameview_counter), path(counter/increment, increment_counter, nameincrement_counter), path(custom/path, custom_deferred, namecustom_deferred), )三、核心源码逐段剖析3.1 无 settings.py 的最小 Django 应用示例刻意省略了独立的settings.py、manage.py与项目脚手架直接在入口模块内通过settings.configure(...)完成最小配置main.pysettings.configure( DEBUGTrue, SECRET_KEYthisisthesecretkey, ROOT_URLCONF__name__, MIDDLEWARE_CLASSES( django.middleware.common.CommonMiddleware, django.middleware.csrf.CsrfViewMiddleware, django.middleware.clickjacking.XFrameOptionsMiddleware, ), ALLOWED_HOSTS[*], )要点说明ROOT_URLCONF__name__把当前模块当作 URL 配置入口配合上方urlpatterns即完成路由装载MIDDLEWARE_CLASSES是 Django 2.x 之前引入的配置键名示例保留它以最小化配置面在新版本 Django 中通常使用MIDDLEWARE此处沿用源码写法以保证示例行为与仓库一致ALLOWED_HOSTS[*]保证 App Engine 生成的*.appspot.com域名请求不会被主机头校验拦截SECRET_KEY仅为演示值生产环境务必通过环境变量注入真实密钥。3.2 WSGI 接入wrap_wsgi_app 与 use_deferredDeferred 服务能够拦截并处理任务回调关键在于把标准 WSGI 应用用 App Engine 的wrap_wsgi_app包装并显式打开use_deferredTruemain.pyapp wrap_wsgi_app(get_wsgi_application(), use_deferredTrue)这一行是整套方案的开关get_wsgi_application()产出标准 Django WSGI 应用wrap_wsgi_app注入 App Engine 打包 API 的兼容处理use_deferredTrue则告知中间件当请求命中 Deferred 任务端点时转交给 Deferred 处理器而非业务视图。若遗漏该参数后台任务回调将无法被正确识别执行。对比同目录下的 Flask 版本写法完全同构app.wsgi_app wrap_wsgi_app(app.wsgi_app, use_deferredTrue)。3.3 数据模型基于 NDB 的计数器计数值存放于 App Engine 的 NDB 数据存储main.pyclass Counter(ndb.Model): count ndb.IntegerProperty(indexedFalse) def do_something_later(key, amount): entity Counter.get_or_insert(key, count0) entity.count amount entity.put()indexedFalse关闭该属性索引——计数器只需要读写、不需要按值查询可降低写入成本get_or_insert(key, count0)是原子化的取或建操作避免并发首写时产生重复实体do_something_later(key, amount)即被投递到后台的任务函数它接收任务参数并对计数累加。3.4 三种 defer 调度形态视图函数 increment_counter 集中展示了deferred.defer()的三种典型用法# 形态一使用默认 URL 与队列、无任务名、立即执行 deferred.defer(do_something_later, my_key, 10) # 形态二默认参数但 60 秒后执行 deferred.defer(do_something_later, my_key, 10, _countdown60) # 形态三自定义任务端点路径并延迟 120 秒执行 deferred.defer(do_something_later, my_key, 10, _url/custom/path, _countdown120)参数语义整理如下参数含义示例取值位置参数传递给任务函数的参数my_key键、10递增量_countdown延迟执行的秒数60、120缺省即尽快执行_url任务回调的 HTTP 路径默认使用系统内置端点此处显式指向/custom/path_queue/_task_name未演示指定任务队列与任务名缺省使用默认队列、自动命名注意所有以_开头的关键字参数属于 Deferred API 的调度元数据不会被序列化传递给任务函数本身。任务函数的普通参数会被 pickle 序列化后随任务一起投递这也解释了 app.yaml 中需要设置NDB_USE_CROSS_COMPATIBLE_PICKLE_PROTOCOL的原因见下文。3.5 自定义任务回调端点当_url指定了非默认路径时应用需要自己提供对应的处理视图。custom_deferred 演示了如何手工把 WSGI 环境交给 Deferred 处理器def custom_deferred(request): print(Executing deferred task.) # request.environ 即 WSGI 的 environ 字典见 PEP 3333 response, status, headers deferred.Handler().post(request.environ) return HttpResponse(response, statusstatus.value)request.environ携带完整的 WSGI 环境变量PEP 3333Deferred 处理器依赖其中解析出任务载荷deferred.Handler().post(environ)负责反序列化任务、调用目标函数并生成响应status是 HTTP 状态对象取其.value得到数值状态码交给 Django 的HttpResponse。需要留意的是常规的use_deferredTrue中间件会自动处理系统默认端点只有像示例这样显式自定义_url时才需要自行实现回调视图。四、app.yaml 与运行环境配置应用运行于 App Engine 标准环境 Python 3 运行时app.yaml 的配置如下runtime: python313 app_engine_apis: true env_variables: NDB_USE_CROSS_COMPATIBLE_PICKLE_PROTOCOL: True三个配置项逐一说明runtime: python313声明运行时版本。Bundled Services 兼容层要求 Python 3 运行时仓库当前采用python313app_engine_apis: true关键开关。它启用 App Engine 打包 APIBundled APIs的访问权限没有它google.appengine.ext.deferred、ndb等模块将无法在 Python 3 运行时工作NDB_USE_CROSS_COMPATIBLE_PICKLE_PROTOCOL: TrueDeferred 任务需要把函数参数 pickle 序列化。该环境变量使 NDB 实体采用跨版本兼容的 pickle 协议保证任务载荷在 Python 2 时代与 Python 3 时代之间可以互通示例的测试与历史数据迁移场景依赖于此。依赖清单见 requirements.txtDjango6.1.1; python_version 3.12 django-environ0.13.0 google-cloud-logging3.5.0 appengine-python-standard0.3.1其中appengine-python-standard是 Python 3 运行时提供 Bundled API 兼容层的核心依赖版本下限0.3.1保证wrap_wsgi_app与deferred模块可用django-environ用于从环境变量读取配置google-cloud-logging提供云端日志对接。五、部署方式按 deferred/README.md 的说明部署只需一条命令gcloud app deploy在项目根目录执行后gcloud会读取 app.yaml上传应用并分配https://版本-dot-项目.appspot.com形式的访问地址。部署完成后即可通过GET /counter/get查看计数、GET /counter/increment触发后台递增任务。六、端到端测试基于版本隔离的并行测试方案Bundled Services 只能在正确配置的 App Engine 应用内运行因此本地单元测试无法覆盖 Deferred 服务的真实行为必须部署到 App Engine 后测试。仓库给出的 main_test.py 提供了一套精巧的测试策略其流程为以独立版本部署调用gcloud app deploy --no-promote --versionuuid启动一个新版本但不向它路由任何网络流量--no-promote的作用直接访问版本专属 URL形如https://version_id-dot-project_id.appspot.com绕开默认流量分配只与本次测试的版本交互验证后删除版本测试结束调用gcloud app versions delete version_id清理资源——正因为该版本从未被路由流量删除不会影响线上服务并行安全由于每个测试都使用独立版本、独立 URL多个测试可以同时运行而互不干扰。测试的计时逻辑精确对应三次 defer 调度main_test.py触发后经过时间期望计数值依据40 秒10立即执行的任务增量 10已完成100 秒2060 秒延时的任务增量 10已完成160 秒30120 秒延时的任务增量 10已完成再等 30 秒30无新任务计数值保持不变测试基础设施方面gcloud_cli()封装了带--quiet --formatjson的 gcloud 调用并用backoff做指数退避重试wait_for_app()会轮询等待新版本初始化完成后再进入断言versionfixture 承担部署与清理的完整生命周期。测试依赖见 requirements-test.txtbackoff、pytest、requestsnox 测试版本矩阵在 noxfile_config.py 中配置忽略 3.8–3.11 与 3.13默认在 3.12 下运行。七、迁移要点与常见坑位结合源码与配置将这套 Django 版 Deferred 示例迁移到自己的应用时建议重点核对以下几点务必开启app_engine_apis: true否则google.appengine.ext.deferred与ndb导入即失败务必以wrap_wsgi_app(get_wsgi_application(), use_deferredTrue)作为最终 WSGI 入口且该包装必须包住 Django 应用最外层否则任务回调无法被中间件捕获自定义_url时必须自行实现回调视图并正确调用deferred.Handler().post(environ)同时确认该路由能接收 POST 请求示例中测试直接以 GET 驱动任务投递但回调端点按 WSGI 规范应接受任务推送任务函数与参数需可 pickle 序列化涉及 NDB 实体跨环境传递时保持NDB_USE_CROSS_COMPATIBLE_PICKLE_PROTOCOL开启测试必须真部署Bundled Services 依赖 App Engine 环境本地pytest无法替代可复用仓库的独立版本 专属 URL 用完即删模式实现并行 CI。延伸阅读三个框架版本的横向对比见 deferred 目录 下的 README.mdFlask 版等价实现deferred/flask/main.py原生 WSGI 版deferred/wsgi同一迁移体系下的其他打包服务示例Mail、Blobstore见 bundled-services 目录仓库整体的 Python 3 标准环境示例见 appengine/standard_python3。赞分享示例工程【免费下载链接】python-docs-samplesCode samples used on cloud.google.com项目地址https://gitcode.com/GitHub_Trending/py/python-docs-samples点击查看免费下载相关推荐python-docs-samples 实战在 App Engine Python 3 中接入 Bundled Deferred 服务Flask / Django / WSGI 三框架实现python docs samples 实战在 App Engine Python 3 中接入 Bundled Deferred 服务Flask / Dja示例工程使用 Django 访问 App Engine 标准环境 Python 3 的 bundled Blobstore 服务 API使用 Django 访问 App Engine 标准环境 Python 3 的 bundled Blobstore 服务 API 在 App Engine 标准示例工程OpenProject 应用层开发规范app/CLAUDE.md 代码约定与工作包语义标识符机制OpenProject 应用层开发规范app/CLAUDE.md 代码约定与工作包语义标识符机制 app/CLAUDE.md 是 OpenProject 仓库示例工程上一篇从激活困境到自动化解决方案KMS_VL_ALL_AIO的技术实现与应用指南下一篇ik_llama.cpp FlashMLA 崩溃排障实录GGML_ASSERT(fms.S[j] 0) 失败的定位、修复与复现验证创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考