Airflow DAG中Unix时间戳转日期时间字段失败问题求助

• 字符串插值参数错误：get_data_str函数执行字符串格式化时，execution_time字段传入的是未调用的函数对象get_execution_time，而非函数执行后的返回值，导致直接输出函数内存地址，触发下游解析错误返回400。
• 时间格式不符合要求：当前get_execution_time返回的时间格式为YYYY-MM-DD HH:MM:SS，和预期的带时区标记的ISO 8601格式不匹配。
• 任务ID冲突：循环生成SimpleHttpOperator任务时，所有任务使用相同的task_id，会导致DAG加载失败。
• 冗余参数配置：SimpleHttpOperator同时传入了data参数和python_callable参数，两者功能冲突，python_callable配置不会生效。

1. 修改get_data_str函数的格式化参数，将函数对象替换为已生成的时间变量
2. 调整时间格式化逻辑，输出符合要求的ISO 8601格式时间
3. 调整循环内的task_id为唯一值，删除冗余的python_callable和provide_context参数

from airflow import DAG
from random import random, seed, choice
from datetime import datetime, timedelta
from airflow.operators.python_operator import PythonOperator
from airflow.operators.postgres_operator import PostgresOperator
from airflow.operators.http_operator import SimpleHttpOperator
import time
import warnings
import requests
from datetime import datetime
import calendar
from functools import wraps

warnings.filterwarnings("ignore",category=DeprecationWarning)


def get_execution_time():
    ts = int("1284101485")
    # 调整时间格式匹配预期输出
    return datetime.utcfromtimestamp(ts).strftime('%Y-%m-%dT%H:%M:%S+00:00')
    

model_list = ['m1', 'm2', 'm3', 'm4']
def get_data_str(model_name):
    table_name = 'model_i7,'
    execution_time=get_execution_time()
    # 修复参数错误，使用已调用得到的execution_time变量而非函数对象
    parameters = ',model_name={model_name},execution_time={execution_time}'.format(
    model_name=model_name,execution_time=execution_time)
    return table_name + 'model_owner=test_developer' + parameters   


default_args = {
    'owner': 'developer',
    'depends_on_past': False,
    'start_date': datetime(2021,10,4),
    'retries': 1,
    'retry_delay': timedelta(minutes=1),
    'catchup' : False,
    'dagrun_timeout' : timedelta(hours=3),
    'email_on_success': False,
    'email_on_failure': False,
    'email_on_retry': False

}

with DAG(
    dag_id='testi7',
    schedule_interval=None,
    tags=['testi7'],
    access_control = {'developer':{'can_dag_read','can_dag_edit'}},
    default_args=default_args 
) as dag:

    for model_name in model_list:
        extracting_metrics = SimpleHttpOperator(
            # 修复task_id冲突问题
            task_id=f'extracting_metrics_{model_name}',
            http_conn_id ='metrics_api',
            endpoint='/write',
            data=get_data_str(model_name)
)

修复后生成的指标参数中execution_time会显示为正确的时间值2010-09-10T10:51:25+00:00，不会再出现函数内存地址，HTTP 400报错将被解决，输出格式匹配需求。

内容的提问来源于stack exchange，提问作者Pravek