이전에는 지연된 패키지 google.appengine.ext.deferred가 Python 2의 웹 앱 프레임워크에 의존했습니다. 웹 앱 프레임워크가 Python 3용 App Engine 서비스 SDK에서 삭제되었으므로 Python 2 앱을 Python 3으로 업그레이드할 때 몇 가지 항목을 변경해야 합니다.
Deferred API 사용 설정
Python 3용 Deferred API를 사용 설정하기 위해 더 이상 app.yaml 파일에서 builtins.deferred를 on으로 설정할 필요가 없습니다. 대신 API를 사용 설정하려면 호출에서 use_deferred=True를 wrap_wsgi_app()으로 전달해야 합니다.
앱에서 기본 /_ah/queue/deferred 엔드포인트를 사용하는 경우 Python 3에서 deferred.defer()를 사용하면 Python 2와 동일하게 유지됩니다.
앱에서 지연된 태스크 실행에 커스텀 URL을 사용하는 경우 Python 2용 deferred 모듈의 TaskHandler 클래스가 이 API의 Python 3 버전에서 삭제되었으므로 몇 가지 항목을 변경해야 합니다.
지연된 태스크를 실행할 수 있도록 커스텀 URL을 설정하려면 앱이 deferred.Handler 클래스(Python 2에서는 deferred.TaskHandler)에서 post 또는 run_from_request 메서드를 재정의하고 WSGI 요청 파라미터가 포함된 사전을 나타내는 environ 파라미터를 전달하면 됩니다. 그런 다음 Python 3 샘플과 같이 커스텀 엔드포인트에서 post 메서드를 호출할 수 있습니다.
요청 라우팅 및 environ 사전 액세스와 같은 Python 3 Deferred API의 엔드 투 엔드 사용량은 앱에서 마이그레이션하는 웹 프레임워크에 따라 달라집니다. 다음 섹션에서 Python 2 예시와 Python 3 예시의 코드 변경사항을 비교합니다.
Python 3 예시
다음 예시에서는 Flask 앱 및 Django 앱에서 기본 엔드포인트와 커스텀 엔드포인트를 사용하여 지연된 태스크를 실행하는 방법을 보여줍니다.
Flask
importosfromflaskimportFlask,requestfromgoogle.appengine.apiimportwrap_wsgi_appfromgoogle.appengine.extimportdeferredfromgoogle.appengine.extimportndbmy_key=os.environ.get("GAE_VERSION","Missing")app=Flask(__name__)app.wsgi_app=wrap_wsgi_app(app.wsgi_app,use_deferred=True)classCounter(ndb.Model):count=ndb.IntegerProperty(indexed=False)defdo_something_later(key,amount):entity=Counter.get_or_insert(key,count=0)entity.count+=amountentity.put()@app.route("/counter/increment")defincrement_counter():# Use default URL and queue name, no task name, execute ASAP.deferred.defer(do_something_later,my_key,10)# Use default URL and queue name, no task name, execute after 1 minute.deferred.defer(do_something_later,my_key,10,_countdown=60)# Providing non-default task queue argumentsdeferred.defer(do_something_later,my_key,10,_url="/custom/path",_countdown=120)return"Deferred counter increment."@app.route("/counter/get")defview_counter():counter=Counter.get_or_insert(my_key,count=0)returnstr(counter.count)@app.route("/custom/path",methods=["POST"])defcustom_deferred():print("Executing deferred task.")# request.environ contains the WSGI `environ` dictionary (See PEP 0333)returndeferred.Handler().post(request.environ)
Django
importosfromdjango.confimportsettingsfromdjango.core.wsgiimportget_wsgi_applicationfromdjango.httpimportHttpResponsefromdjango.urlsimportpathfromgoogle.appengine.apiimportwrap_wsgi_appfromgoogle.appengine.extimportdeferredfromgoogle.appengine.extimportndbmy_key=os.environ.get("GAE_VERSION","Missing")classCounter(ndb.Model):count=ndb.IntegerProperty(indexed=False)defdo_something_later(key,amount):entity=Counter.get_or_insert(key,count=0)entity.count+=amountentity.put()defincrement_counter(request):# Use default URL and queue name, no task name, execute ASAP.deferred.defer(do_something_later,my_key,10)# Use default URL and queue name, no task name, execute after 1 minute.deferred.defer(do_something_later,my_key,10,_countdown=60)# Providing non-default task queue argumentsdeferred.defer(do_something_later,my_key,10,_url="/custom/path",_countdown=120)returnHttpResponse("Deferred counter increment.")defview_counter(request):counter=Counter.get_or_insert(my_key,count=0)returnHttpResponse(str(counter.count))defcustom_deferred(request):print("Executing deferred task.")# request.environ contains the WSGI `environ` dictionary (See PEP 0333)response,status,headers=deferred.Handler().post(request.environ)returnHttpResponse(response,status=status.value)urlpatterns=(path("counter/get",view_counter,name="view_counter"),path("counter/increment",increment_counter,name="increment_counter"),path("custom/path",custom_deferred,name="custom_deferred"),)settings.configure(DEBUG=True,SECRET_KEY="thisisthesecretkey",ROOT_URLCONF=__name__,MIDDLEWARE_CLASSES=("django.middleware.common.CommonMiddleware","django.middleware.csrf.CsrfViewMiddleware","django.middleware.clickjacking.XFrameOptionsMiddleware",),ALLOWED_HOSTS=["*"],)app=wrap_wsgi_app(get_wsgi_application(),use_deferred=True)
프레임워크 제외
importosimportrefromgoogle.appengine.apiimportwrap_wsgi_appfromgoogle.appengine.extimportdeferredfromgoogle.appengine.extimportndbmy_key=os.environ.get("GAE_VERSION","Missing")classCounter(ndb.Model):count=ndb.IntegerProperty(indexed=False)defdo_something_later(key,amount):entity=Counter.get_or_insert(key,count=0)entity.count+=amountentity.put()defIncrementCounter(environ,start_response):# Use default URL and queue name, no task name, execute ASAP.deferred.defer(do_something_later,my_key,10)# Use default URL and queue name, no task name, execute after 1 minute.deferred.defer(do_something_later,my_key,10,_countdown=60)# Providing non-default task queue argumentsdeferred.defer(do_something_later,my_key,10,_url="/custom/path",_countdown=120)start_response("200 OK",[("Content-Type","text/html")])return[b"Deferred counter increment."]defViewCounter(environ,start_response):counter=Counter.get_or_insert(my_key,count=0)start_response("200 OK",[("Content-Type","text/html")])return[str(counter.count).encode("utf-8")]classCustomDeferredHandler(deferred.Handler):"""Deferred task handler that adds additional logic."""defpost(self,environ):print("Executing deferred task.")returnsuper().post(environ)routes={"counter/increment":IncrementCounter,"counter/get":ViewCounter,"custom/path":CustomDeferredHandler(),}classWSGIApplication:def__call__(self,environ,start_response):path=environ.get("PATH_INFO","").lstrip("/")forregex,handlerinroutes.items():match=re.search(regex,path)ifmatchisnotNone:returnhandler(environ,start_response)start_response("404 Not Found",[("Content-Type","text/plain")])return[b"Not found"]app=wrap_wsgi_app(WSGIApplication(),use_deferred=True)
[[["이해하기 쉬움","easyToUnderstand","thumb-up"],["문제가 해결됨","solvedMyProblem","thumb-up"],["기타","otherUp","thumb-up"]],[["이해하기 어려움","hardToUnderstand","thumb-down"],["잘못된 정보 또는 샘플 코드","incorrectInformationOrSampleCode","thumb-down"],["필요한 정보/샘플이 없음","missingTheInformationSamplesINeed","thumb-down"],["번역 문제","translationIssue","thumb-down"],["기타","otherDown","thumb-down"]],["최종 업데이트: 2025-09-04(UTC)"],[[["\u003cp\u003eThe Deferred API for Python 3 on Google App Engine no longer requires setting \u003ccode\u003ebuiltins.deferred\u003c/code\u003e in \u003ccode\u003eapp.yaml\u003c/code\u003e; instead, you must use \u003ccode\u003euse_deferred=True\u003c/code\u003e when calling \u003ccode\u003ewrap_wsgi_app()\u003c/code\u003e.\u003c/p\u003e\n"],["\u003cp\u003eThe default behavior of the Deferred API in Python 3 remains consistent with Python 2, using the \u003ccode\u003e/_ah/queue/deferred\u003c/code\u003e URL and the default queue, but keep in mind that Cloud Tasks will differ.\u003c/p\u003e\n"],["\u003cp\u003eWhen using the local development server for testing, you need to set \u003ccode\u003eDEFERRED_USE_CROSS_COMPATIBLE_PICKLE_PROTOCOL: 'True'\u003c/code\u003e in your \u003ccode\u003eapp.yaml\u003c/code\u003e to make sure the Deferred API works.\u003c/p\u003e\n"],["\u003cp\u003eCustom URLs for deferred task execution in Python 3 require using the \u003ccode\u003edeferred.Handler\u003c/code\u003e class's \u003ccode\u003epost\u003c/code\u003e or \u003ccode\u003erun_from_request\u003c/code\u003e methods, as the \u003ccode\u003eTaskHandler\u003c/code\u003e from Python 2 is no longer available.\u003c/p\u003e\n"],["\u003cp\u003eThe way a Python 3 App uses the Deferred API, including its handling of requests and accessing the \u003ccode\u003eenviron\u003c/code\u003e dictionary, is based on the web framework that is being migrated to, as shown in the provided examples with Flask, Django and without a framework.\u003c/p\u003e\n"]]],[],null,["# Deferred API for Python 3\n\nThis page describes how to use the Deferred API, one of the legacy bundled services,\nwith the [Python 3 runtime](/appengine/docs/standard/python3) for\nthe standard environment. Your app can access the bundled services\nthrough the [**App Engine services SDK for Python 3**](https://github.com/GoogleCloudPlatform/appengine-python-standard).\n\nOverview\n--------\n\nPreviously, the Deferred package [`google.appengine.ext.deferred`](/appengine/docs/legacy/standard/python/refdocs/google.appengine.ext.deferred.deferred)\ndepended on the webapp framework in Python 2. Since the webapp framework has\nbeen removed in the App Engine services SDK for Python 3, you need to\nmake some changes when upgrading your Python 2 app to Python 3.\n\nEnabling the Deferred API\n-------------------------\n\nTo enable the Deferred API for Python 3, you no longer need to set\n[`builtins.deferred`](/appengine/docs/legacy/standard/python/config/appref#builtins)\nto `on` in the `app.yaml` file. Instead, to enable the API, you must pass\n`use_deferred=True` in the call to `wrap_wsgi_app()`.\n| **Important:** When using the [local development server](/appengine/docs/standard/python3/services/access#testing_and_deployment) to test Python 3 apps that use the [Deferred API](/appengine/docs/legacy/standard/python/refdocs/google.appengine.ext.deferred.deferred), you must set the following environment variable in your `app.yaml`: \n| `DEFERRED_USE_CROSS_COMPATIBLE_PICKLE_PROTOCOL: 'True'`\n\nSimilarites and differences\n---------------------------\n\nBy default, the Deferred API for Python 3 uses the same URL `/_ah/queue/deferred`\nand the same [default queue](/appengine/docs/standard/python/taskqueue/push/migrating-push-queues#creating_queues)\nas it did in Python 2. Note that for apps migrating to [Cloud Tasks](/tasks), the default queue is\n[not created automatically](/appengine/docs/legacy/standard/python/taskqueue/push/migrating-push-queues#creating_queues)\nand the [deferred tasks library is not available](/appengine/docs/legacy/standard/python/taskqueue/push/migrating-push-queues#features-not-available).\n\nIf your app uses the default `/_ah/queue/deferred` endpoint, using\n[`deferred.defer()` in Python 3](/appengine/docs/standard/python3/reference/services/bundled/google/appengine/ext/deferred/deferred)\nremains the same as\n[Python 2](/appengine/docs/legacy/standard/python/taskqueue/push/creating-tasks#using_the_instead_of_a_worker_service).\nIf your app uses a custom URL for execution of deferred tasks, you need to make\nsome changes since the `TaskHandler` class in the `deferred` module for Python 2\nhas been removed in the Python 3 version of this API.\n\nTo set a custom URL for execution of deferred tasks, the app can override either\nthe `post` or the `run_from_request` method in the\n[`deferred.Handler` class](/appengine/docs/standard/python3/reference/services/bundled/google/appengine/ext/deferred/Handler)\n(formerly `deferred.TaskHandler` in Python 2), and pass the `environ` parameter\nwhich represents a dictionary containing WSGI request parameters. The `post` method can then be\ncalled from the custom endpoint (as shown in the [Python 3 samples](#python-3-examples)).\n\nThe end-to-end usage of the Python 3 Deferred API, such as routing of requests and\naccessing the [`environ` dictionary](https://www.python.org/dev/peps/pep-0333/#id19),\ndepends on the web framework the app is migrating to. Compare code changes made\nfrom the Python 2 example to the Python 3 examples in the following sections.\n\nPython 3 examples\n-----------------\n\nThe following example shows how to execute a deferred task using a\ndefault endpoint and a custom endpoint in a Flask app and Django app. \n\n### Flask\n\n import os\n\n from flask import Flask, request\n from google.appengine.api import wrap_wsgi_app\n from google.appengine.ext import deferred\n from google.appengine.ext import ndb\n\n my_key = os.environ.get(\"GAE_VERSION\", \"Missing\")\n\n app = Flask(__name__)\n app.wsgi_app = wrap_wsgi_app(app.wsgi_app, use_deferred=True)\n\n\n class Counter(ndb.Model):\n count = ndb.IntegerProperty(indexed=False)\n\n\n def do_something_later(key, amount):\n entity = Counter.get_or_insert(key, count=0)\n entity.count += amount\n entity.put()\n\n\n @app.route(\"/counter/increment\")\n def increment_counter():\n # Use default URL and queue name, no task name, execute ASAP.\n deferred.defer(do_something_later, my_key, 10)\n\n # Use default URL and queue name, no task name, execute after 1 minute.\n deferred.defer(do_something_later, my_key, 10, _countdown=60)\n\n # Providing non-default task queue arguments\n deferred.defer(do_something_later, my_key, 10, _url=\"/custom/path\", _countdown=120)\n\n return \"Deferred counter increment.\"\n\n\n @app.route(\"/counter/get\")\n def view_counter():\n counter = Counter.get_or_insert(my_key, count=0)\n return str(counter.count)\n\n\n @app.route(\"/custom/path\", methods=[\"POST\"])\n def custom_deferred():\n print(\"Executing deferred task.\")\n # request.environ contains the WSGI `environ` dictionary (See PEP 0333)\n return deferred.Handler().post(request.environ)\n\n### Django\n\n import os\n\n from django.conf import settings\n from django.core.wsgi import get_wsgi_application\n from django.http import HttpResponse\n from django.urls import path\n from google.appengine.api import wrap_wsgi_app\n from google.appengine.ext import deferred\n from google.appengine.ext import ndb\n\n my_key = os.environ.get(\"GAE_VERSION\", \"Missing\")\n\n\n class Counter(ndb.Model):\n count = ndb.IntegerProperty(indexed=False)\n\n\n def do_something_later(key, amount):\n entity = Counter.get_or_insert(key, count=0)\n entity.count += amount\n entity.put()\n\n\n def increment_counter(request):\n # Use default URL and queue name, no task name, execute ASAP.\n deferred.defer(do_something_later, my_key, 10)\n\n # Use default URL and queue name, no task name, execute after 1 minute.\n deferred.defer(do_something_later, my_key, 10, _countdown=60)\n\n # Providing non-default task queue arguments\n deferred.defer(do_something_later, my_key, 10, _url=\"/custom/path\", _countdown=120)\n\n return HttpResponse(\"Deferred counter increment.\")\n\n\n def view_counter(request):\n counter = Counter.get_or_insert(my_key, count=0)\n return HttpResponse(str(counter.count))\n\n\n def custom_deferred(request):\n print(\"Executing deferred task.\")\n # request.environ contains the WSGI `environ` dictionary (See PEP 0333)\n response, status, headers = deferred.Handler().post(request.environ)\n return HttpResponse(response, status=status.value)\n\n\n urlpatterns = (\n path(\"counter/get\", view_counter, name=\"view_counter\"),\n path(\"counter/increment\", increment_counter, name=\"increment_counter\"),\n path(\"custom/path\", custom_deferred, name=\"custom_deferred\"),\n )\n\n settings.configure(\n DEBUG=True,\n SECRET_KEY=\"thisisthesecretkey\",\n ROOT_URLCONF=__name__,\n MIDDLEWARE_CLASSES=(\n \"django.middleware.common.CommonMiddleware\",\n \"django.middleware.csrf.CsrfViewMiddleware\",\n \"django.middleware.clickjacking.XFrameOptionsMiddleware\",\n ),\n ALLOWED_HOSTS=[\"*\"],\n )\n\n app = wrap_wsgi_app(get_wsgi_application(), use_deferred=True)\n\n### Without any framework\n\n import os\n import re\n\n from google.appengine.api import wrap_wsgi_app\n from google.appengine.ext import deferred\n from google.appengine.ext import ndb\n\n my_key = os.environ.get(\"GAE_VERSION\", \"Missing\")\n\n\n class Counter(ndb.Model):\n count = ndb.IntegerProperty(indexed=False)\n\n\n def do_something_later(key, amount):\n entity = Counter.get_or_insert(key, count=0)\n entity.count += amount\n entity.put()\n\n\n def IncrementCounter(environ, start_response):\n # Use default URL and queue name, no task name, execute ASAP.\n deferred.defer(do_something_later, my_key, 10)\n\n # Use default URL and queue name, no task name, execute after 1 minute.\n deferred.defer(do_something_later, my_key, 10, _countdown=60)\n\n # Providing non-default task queue arguments\n deferred.defer(do_something_later, my_key, 10, _url=\"/custom/path\", _countdown=120)\n\n start_response(\"200 OK\", [(\"Content-Type\", \"text/html\")])\n return [b\"Deferred counter increment.\"]\n\n\n def ViewCounter(environ, start_response):\n counter = Counter.get_or_insert(my_key, count=0)\n start_response(\"200 OK\", [(\"Content-Type\", \"text/html\")])\n return [str(counter.count).encode(\"utf-8\")]\n\n\n class CustomDeferredHandler(deferred.Handler):\n \"\"\"Deferred task handler that adds additional logic.\"\"\"\n\n def post(self, environ):\n print(\"Executing deferred task.\")\n return super().post(environ)\n\n\n routes = {\n \"counter/increment\": IncrementCounter,\n \"counter/get\": ViewCounter,\n \"custom/path\": CustomDeferredHandler(),\n }\n\n\n class WSGIApplication:\n def __call__(self, environ, start_response):\n path = environ.get(\"PATH_INFO\", \"\").lstrip(\"/\")\n for regex, handler in routes.items():\n match = re.search(regex, path)\n if match is not None:\n return handler(environ, start_response)\n\n start_response(\"404 Not Found\", [(\"Content-Type\", \"text/plain\")])\n return [b\"Not found\"]\n\n\n app = wrap_wsgi_app(WSGIApplication(), use_deferred=True)\n\nCode samples\n------------\n\nTo view the complete code samples from this guide, see\n[GitHub](https://github.com/GoogleCloudPlatform/python-docs-samples/tree/main/appengine/standard_python3/bundled-services)."]]