简体中文
Tencent Cloud GitHub

定时调用无服务器云函数

本文将向您介绍并演示借助 Tencent Cloud SDK for Python 如何实现本机定时调用无服务器云函数

概述

谈论到定时调用无服务器云函数时，或许我们首先想到的是利用无服务器云函数的定时触发器功能来实现。当我们需要在大量不确定个数的指定时间调用无服务器云函数时定时触发器将不能为我们提供便利，因为无服务器云函数的触发器个数受触发器配额限制。

通过借助 Tencent Cloud SDK for Python 的无服务器云函数定时调用机制，我们可以快速在本机运行环境中实现在在多个指定时间调用指定的无服务器云函数并获得运行结果，这也称之为按时间计划调用。

在下文中我们将通过一个简单的应用场景向您进一步介绍并演示如何利用 Tencent Cloud SDK for Python 实现对指定无服务器云函数的本机定时调用。

场景

设定

假定我们在数据中心 ap-shanghai 的命名空间 default 拥有一个已创建就绪的无服务器云函数 hello，其无服务器云函数的配置为：

要求

在该场景中我们要求无服务器云函数 hello 实现发起一个 HTTP 请求到 https://libget.com/hello 并获得响应正文（Body）作为无服务器云函数的返回值。

此外，在本地运行环境创建一个 Python 脚本并实现计划在本机时间的 10 秒后、20 秒后和 30 秒后调用无服务器云函数 hello 并获得无服务器云函数的实际返回值。

指导

在下文中，我们将通过几个步骤向您演示编写符合当前场景要求的代码示例：

编写无服务器云函数

首先，我们需要根据本场景要求编写无服务器云函数 hello 的源代码。

导入模块

为便于发起 HTTP GET 请求，我们采用 Python HTTP 库 requests 实现。

import requests

入口函数

接着，我们需要为无服务器云函数 hello 编写入口函数。

def main(
    event: dict,
    context: dict
) -> str:
    return requests.get('https://libget.com/hello').text

在上文示例代码中，我们导入依赖模块 requests 并使用 requests.get 函数向 https://libget.com/hello 发起 HTTP GET 请求并将响应正文（Body）作为无服务器云函数 hello 的返回值。

编写本地脚本

然后，我们需要根据本场景要求编写一个 Python 脚本并实现计划定时调用无服务器云函数 hello。

导入模块

首先我们需要在无服务器云函数 hello 中导入 Tencent Cloud SDK for Python 相关命名空间和模块：

import time

from tencent.cloud.auth import credentials
from tencent.cloud.serverless import functions

回调函数

为便于获取计划定时调用的无服务器云函数运行结果，我们需要定义一个调用完成后的回调函数。当指定无服务器云函数在指定时间调用并运行完成后，Tencent Cloud SDK for Python 将调用指定回调函数。

def completed_callback(
    function_schedule: functions.FunctionSchedule
):
    # Check if serverless cloud function is called and run successfully
    if not function_schedule.is_successful:
        # Throws an exception thrown by the task
        raise function_schedule.exception   

    # Print the actual return value of the serverless cloud function
    print('info: ' + str(function_schedule.return_value))

入口函数

接着，我们需要为 Python 脚本编写入口函数代码。

def main():
    # Instantiate a serverless cloud function product client
    function_client: functions.Client = functions.Client(
        # Instantiate an access credential
        access_credentials = credentials.Credentials(
            secret_id = 'AKIDF3sMOAU1pOgkmrKHchX6TZQ1Mo1C5qa7',
            secret_key = 'b4JL8fwxkIgsKMXGi39yYt0ECxZw4wZf'
        )
    )

    # Get the current native UNIX timestamp value
    current_unix_timestamp: int = int(time.time())

    # Loop 3 times to create a scheduled invoke task
    for loop_count in range(3):
        function_client.schedule_invoke(
            region_id = 'ap-shanghai',
            namespace_name = 'default',
            function_name = 'hello',
            invoke_timestamp = current_unix_timestamp + ((loop_count + 1) * 10),
            invoked_callback = completed_callback   
        )

    # Run all scheduled tasks until completion
    function_client.run_schedule()

在上文示例代码中我们首先实例化一个无服务器云函数产品客户端和访问凭据，接着获取当前本机 UNIX 时间戳，然后循环 3 次并分别创建相对于前一步获取的本机 UNIX 时间戳的 10 秒、20 秒和 30 秒定时无服务器云函数调用任务，最后运行已创建的所有定时调用任务并等待完成。

值得注意的是上文中示例代码以同步编程方式使用 Tencent Cloud SDK for Python，对于异步编程方式将无需调用无服务器云函数产品客户端实例的 run_schedule 方法，所有已创建的定时调用任务将被 Tencent Cloud SDK for Python 自动调度运行。

运行

最后，我们运行本场景中所创建并编写的本地 Python 脚本，如果一切正常将打印输出：

info: hello libget!
info: hello libget!
info: hello libget!
上一页无服务器云函数间调用下一页原生调用无服务器云函数

最后更新于