python-docs-samples 实战：在 App Engine 标准环境中以多种方式请求 URL（urlfetch 全解析）

• 示例工程

导读

本指南围绕 appengine/standard/urlfetch/README.md 所介绍的示例应用展开：它演示了在 App Engine 标准环境（Python 2.7 runtime）中请求外部 URL 的多种方式。读完本文，你将掌握urlfetch同步调用、异步 RPC 调用与回调、基于requests库的透明适配，以及如何部署与测试这些示例，可直接套用到自己的 App Engine 项目中。

示例整体结构与运行前提

README 指出该示例应用的目标是"demonstrates different ways to request a URL on App Engine"。仓库中对应的完整实现位于 appengine/standard/urlfetch 目录，包含三个可独立部署的子应用：

三个子应用均使用 Python 2.7 的 App Engine 标准环境配置，例如 snippets/app.yaml：

runtime: python27 api_version: 1 threadsafe: yes handlers: - url: .* script: main.app

要点说明：

• runtime: python27表示运行在 App Engine 标准环境的 Python 2.7 沙箱中，只有该环境才内置google.appengine.api.urlfetch；
• threadsafe: yes允许并发处理请求，因此异步 RPC 模式才有实际意义；
• handlers将全部 URL 交给 WSGI 应用对象（main.app或rpc.app）处理。

方式一：同步请求——urlfetch.fetch与urllib2对比

snippets/main.py 首先对比了两种同步获取 URL 的写法。

使用urllib2

url = "http://www.google.com/humans.txt" try: result = urllib2.urlopen(url) self.response.write(result.read()) except urllib2.URLError: logging.exception("Caught exception fetching url")

这是标准的 Python 2.7 标准库写法：urlopen返回类文件对象，read()读取响应体。在 App Engine 环境中，urllib2底层也会被路由到 urlfetch 服务，因此可正常工作，但错误类型是urllib2.URLError，语义上与 urlfetch 自有异常体系不同。

使用urlfetchAPI

from google.appengine.api import urlfetch url = "http://www.google.com/humans.txt" try: result = urlfetch.fetch(url) if result.status_code == 200: self.response.write(result.content) else: self.response.status_code = result.status_code except urlfetch.Error: logging.exception("Caught exception fetching url")

与urllib2相比，urlfetch.fetch直接返回带有status_code与content的结果对象，无需先捕获 HTTP 状态再单独读取内容，且统一以urlfetch.Error捕获底层网络错误，更适合在 App Engine 上精细控制错误处理。代码中对该响应非 200 的情况回写同等的状态码，保持了对外语义一致。

在 snippets/main.py 中，UrlLibFetchHandler挂载在/，UrlFetchHandler挂载在/url_fetch：

app = webapp2.WSGIApplication( [ ("/", UrlLibFetchHandler), ("/url_fetch", UrlFetchHandler), ("/url_post", UrlPostHandler), ("/submit_form", SubmitHandler), ], debug=True, )

方式二：通过urlfetch提交 POST 表单

同一文件中的UrlPostHandler演示了以application/x-www-form-urlencoded格式提交表单：

form_fields = { "first_name": "Albert", "last_name": "Johnson", } form_data = urllib.urlencode(UrlPostHandler.form_fields) headers = {"Content-Type": "application/x-www-form-urlencoded"} result = urlfetch.fetch( url="http://localhost:8080/submit_form", payload=form_data, method=urlfetch.POST, headers=headers, )

关键参数：

• payload：请求体，必须先经urllib.urlencode编码为 URL 编码格式；
• method：显式指定为urlfetch.POST（默认是GET）；
• headers：设置Content-Type，服务端据此解析表单字段。

配套的SubmitHandler演示了接收端如何取回字段：self.request.get("first_name")。整个链路形成一个完整的"发送端 → 接收端"闭环，可直接替换目标 URL 用于对接你自己的表单接口。

方式三：异步 RPC 调用——create_rpc+make_fetch_call

async/rpc.py 演示了 urlfetch 的异步能力。核心思路是：先创建一个 RPC 对象，再发起不阻塞的抓取调用，在"等待结果期间"做其他事情，最后统一收口。

rpc = urlfetch.create_rpc() urlfetch.make_fetch_call(rpc, "http://www.google.com/") # ... do other things ... try: result = rpc.get_result() if result.status_code == 200: text = result.content self.response.write(text) else: self.response.status_int = result.status_code self.response.write( "URL returned status code {}".format(result.status_code) ) except urlfetch.DownloadError: self.response.status_int = 500 self.response.write("Error fetching URL")

执行流程可拆解为：

1. create_rpc()创建未启动的 RPC 句柄；
2. make_fetch_call(rpc, url)发起抓取，立刻返回而不阻塞当前请求线程；
3. 业务代码可以继续执行其他逻辑（源码注释中的# ... do other things ...即示意点）；
4. rpc.get_result()在结果就绪后取得响应；若下载失败则抛出urlfetch.DownloadError，示例中以 500 状态码响应。

方式四：批量异步请求与回调处理

同文件中的UrlFetchRpcCallbackHandler进一步展示了"一次发起多个异步请求 + 回调处理"的进阶用法：

def handle_result(rpc): result = rpc.get_result() self.response.write(result.content) logging.info("Handling RPC in callback: result {}".format(result)) urls = [ "http://www.google.com", "http://www.github.com", "http://www.travis-ci.org", ] rpcs = [] for url in urls: rpc = urlfetch.create_rpc() rpc.callback = functools.partial(handle_result, rpc) urlfetch.make_fetch_call(rpc, url) rpcs.append(rpc) # ... do other things ... for rpc in rpcs: rpc.wait()

要点：

• 为每个 URL 单独创建 RPC，并通过rpc.callback绑定处理函数；由于循环变量会在回调执行时变化，示例使用functools.partial(handle_result, rpc)提前把各自的 RPC 对象绑定进去，避免闭包共享问题；
• 所有请求同时发出，互不阻塞，这正是threadsafe: yes环境下提升吞吐的关键手段；
• 最后对每个 RPC 调用rpc.wait()，保证所有回调完成后再结束请求处理，日志记录"Done waiting for RPCs"。

该应用的 WSGI 路由见 async/rpc.py：/对应基础异步示例，/callback对应回调示例。

方式五：用requests库透明地走 urlfetch

requests/main.py 展示了在 App Engine 上使用第三方 HTTP 库requests的最优雅方式——借助requests_toolbelt的 App Engine 适配器，让 requests 的全部调用自动改为经由 urlfetch：

import requests import requests_toolbelt.adapters.appengine # Use the App Engine Requests adapter. # This makes sure that Requests uses URLFetch. requests_toolbelt.adapters.appengine.monkeypatch() app = Flask(__name__) @app.route("/") def index(): url = "http://www.google.com/humans.txt" response = requests.get(url) response.raise_for_status() return response.text

这段代码的价值在于：

• 业务代码无需感知底层是 urlfetch 还是普通 socket，requests.get(url)写法与本地开发完全一致；
• monkeypatch()在导入后立即执行，一次性把 requests 的默认 HTTP 适配器替换为 App Engine 版，适用于模块内所有后续请求；
• 通过response.raise_for_status()可以把非 2xx 状态转成异常，配合文件末尾的@app.errorhandler(500)记录堆栈日志，形成完整的错误处理闭环。

由于该子应用引入了requests_toolbelt这类第三方库，需要在 appengine_config.py 中通过vendor.add("lib")把安装到lib目录的依赖加载进沙箱路径，这是 Python 2.7 标准环境引入第三方库的标准做法。

测试与验证：仓库提供的可运行断言

三个子应用都配有 pytest 测试，可直接验证上述行为，也为我们理解底层调用链提供了依据。

snippets/main_test.py 使用webtest.TestApp包裹 WSGI 应用，断言：

• GET /（urllib2 路径）与GET /url_fetch（urlfetch 路径）的响应体均包含"Google"；
• 通过mock.patch("main.urlfetch")模拟fetch返回content="Albert"，从而断言 POST 表单流程正确输出字段值，验证了payload/headers/method参数按预期传递。

async/rpc_test.py 覆盖了更多分支：

• mockcreate_rpc/get_result返回 200 时响应体含"I'm Feeling Lucky"；
• 抛出urlfetch.DownloadError时返回 500 且响应"Error fetching URL"；
• 返回 404 状态码时透传 404；
• 对/callback路由断言rpc_mock.wait.assert_called_with()，确认回调模式下每个 RPC 都执行了wait()。

requests/main_test.py 则先用 testbed 激活 App Engine stub 再导入main，随后通过 Flask 测试客户端断言根路由返回 200 且包含"Google"——注意它的注释特别强调：必须"先激活 testbed 再 import main"，否则 urlfetch stub 不可用，这正说明了该请求在测试环境下依然走的是被 monkeypatch 后的 urlfetch 路径。

小结：如何选择适合自己的请求方式

综合 README 指向的三个示例，选择建议如下：

• 单一简单请求：优先用urlfetch.fetch同步调用，代码直观、错误语义清晰（snippets）；
• 需要并发抓取多个 URL：使用create_rpc+make_fetch_call，配合回调与wait()收口（async）；
• 已有基于 requests 的代码库：用requests_toolbelt.adapters.appengine.monkeypatch()一行接入，迁移成本最低（requests）。

需要说明的是，本文所涉示例全部基于 Python 2.7 的 App Engine 标准环境 API，适用于该历史运行时的存量项目迁移与学习参考；若在较新的运行环境中开发，应参照对应版本的官方网络库方案。阅读本文后，你可以直接以 appengine/standard/urlfetch 的三个子目录为模板，替换其中的目标 URL 与处理逻辑，快速搭建自己的外部请求处理服务。

• 示例工程