Ceci est une version améliorée des astuces que j'ai publiées hier. La partie à notifier est séparée en passant une fonction de gestionnaire afin qu'un traitement arbitraire puisse être effectué.
J'ai également essayé d'utiliser functools.wrap, que @methane m'a appris. J'ai essayé d'utiliser la fonction avec ce décorateur dans @task du pavé, mais l'aide est devenue None et la docstring n'a pas été affichée car je ne l'ai pas utilisée! (Quand j'ai utilisé functools.wrap, j'ai un docstring dans l'aide)
timeout.py
from functools import wraps
def on_timeout(limit, handler, hint=None):
'''
S'il ne se termine pas dans l'heure d'exécution spécifiée, gestionnaire d'indices/Appel avec limite comme argument
@on_timeout(limit=3600, handler=notify_func, hint=u'Calcul long')
def long_time_function():
'''
def notify_handler(signum, frame):
handler("'%s' is not finished in %d second(s)." % (hint, limit))
def __decorator(function):
def __wrapper(*args, **kwargs):
import signal
signal.signal(signal.SIGALRM, notify_handler)
signal.alarm(limit)
result = function(*args, **kwargs)
signal.alarm(0)
return result
return wraps(function)(__wrapper)
return __decorator
L'utilisation réelle est la suivante
main.py
from timeout import on_timeout
def handler_func(msg):
print msg #Effectuez réellement un traitement de notification approprié
@on_timeout(limit=1, handler=handler_func, hint=u'Calcul long')
def long_time_function():
import time
time.sleep (2)
if __name__ == "__main__":
long_time_function()
Quand ceci est exécuté, le processus est en veille (2) même si la limite est de 1 seconde, donc handler_func est appelé après 1 seconde, et l'affichage est le suivant.
'Calcul long' is not finished in 1 second(s).
Avec cela, il semble qu'il puisse être appliqué à d'autres qu'AWS.
Recommended Posts