AirFlow调度Python脚本：如何获取+5.30时区本地时间存入SQL表？

方案一：通过AirFlow上下文传递时区化时间 #

AirFlow任务实例的上下文自带执行时间变量，可直接获取并转换为目标时区：

• 在PythonOperator的调用函数中，通过**kwargs参数获取上下文，其中execution_date是带时区信息的UTC datetime对象
• 将其转换为Asia/Kolkata（+5:30对应的时区），得到的时间自带时区属性，适合后续SQL计算

示例代码：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import pendulum

def save_local_time(**kwargs):
    # 从AirFlow上下文获取UTC执行时间
    utc_execution_date = kwargs['execution_date']
    # 转换为+5:30时区
    local_tz = pendulum.timezone('Asia/Kolkata')
    local_time = utc_execution_date.in_timezone(local_tz)
    
    # 存入SQL表示例（以PostgreSQL为例）
    # cursor.execute("INSERT INTO your_table (local_time) VALUES (%s)", (local_time,))

with DAG(
    dag_id='local_time_dag',
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    save_task = PythonOperator(
        task_id='save_local_time',
        python_callable=save_local_time,
        provide_context=True  # 开启上下文传递
    )

方案二：在Python脚本内直接生成带时区的本地时间 #

如果不想依赖AirFlow上下文，可在脚本内直接生成目标时区的时间，避免使用datetime.now()（默认返回无时区的naive对象，会被AirFlow的UTC环境影响）：

• 使用Python 3.9+内置的zoneinfo库，或AirFlow预装的pendulum库生成带时区的datetime对象
• 生成的时间格式规范，自带时区偏移，可直接用于SQL存储和后续计算

用zoneinfo实现（无需额外安装） #

from datetime import datetime
from zoneinfo import ZoneInfo

# 获取+5:30时区的当前时间
local_time = datetime.now(tz=ZoneInfo('Asia/Kolkata'))
# 存入SQL时可直接传入对象，或转换为ISO 8601格式字符串：local_time.isoformat()

用pendulum实现（AirFlow默认预装） #

import pendulum

# 获取+5:30时区的当前时间
local_time = pendulum.now('Asia/Kolkata')
# 直接存入支持时区的SQL数据库即可

关键注意事项 #

• 禁止使用datetime.now()或datetime.utcnow()，这两个方法返回的naive datetime对象存入数据库时，会被AirFlow的UTC环境解析为UTC时间，导致时区偏差
• 优先使用带时区信息的datetime对象存入SQL表，现代数据库（PostgreSQL、MySQL 8.0+等）均支持TIMESTAMP WITH TIME ZONE类型，可避免后续计算的时区歧义
• 若数据库不支持带时区类型，可存储UTC时间并记录时区信息，或直接存储包含时区偏移的ISO 8601格式字符串，后续计算时再解析

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