如何仅运行一个函数 5 分钟?


import logging
from a.x.models import X
from a.x.management.commands.syncx 
import Command as SyncCommand
from a.x.adapter_classes import ADAPTER_CLASSES
LOGGER = logging.getLogger(__name__)

def logger_function(code):
if not X.objects.filter(code=code).exists():
X.objects.create(code=code)
LOGGER.info(f"{X} created")
args = []
kwargs = {'x_code': code,
'class': False,
'database': 'default'}
try:
LOGGER.info(f"Starting syncx command for {code}")
#or this command needs to be run just 5 minutes for every key
SyncCommand().handle(*args, **kwargs)
LOGGER.info(f"There is no error for {code}")
except Exception as error:
with open("logger.txt", "a") as file:
file.write(f"{code}'s error is : {error}")
LOGGER.info(f"Logging error about {code}n")

def run():
for key in ADAPTER_CLASSES.keys():
#this function needs to be run just 5 minutes for every key
logger_function(key)

我的logger_function需要运行 5 分钟。有没有带计时器的计时器装饰器或线程破坏器?我该怎么做。 我的 for 循环正在移入键并发送到记录器功能,如果有任何问题,请尝试除了阻止其 okey ,但如果我的 SyncCommand 一切正常,则可能需要几个小时,但我只想在前 5 分钟内记录错误。

有没有定时器装饰器

如果允许您使用外部库,我建议您查看超时装饰器。

# importing the required module 
import timeit 

# code snippet to be executed only once 
mysetup = "from math import sqrt"

# code snippet whose execution time is to be measured 
mycode = ''' 
def example(): 
mylist = [] 
for x in range(100): 
mylist.append(sqrt(x)) 
'''

# timeit statement 
print timeit.timeit(setup = mysetup, 
stmt = mycode, 
number = 10000) 

我使用信号库解决了它

def handler(signum, frame):
raise Exception(None)
#do stuff func
for key in ADAPTER_CLASSES.keys():
signal.signal(signal.SIGALRM, handler)
signal.alarm(300) #5 min
logger_function(key)
signal.alarm(0)

最新更新