本頁面由 Cloud Translation API 翻譯而成。

Python 3 的延遲 API

本頁面說明如何使用延遲 API (舊版套裝服務之一)，搭配標準環境的 Python 3 執行階段。您的應用程式可以透過 Python 3 專用的 App Engine 服務 SDK存取套裝組合服務。

總覽

先前，延遲包裹 google.appengine.ext.deferred 會依賴 Python 2 中的 webapp 架構。由於 Python 3 專用的 App Engine 服務 SDK 已移除 webapp 架構，因此您需要在將 Python 2 應用程式升級至 Python 3 時進行一些變更。

啟用延遲 API

如要為 Python 3 啟用延遲 API，您不再需要在 app.yaml 檔案中將 builtins.deferred 設為 on。相反地，如要啟用 API，您必須在呼叫 wrap_wsgi_app() 時傳遞 use_deferred=True。

相似之處和差異

根據預設，Python 3 的延遲 API 會使用與 Python 2 相同的網址 /_ah/queue/deferred 和預設佇列。請注意，對於遷移至 Cloud Tasks 的應用程式，系統不會自動建立預設佇列，且延後的任務程式庫不可用。

如果應用程式使用預設的 /_ah/queue/deferred 端點，在 Python 3 中使用 deferred.defer() 的做法與在 Python 2 中相同。如果您的應用程式使用自訂網址執行延後的作業，您需要進行一些變更，因為 Python 3 版的這個 API 已移除 Python 2 版 deferred 模組中的 TaskHandler 類別。

如要為執行延遲工作設定自訂網址，應用程式可以覆寫 deferred.Handler 類別 (舊稱 Python 2 中的 deferred.TaskHandler) 中的 post 或 run_from_request 方法，並傳遞 environ 參數，該參數代表包含 WSGI 要求參數的字典。接著，您可以從自訂端點呼叫 post 方法 (如 Python 3 範例所示)。

Python 3 延遲 API 的端對端用法 (例如要求路由和存取 environ 字典) 取決於應用程式要遷移至的網路架構。請比較下列各節中 Python 2 範例與 Python 3 範例的程式碼變更。

Python 3 範例

以下範例說明如何在 Flask 應用程式和 Django 應用程式中，使用預設端點和自訂端點執行延遲工作。

Flask

import os

from flask import Flask, request
from google.appengine.api import wrap_wsgi_app
from google.appengine.ext import deferred
from google.appengine.ext import ndb

my_key = os.environ.get("GAE_VERSION", "Missing")

app = Flask(__name__)
app.wsgi_app = wrap_wsgi_app(app.wsgi_app, use_deferred=True)


class Counter(ndb.Model):
    count = ndb.IntegerProperty(indexed=False)


def do_something_later(key, amount):
    entity = Counter.get_or_insert(key, count=0)
    entity.count += amount
    entity.put()


@app.route("/counter/increment")
def increment_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 arguments
    deferred.defer(do_something_later, my_key, 10, _url="/custom/path", _countdown=120)

    return "Deferred counter increment."


@app.route("/counter/get")
def view_counter():
    counter = Counter.get_or_insert(my_key, count=0)
    return str(counter.count)


@app.route("/custom/path", methods=["POST"])
def custom_deferred():
    print("Executing deferred task.")
    # request.environ contains the WSGI `environ` dictionary (See PEP 0333)
    return deferred.Handler().post(request.environ)

Django

import os

from django.conf import settings
from django.core.wsgi import get_wsgi_application
from django.http import HttpResponse
from django.urls import path
from google.appengine.api import wrap_wsgi_app
from google.appengine.ext import deferred
from google.appengine.ext import ndb

my_key = os.environ.get("GAE_VERSION", "Missing")


class Counter(ndb.Model):
    count = ndb.IntegerProperty(indexed=False)


def do_something_later(key, amount):
    entity = Counter.get_or_insert(key, count=0)
    entity.count += amount
    entity.put()


def increment_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 arguments
    deferred.defer(do_something_later, my_key, 10, _url="/custom/path", _countdown=120)

    return HttpResponse("Deferred counter increment.")


def view_counter(request):
    counter = Counter.get_or_insert(my_key, count=0)
    return HttpResponse(str(counter.count))


def custom_deferred(request):
    print("Executing deferred task.")
    # request.environ contains the WSGI `environ` dictionary (See PEP 0333)
    response, status, headers = deferred.Handler().post(request.environ)
    return HttpResponse(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)

不使用任何架構

import os
import re

from google.appengine.api import wrap_wsgi_app
from google.appengine.ext import deferred
from google.appengine.ext import ndb

my_key = os.environ.get("GAE_VERSION", "Missing")


class Counter(ndb.Model):
    count = ndb.IntegerProperty(indexed=False)


def do_something_later(key, amount):
    entity = Counter.get_or_insert(key, count=0)
    entity.count += amount
    entity.put()


def IncrementCounter(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 arguments
    deferred.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."]


def ViewCounter(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")]


class CustomDeferredHandler(deferred.Handler):
    """Deferred task handler that adds additional logic."""

    def post(self, environ):
        print("Executing deferred task.")
        return super().post(environ)


routes = {
    "counter/increment": IncrementCounter,
    "counter/get": ViewCounter,
    "custom/path": CustomDeferredHandler(),
}


class WSGIApplication:
    def __call__(self, environ, start_response):
        path = environ.get("PATH_INFO", "").lstrip("/")
        for regex, handler in routes.items():
            match = re.search(regex, path)
            if match is not None:
                return handler(environ, start_response)

        start_response("404 Not Found", [("Content-Type", "text/plain")])
        return [b"Not found"]


app = wrap_wsgi_app(WSGIApplication(), use_deferred=True)

程式碼範例

如要查看本指南的完整程式碼範例，請前往 GitHub。