Python 的函数框架

用于编写可移植 Python 函数的开源 FaaS（函数即服务）框架——由 Google Cloud Functions 团队提供给您。

Functions Framework 允许您编写在许多不同环境中运行的轻量级函数，包括：

• 谷歌云函数
• 您的本地开发机器
• Cloud Run 和 Cloud Run for Anthos
• 基于Knative的环境

该框架允许您从：

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

至：

curl http://my-url
# Output: Hello world!

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

特征

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

安装

通过以下方式安装 Functions Framework pip：

pip install functions-framework

或者，对于部署，将 Functions Framework 添加到您的requirements.txt文件中：

functions-framework==3.*

快速入门

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

创建一个main.py包含以下内容的文件：

导入 函数_框架

@functions_framework 。http 
def hello(request):
    return "Hello world!"

你的函数被传递了一个参数(request)，它是一个 FlaskRequest对象。

运行以下命令：

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
# Output: Hello world!

快速入门：CloudEvent 函数

创建一个main.py包含以下内容的文件：

import functions_framework

@functions_framework.cloud_event
def hello_cloud_event(cloud_event):
   print(f"Received event with ID: {cloud_event['id']} and data {cloud_event.data}")

您的函数被传递了一个CloudEvent参数。

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

functions-framework --target=hello_cloud_event

在另一个终端中，curlFunctions Framework 服务器：

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：

Received event with ID: A234-1234-1234 and data 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("Received", context.event_id)
   

2. 在端口 8080 上启动 Functions Framework：

   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.
   

   在运行 Functions Framework 的终端中：

    * 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
   Received 1
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   Received 2
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   Received 5
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   Received 6
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   Received 7
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   Received 8
   127.0.0.1 - - [11/Aug/2021 14:42:22] "POST / HTTP/1.1" 200 -
   Received 9
   127.0.0.1 - - [11/Aug/2021 14:42:39] "POST / HTTP/1.1" 200 -
   Received 3
   127.0.0.1 - - [11/Aug/2021 14:42:39] "POST / HTTP/1.1" 200 -
   Received 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. 使用 Functions 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
    # Output: Serving function...
   

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

    curl localhost:8080
    # Output: Hello World!
   

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

谷歌云函数

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

在 Cloud Functions 上，无需使用 Functions Framework：您无需将其添加到requirements.txt文件中。

编写函数后，您可以使用gcloud命令行工具从本地机器上简单地部署它。查看 Cloud Functions 快速入门。

Cloud Run/GKE 上的 Cloud Run

编写函数并将函数框架添加到requirements.txt文件后，剩下的就是创建容器映像。查看适用于 Python 的 Cloud Run 快速入门，以创建容器映像并将其部署到 Cloud Run。Dockerfile当你构建你的容器时，你会写一个。这Dockerfile允许您准确指定进入容器的内容（包括自定义二进制文件、特定操作系统等）。这是一个Dockerfile调用函数框架的示例。

如果您想进一步控制环境，可以将容器映像部署到 GKE 上的 Cloud Run。借助 Cloud Run on GKE，您可以在 GKE 集群上运行您的函数，这使您可以更好地控制环境（包括使用基于 GPU 的实例、更长的超时时间等等）。

基于 Knative 的容器环境

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

配置函数框架

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

启用 Google Cloud 函数事件

Functions Framework 可以将传入的 Google Cloud Functions事件负载解组到event对象context。当它收到请求时，这些将作为参数传递给您的函数。请注意，您的函数必须使用event-style 函数签名：

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

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

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

请参阅运行示例。

高级示例

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

贡献

欢迎和鼓励对该库的贡献。有关如何开始的更多信息，请参阅CONTRIBUTING 。