调度和处理在给定日期运行的耗时函数

问题描述

我有一个 python 函数，它应该运行给定的日期和时间，它从一个 url 抓取视频，将其转换为给定的格式并将其上传到存储。

这个调度函数同步依赖其他3个函数（有关注点分离）

Def getVideo(url):
    #1. download the video from an URL
    Return scraped_video 

Def convertVideo(scraped_video):
    #2. Convert the video to a given format
    Return file_ouput_path

Def sendVideo(file):
    #3. Upload the video to a given Gdrive or DropBox


GrabAndConvertThenSend(url,notification_email):
    Try:
        Temp_video = getVideo(url)
        file_ouput_path = convertVideo(Temp_video)
        sendVideo(file_output_path)
        # send notification email with the link
        # Schedule the next run
    Except Exception as e:
        Print(e)

函数 GrabAndConvertThenSend() 通过 APScheduler 处理以在给定日期运行

1.如何实现重试？

有时，由于网络问题或 API 可用性，主要功能可能会中止。 例如：该功能已经下载了视频，但未能将其上传到存储中。

如何在不重新下载视频的情况下从停止位置继续播放？ 我想在数据库中存储状态（下载、转换、上传），这是正确的方法吗？

2.像这样链接函数是正确的方法吗？ 或者我应该依赖事件/侦听器， 甚至排队作业（当任务 A 完成时，它会将任务 B 排队），但是如何实现这一点，因为我的功能有一个单独的关注点

解决方法

如何实现重试？

这是一个通用的设计概念。在这里您可以使用恢复点。它包括设计一个工作流程，其中一个步骤的输入仅在该步骤产生其完整输出后才被删除。如果处理后来中断，您可以在最后一个成功步骤后重新启动作业。由于关注点分离，它不应在当前功能中实现，而应在负责以下内容的经理中实现：

• 在最后一步成功后重新启动
• 成功后调用下一步

在您的伪代码中，此管理器功能应在 GrabAndConvertThenSend 中实现：

Def GrabAndConvertThenSend(url,notification_email):
    Try:
        if not is_a_raw_video_file_present():
            Temp_video = getVideo(url)
            save_to_(temp_path)
            rename(temp_path,raw_video_path)
            remove(raw_video_file)
        if not is_a_converted_video_file_present():
            file_ouput_temp_path = convertVideo(open(raw_video_path).read())
            rename(file_output_temp_path,file_output_path)
            remove(raw_video_path)
        if is_a_converted_video_file_present():
            sendVideo(file_output_path)
            remove(file_output_path)
        # send notification email with the link
        # Schedule the next run
    Except Exception as e:
        Print(e)
        # schedule with a shorter time to finish processing

像这样链接函数是否正确？

这是一个品味问题。重要的是特征。上面的伪代码实现了恢复点并将其用于重试。如果它们满足您的要求，您可以使用其他作业管理工具，或者像这样使用手动实现

,

你可以试试simple_scheduler。对于每个事件，它提供 3 次尝试。

from simple_scheduler.event import event_scheduler
TZ = "America/Los_Angeles"
WHEN = ["mon|09:00","fri|16:00"]

event_scheduler.add_job(target = GrabAndConvertThenSend,args = (url,notification_email),when = WHEN,tz = TZ)
event_scheduler.run()