使用 Python 的 azure 函数使用 Timer trigger 触发 HttpTrigger 函数

Question

我想在每天晚上 9 点使用 Timer 触发器触发 HttpTrigger 函数。是否可以在 Azure 函数中用于 python？

我无法理解如何实现它？有没有办法实现上述场景？

请检查以下代码。

HttpTrigger 功能代码

import logging

import azure.functions as func


def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')
            logging.info(name)

    if name:
        return func.HttpResponse(f"Hello {name}!")
    else:
        return func.HttpResponse(
             "Please pass a name on the query string or in the request body",
             status_code=400
        )

TimerTrigger 功能代码：-

import datetime
import logging
import requests

import azure.functions as func


def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()

    if mytimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function ran at %s', utc_timestamp)

    url = '<My URL>'
    PARAMS = {"name": "Akshay"}

    response_decoded_json = requests.post(url=url, params =PARAMS)
    print(response_decoded_json.text)
    logging.info("Result is :- ")
    logging.info(response_decoded_json.text)

请查看以下错误日志图片：-

Answer 1

当然这是可用的，只需在您的计时器触发函数中执行获取/发布请求。下面是我的测试代码。

import datetime
import logging
import requests

import azure.functions as func

def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()

    if mytimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function ran at %s', utc_timestamp)
    url = 'your function url'
    PARAMS = {"name": "test"}

    response_decoded_json = requests.post(url=url, params =PARAMS)
    print(response_decoded_json.text)

我得到了http触发函数的响应，根据你的要求，将cron表达式改为0 0 21 * * *。

更新：我测试这是可用的。

定时器功能代码：

import datetime
import logging
import requests

import azure.functions as func


def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()

    if mytimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function ran at %s', utc_timestamp)
    url = 'https://functionname.azurewebsites.net/api/httptest?code=master key'
    PARAMS = {"name": "test"}

    response_decoded_json = requests.post(url=url, params =PARAMS)
    logging.info(response_decoded_json.text)

HTTP功能代码：

import logging

import azure.functions as func


def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        return func.HttpResponse(f"Hello {name}!")
    else:
        return func.HttpResponse(
             "Please pass a name on the query string or in the request body",
             status_code=400
        )

上传函数时记得在requirements.txt中添加requests==2.22.0。

假设主要问题是关于如何获取httptrigger函数的url。您需要的主要参数是函数APP名称、函数名称和主密钥（或默认密钥）的url。 Function APP名称和函数名称其实你已经知道了。

关于代码（key），你可以先部署任何功能，然后进入管理页面你将能够得到代码并粘贴在url之后你将得到完整的http触发url。然后上传两个函数，就可以了。