0

I trying to import newer files twice a day at 12am and 12pm. I am using the following format;

now = datetime.now()
day_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
day_end = day_start + timedelta(hours=12)

This will only capture one time. Is there a way to do this differently?

borfo
  • 201
  • 2
  • 4
  • 14
  • 1
    Use `cron` to run your script on a schedule. – Barmar Jun 02 '21 at 17:22
  • If you are using linux or mac, I think cronjob might be the way to go, but it's just my opinion – Mark Jun 02 '21 at 17:23
  • 2
    Does this answer your question? [Scheduling Python Script to run every hour accurately](https://stackoverflow.com/questions/22715086/scheduling-python-script-to-run-every-hour-accurately) – nofinator Jun 02 '21 at 18:07

2 Answers2

0

For production ready tool that is pure python, I would use airflow. Airflow is pure Python pipelines orchestral tool that will allows building of pipelines/dags/jobs with beautiful UI to see how the tasks are doing, plus aREST-APIs to interact with tasks from other tools.

For simple things, there multiple libraries: examples:

You could use schedule. It’s available via python -m pip install --user schedule


import schedule
import time

def file_loader():
    print("I'm loading files...")

schedule.every().day.at("12:00").do(file_loader)
schedule.every().day.at("0:00").do(file_loader)


while True:
    schedule.run_pending()
    time.sleep(1)

Anothe tool is apscheduler also in PyPI pip install apscheduler


from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor


jobstores = {
    'mongo': {'type': 'mongodb'},
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': {'type': 'threadpool', 'max_workers': 20},
    'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BackgroundScheduler()

scheduler.add_job(file_loader, 'cron', hour=12, id='file_loader_id_12')

scheduler.add_job(file_loader, 'cron', hour=0, id='file_loader_id_24')

scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

scheduler.start()
Prayson W. Daniel
  • 14,191
  • 4
  • 51
  • 57
0

You could keep the script running the whole day by just putting the program to sleep for 12 hours

Put your task in an infinite loop with an input-based exit as follows

import datetime as dt

# Initialize to Start of Day
LAST_TASK_TIME = dt.datetime.combine(dt.date.today(), dt.time(0))

while True:
    if (dt.datetime.now() - LAST_TASK_TIME).total_seconds() / 3600 >= 12:
        do_task()
        
        LAST_TASK_TIME += dt.timedelta(hours=12)

        if input("Do you want to exit [Y / N] : ").strip().upper()[0] == 'Y':
            break
        else:
            sleep(43195) # Put the program to sleep for 12 hours (-5 seconds)

post_exit_code_if_required()

This code will now run whatever code you want, once every 12 hours along with asking for a continuation confirmation before waiting for the next task time

If you do not want this waiting loop to interrupt your program flow, run this very code in a thread for parallel execution with some other main task (for example, a GUI Application on __main__)

Sagar
  • 404
  • 3
  • 14