Python函数框架

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

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

• Google Cloud Functions
• 您的本地开发机器
• Cloud Run和Cloud Run for Anthos
• Knative环境

该框架允许您从

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

过渡到

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

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

功能

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

安装

通过pip安装Functions框架

pip install functions-framework

或者，为了部署，将Functions框架添加到您的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
# Output: 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"Received event with ID: {cloud_event['id']} and data {cloud_event.data}")

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

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

functions-framework --target=hello_cloud_event

在另一个终端中，使用curl调用Functions 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框架

   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

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

在Cloud Functions上，使用Functions框架不是必需的：您无需将其添加到您的requirements.txt文件中。

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

Cloud Run/Cloud Run on GKE

一旦你编写了你的函数并将函数框架添加到你的requirements.txt文件中，剩下的就是创建一个容器镜像。请查看Python的Cloud Run快速入门教程[创建容器镜像并将其部署到Cloud Run](https://cloud.google.com/run/docs/quickstarts/build-and-deploy)，以创建容器镜像并将其部署到Cloud Run。构建容器时，你需要编写一个Dockerfile。这个Dockerfile允许你指定容器中包含的确切内容（包括自定义的二进制文件、特定的操作系统等）。[这里是一个调用函数框架的示例Dockerfile](https://github.com/GoogleCloudPlatform/functions-framework-python/blob/main/examples/cloud_run_http)

如果你想对环境有更多的控制，你可以将容器镜像部署到GKE上的Cloud Run[部署你的容器镜像到GKE上的Cloud Run](https://cloud.google.com/run/docs/quickstarts/prebuilt-deploy-gke)。使用GKE上的Cloud Run，你可以在GKE集群上运行你的函数，这为你提供了对环境的额外控制（包括使用基于GPU的实例、更长的超时时间等）。

基于Knative的容器环境

Cloud Run和GKE上的Cloud Run都实现了Knative Serving API。函数框架设计为与Knative环境兼容。只需构建并部署你的容器到Knative环境即可。

配置函数框架

你可以使用命令行标志或环境变量来配置函数框架。如果你两者都指定了，环境变量将被忽略。

启用Google Cloud Function事件

函数框架可以将传入的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。