Python 函数框架

[![Python 单元测试 CI][ff_python_unit_img]][ff_python_unit_link] [![Python 代码风格检查 CI][ff_python_lint_img]][ff_python_lint_link] [![Python 一致性测试 CI][ff_python_conformance_img]][ff_python_conformance_link]

这是一个开源的 FaaS（函数即服务）框架，用于编写可移植的 Python 函数 —— 由 Google Cloud Functions 团队为您带来。

函数框架让您可以编写轻量级函数，这些函数可以在多种不同的环境中运行，包括：

• Google Cloud Functions
• 您的本地开发机器
• Cloud Run 和适用于 Anthos 的 Cloud Run
• 基于 Knative 的环境

该框架允许您从：

def hello(request):
    return "Hello world!"

转变为：

curl http://my-url
# 输出：Hello world!

无需担心编写 HTTP 服务器或复杂的请求处理逻辑。

特性

• 快速启动本地开发服务器进行测试
• 响应请求调用函数
• 自动解析符合 CloudEvents 规范的事件
• 在无服务器平台之间可移植

安装

通过 pip 安装函数框架：

pip install functions-framework

或者，对于部署，将函数框架添加到您的 requirements.txt 文件中：

functions-framework==3.*

快速入门

快速入门：HTTP 函数（Hello World）

创建一个 main.py 文件，内容如下：

import flask
import functions_framework

@functions_framework.http
def hello(request: flask.Request) -> flask.typing.ResponseReturnValue:
    return "Hello world!"

您的函数接收一个参数 (request)，它是一个 Flask Request 对象。

运行以下命令：

functions-framework --target hello --debug
 * Serving Flask app "hello" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://0.0.0.0:8080/ (Press CTRL+C to quit)

（如果您安装了多个语言框架，也可以使用 functions-framework-python）。

在浏览器中打开 http://localhost:8080/ 并看到 Hello world!。

或者使用另一个终端窗口中的 curl 向此函数发送请求：

curl localhost:8080
# 输出：Hello world!

快速入门：CloudEvent 函数

创建一个 main.py 文件，内容如下：

import functions_framework
from cloudevents.http.event import CloudEvent

@functions_framework.cloud_event
def hello_cloud_event(cloud_event: CloudEvent) -> None:
   print(f"收到 ID 为 {cloud_event['id']} 的事件，数据为 {cloud_event.data}")

您的函数接收一个 CloudEvent 参数。

运行以下命令在本地运行 hello_cloud_event 目标：

functions-framework --target=hello_cloud_event

在另一个终端中，使用 curl 向函数框架服务器发送请求：

curl -X POST localhost:8080 \
   -H "Content-Type: application/cloudevents+json" \
   -d '{
	"specversion" : "1.0",
	"type" : "example.com.cloud.event",
	"source" : "https://example.com/cloudevents/pull",
	"subject" : "123",
	"id" : "A234-1234-1234",
	"time" : "2018-04-05T17:31:00Z",
	"data" : "hello world"
}'

运行 functions-framework 的终端输出：

收到 ID 为 A234-1234-1234 的事件，数据为 hello world

有关发送 CloudEvents 负载的更多信息，请参阅 examples/cloud_run_cloud_events 说明。

快速入门：错误处理

该框架包含一个错误处理程序，类似于 flask.Flask.errorhandler 函数，允许您使用装饰器处理特定的错误类型：

import functions_framework


@functions_framework.errorhandler(ZeroDivisionError)
def handle_zero_division(e):
    return "I'm a teapot", 418


def function(request):
    1 / 0
    return "Success", 200

这个函数将捕获 ZeroDivisionError 并返回不同的响应。

快速入门：Pub/Sub 模拟器

1. 创建一个 main.py 文件，内容如下：

   def hello(event, context):
        print("收到", context.event_id)

2. 在 8080 端口启动函数框架：

   functions-framework --target=hello --signature-type=event --debug --port=8080

3. 在第二个终端中，在 8085 端口启动 Pub/Sub 模拟器。

   export PUBSUB_PROJECT_ID=my-project
   gcloud beta emulators pubsub start \
       --project=$PUBSUB_PROJECT_ID \
       --host-port=localhost:8085

   Pub/Sub 模拟器成功启动后，您应该看到以下内容：

   [pubsub] INFO: Server started, listening on 8085

4. 在第三个终端中，创建一个 Pub/Sub 主题并为该主题附加一个推送订阅，使用 http://localhost:8080 作为其推送端点。发布一些消息到该主题。观察您的函数被 Pub/Sub 消息触发。

   export PUBSUB_PROJECT_ID=my-project
   export TOPIC_ID=my-topic
   export PUSH_SUBSCRIPTION_ID=my-subscription
   $(gcloud beta emulators pubsub env-init)
   
   git clone https://github.com/googleapis/python-pubsub.git
   cd python-pubsub/samples/snippets/
   pip install -r requirements.txt
   
   python publisher.py $PUBSUB_PROJECT_ID create $TOPIC_ID
   python subscriber.py $PUBSUB_PROJECT_ID create-push $TOPIC_ID $PUSH_SUBSCRIPTION_ID http://localhost:8080
   python publisher.py $PUBSUB_PROJECT_ID publish $TOPIC_ID

   命令成功运行后，您应该看到以下内容：

   Created topic: projects/my-project/topics/my-topic
   
   topic: "projects/my-project/topics/my-topic"
   push_config {
     push_endpoint: "http://localhost:8080"
   }
   ack_deadline_seconds: 10
   message_retention_duration {
     seconds: 604800
   }
   .
   Endpoint for subscription is: http://localhost:8080
   
   1
   2
   3
   4
   5
   6
   7
   8
   9
   Published messages to projects/my-project/topics/my-topic.

   在运行函数框架的终端中：

    * Serving Flask app "hello" (lazy loading)
    * Environment: production
      WARNING: This is a development server. Do not use it in a production deployment.
      Use a production WSGI server instead.
    * Debug mode: on
    * Running on http://0.0.0.0:8080/ (Press CTRL+C to quit)
    * Restarting with fsevents reloader
    * Debugger is active!
    * Debugger PIN: 911-794-046
   收到 1
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   收到 2
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   收到 5
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   收到 6
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   收到 7
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   收到 8
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   收到 9
   127.0.0.1 - - [11/Aug/2021 14:42:39] "POST / HTTP/1.1" 200 -
   收到 3
   127.0.0.1 - - [11/Aug/2021 14:42:39] "POST / HTTP/1.1" 200 -
   收到 4
   127.0.0.1 - - [11/Aug/2021 14:42:39] "POST / HTTP/1.1" 200 -

有关从 Pub/Sub 事件中提取数据的更多详细信息，请参阅 https://cloud.google.com/functions/docs/tutorials/pubsub#functions_helloworld_pubsub_tutorial-python

快速入门：构建可部署的容器

1. 安装 Docker 和 pack 工具。

2. 使用函数 buildpacks 从您的函数构建容器：

    pack build \
        --builder gcr.io/buildpacks/builder:v1 \
        --env GOOGLE_FUNCTION_SIGNATURE_TYPE=http \
        --env GOOGLE_FUNCTION_TARGET=hello \
        my-first-function
   

3. 启动构建的容器：

    docker run --rm -p 8080:8080 my-first-function
    # 输出：Serving function...
   

4. 在另一个终端窗口中使用 curl 向此函数发送请求：

    curl localhost:8080
    # 输出：Hello World!
   

在无服务器平台上运行您的函数

Google Cloud Functions

此函数框架基于 Google Cloud Functions 上的 Python 运行时。

在 Cloud Functions 上，无需使用函数框架：您不需要将其添加到 requirements.txt 文件中。 编写完函数后，您可以使用gcloud命令行工具直接从本地机器部署它。查看Cloud Functions快速入门。

Cloud Run/在GKE上运行Cloud Run

编写完函数并将Functions Framework添加到requirements.txt文件后，剩下的就是创建一个容器镜像。查看Python的Cloud Run快速入门以创建容器镜像并将其部署到Cloud Run。构建容器时，您需要编写一个Dockerfile。这个Dockerfile允许您精确指定容器中包含的内容（包括自定义二进制文件、特定操作系统等）。这里是一个调用Functions Framework的Dockerfile示例。

如果您想对环境有更多控制，可以将容器镜像部署到GKE上的Cloud Run。使用GKE上的Cloud Run，您可以在GKE集群上运行函数，这gives您对环境的额外控制（包括使用基于GPU的实例、更长的超时时间等）。

基于Knative的容器环境

Cloud Run和GKE上的Cloud Run都实现了Knative Serving API。Functions Framework设计为与Knative环境兼容。只需构建您的容器并将其部署到Knative环境即可。

配置Functions Framework

您可以使用命令行标志或环境变量配置Functions Framework。如果同时指定两者，环境变量将被忽略。

命令行标志	环境变量	描述
--host	HOST	Functions Framework监听请求的主机。默认值：0.0.0.0
--port	PORT	Functions Framework监听请求的端口。默认值：8080
--target	FUNCTION_TARGET	响应请求时要调用的导出函数的名称。默认值：function
--signature-type	FUNCTION_SIGNATURE_TYPE	编写函数时使用的签名。控制反序列化规则并确定用于调用函数的参数。默认值：http；接受的值：http、event或cloudevent
--source	FUNCTION_SOURCE	包含函数的文件路径。默认值：main.py（在当前工作目录中）
--debug	DEBUG	允许functions-framework在调试模式下运行的标志，包括实时重载。默认值：False

启用Google Cloud Function事件

Functions Framework可以将传入的Google Cloud Functions事件负载反序列化为event和context对象。当函数收到请求时，这些对象将作为参数传递给您的函数。请注意，您的函数必须使用event风格的函数签名：

def hello(event, context):
    print(event)
    print(context)

要启用自动反序列化，请使用--signature-type命令行标志或FUNCTION_SIGNATURE_TYPE环境变量将函数签名类型设置为event。默认情况下，将使用HTTP签名，并禁用自动事件反序列化。

有关此签名类型的更多详细信息，请参阅Google Cloud Functions文档中关于后台函数的部分。

查看运行示例。

高级示例

更高级的指南可以在examples/目录中找到。您还可以在这里找到使用CloudEvent Python SDK的示例。

贡献

欢迎并鼓励对这个库进行贡献。有关如何开始的更多信息，请参阅CONTRIBUTING。