python调度程序在任何时区的特定时区调度作业

编程问答 2022-08-07

问题描述

我试图在每天早上 8:00 在 timezone('Asia/Calcutta') 运行一项工作，所以我开始使用 python 调度程序。示例代码段

import schedule
import time

def job(t):
    print "I'm working...",t
    return

schedule.every().day.at("08:00").do(job,'It is 08:00')

while True:
    schedule.run_pending()
    time.sleep(30) # wait one minute

如果在印度服务器中执行，此代码段效果很好，但如果我在 COLAB 或 AWS（位于不同时区）中运行，则作业会根据该特定时区在上午 8:00 开始。但我想让代码在这个时区（'亚洲/加尔各答'）每天早上 8:00 运行，而不管服务器时区。我浏览了与时区相关的 stackoverflow 中的不同文章，获取偏移并将 8:00 am 更改为类似 8:00 + 一些偏移 4:30 小时的内容。但没有用。

  1. 正在寻找在任何服务器/任何时区上运行 python 代码的最佳方式，但调度程序将在这个时区的时区（'Asia/Calcutta'）中触发。

  2. 有什么方法可以在线程级别或完整的python进程级别（仅针对该脚本/线程，而不是系统级别）更改日期时间时区，所以我想在python中更改一次时区，然后在每个我调用 datetime.Now()，它应该根据新的时区给出时间，

       eastern = timezone('Asia/Calcutta')
   # naive datetime
   naive_dt = datetime.Now()
   # localized datetime
   loc_dt = datetime.Now(eastern)
   print(naive_dt.strftime(fmt))

   print(loc_dt.strftime(fmt))

基于此（answer）对更改系统文件不感兴趣，看起来像pythonic方式

解决方法

我猜为时已晚，但您可以改用 simple_scheduler。它具有您需要的额外时区组件。

from simple_scheduler.event import event_scheduler
def job(t):
    print "I'm working...",t
    return
event_scheduler.add_job(target= job,kwargs = {"t":"It is 08:00"},when = ["*|08:00"] #"*": call function everyday
                        tz = "GMT")
event_scheduler.run()
,

如果您正在寻找支持时区的调度程序，您可能会对 scheduler library 感兴趣。可以在 documentation 中找到与您描述的类似情况的示例。我在下面对其进行了修改以适合您的问题，并另外添加了具有独立时区的第二个作业以进行更好的演示。

披露：我是调度程序库的作者之一

问题 1

设置您的服务器和作业时区并创建有效负载函数。

import datetime as dt
import time
from scheduler import Scheduler
import pytz 

TZ_SERVER = dt.timezone.utc  # can be any valid timezone

TZ_IST = pytz.timezone('Asia/Calcutta')  # IST - India Standard Time
TZ_CEST = pytz.timezone('Europe/Berlin')  # CEST - Central European Summer Time

def task(tzinfo):
    current_time = dt.datetime.now(tzinfo)
    print(f"I'm working... {current_time!s}")

创建一个 Scheduler 实例并在指定的时间和时区安排重复的日常工作：

schedule = Scheduler(tzinfo=TZ_SERVER)

trigger_ist = dt.time(hour=8,tzinfo=TZ_IST)
schedule.daily(trigger_ist,task,args=(TZ_IST,))

trigger_cest = dt.time(hour=8,tzinfo=TZ_CEST)
job_cest = schedule.daily(trigger_cest,args=(TZ_CEST,))

显示时间表的简单概述。

print(schedule)

max_exec=inf,tzinfo=UTC,priority_function=linear_priority_function,#jobs=2

type     function         due at              tzinfo          due in      attempts weight
-------- ---------------- ------------------- ------------ --------- ------------- ------
DAILY    task(..)         2021-07-20 08:00:00 IST            2:36:42         0/inf      1
DAILY    task(..)         2021-07-20 08:00:00 CEST           6:06:42         0/inf      1

创建类似于调度库的执行循环：

while True:
    schedule.exec_jobs()
    time.sleep(60)  # wait one minute

一旦调用了有效负载函数，您应该会看到类似下面的内容。

I'm working... 2021-07-20 08:00:33.173469+05:30

问题 2

我不确定我是否理解您的第二个问题，但始终可以将一个具有时区的 datetime.datetime 对象转换为具有另一个时区的另一个 datetime.datetime 对象。例如。你可以这样做：

now_cest = dt.datetime.now(TZ_CEST)
now_ist = now_cest.astimezone(TZ_IST)
datetimedatetimedatetimepythonpython-3.xschedulerschedulerschedulertimezonetimezone