### Install Schedule using PIP Source: https://schedule.readthedocs.io/en/stable/installation The recommended method for installing the Schedule package is by using pip. This command ensures you get the latest version and is the most straightforward way to add the library to your Python environment. ```bash pip install schedule ``` -------------------------------- ### Manually Install Schedule Source: https://schedule.readthedocs.io/en/stable/installation If you need more control or cannot use a package manager, you can manually install Schedule by copying its single source file. This involves fetching the code from GitHub and placing it into a 'schedule' package within your project. ```python import schedule # Your code here ``` -------------------------------- ### Retrieve All Scheduled Jobs Source: https://schedule.readthedocs.io/en/stable/examples Shows how to get a list of all currently scheduled jobs using the `schedule.get_jobs()` method. This can be useful for monitoring or debugging. ```Python import schedule def hello(): print('Hello world') schedule.every().second.do(hello) all_jobs = schedule.get_jobs() ``` -------------------------------- ### Get Jobs Filtered by Tags Source: https://schedule.readthedocs.io/en/stable/examples Illustrates how to retrieve specific jobs from the scheduler by filtering them using tags. Jobs can be tagged with multiple identifiers. ```Python import schedule def greet(name): print('Hello {}'.format(name)) schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend') schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend') schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer') schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest') friends = schedule.get_jobs('friend') ``` -------------------------------- ### Install Development Requirements Source: https://schedule.readthedocs.io/en/stable/development Installs all necessary tooling and libraries for developing the Schedule library using the provided requirements file. ```Shell pip install -r requirements-dev.txt ``` -------------------------------- ### Install Schedule from AUR using yay Source: https://schedule.readthedocs.io/en/stable/installation For Arch Linux users, the python-schedule package is available in the Arch User Repository (AUR). The 'yay' helper can be used to install it directly from the AUR. ```bash yay -S python-schedule ``` -------------------------------- ### Verify schedule Installation (apt) Source: https://schedule.readthedocs.io/en/stable/faq This command verifies if the 'python3-schedule' package is installed using apt and displays its version. This is an alternative verification method for users who installed via apt. ```Shell dpkg -l | grep python3-schedule ``` -------------------------------- ### Schedule Jobs Using Decorators Source: https://schedule.readthedocs.io/en/stable/examples Illustrates how to use the `@repeat` decorator to schedule a function. The decorator takes the same interval syntax as the `every()` method. ```Python from schedule import every, repeat import time @repeat(every(10).minutes) def job(): print("I am a scheduled job") while True: run_pending() time.sleep(1) ``` -------------------------------- ### Basic Job Scheduling Example Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Demonstrates how to schedule jobs to run at specific intervals, daily at a certain time, and how to continuously run pending jobs. ```Python >>> import schedule >>> import time >>> def job(message='stuff'): >>> print("I'm working on:", message) >>> schedule.every(10).minutes.do(job) >>> schedule.every(5).to(10).days.do(job) >>> schedule.every().hour.do(job, message='things') >>> schedule.every().day.at("10:30").do(job) >>> while True: >>> schedule.run_pending() >>> time.sleep(1) ``` -------------------------------- ### Schedule Jobs with Timezones using .at() Source: https://schedule.readthedocs.io/en/stable/timezones Demonstrates how to schedule jobs at specific times in different timezones using the .at() method. It shows examples of passing timezone as a string and as a pytz timezone object. Requires the 'pytz' library to be installed. ```Python import schedule from pytz import timezone def job(): print("I'm working...") # Pass a timezone as a string schedule.every().day.at("12:42", "Europe/Amsterdam").do(job) # Pass an pytz timezone object schedule.every().friday.at("12:42", timezone("Africa/Lagos")).do(job) ``` -------------------------------- ### Schedule Jobs at Random Intervals Source: https://schedule.readthedocs.io/en/stable/examples Demonstrates scheduling a job to run at random intervals within a specified range using `every(A).to(B).seconds`. ```Python def my_job(): print('Foo') # Run every 5 to 10 seconds. schedule.every(5).to(10).seconds.do(my_job) ``` -------------------------------- ### Verify schedule Installation (pip) Source: https://schedule.readthedocs.io/en/stable/faq This command verifies if the 'schedule' library is installed using pip and displays its version. It's a common step to troubleshoot 'ModuleNotFoundError'. ```Shell pip3 list | grep schedule ``` -------------------------------- ### Install pytz for Timezone Support Source: https://schedule.readthedocs.io/en/stable/faq This command installs the 'pytz' library, which is a required dependency for using timezone features within the 'schedule' library. It resolves 'ModuleNotFoundError: No module named 'pytz''. ```Shell pip install pytz ``` -------------------------------- ### Schedule Jobs at Intervals Source: https://schedule.readthedocs.io/en/stable/examples Demonstrates how to schedule a job to run at various fixed intervals (seconds, minutes, hours, days, weeks) and at specific times within those intervals. ```Python import schedule import time def job(): print("I'm working...") # Run job every 3 second/minute/hour/day/week, # Starting 3 second/minute/hour/day/week from now schedule.every(3).seconds.do(job) schedule.every(3).minutes.do(job) schedule.every(3).hours.do(job) schedule.every(3).days.do(job) schedule.every(3).weeks.do(job) # Run job every minute at the 23rd second schedule.every().minute.at(":23").do(job) # Run job every hour at the 42nd minute schedule.every().hour.at(":42").do(job) # Run jobs every 5th hour, 20 minutes and 30 seconds in. # If current time is 02:00, first execution is at 06:20:30 schedule.every(5).hours.at("20:30").do(job) # Run job every day at specific HH:MM and next HH:MM:SS schedule.every().day.at("10:30").do(job) schedule.every().day.at("10:30:42").do(job) schedule.every().day.at("12:42", "Europe/Amsterdam").do(job) # Run job on a specific day of the week schedule.every().monday.do(job) schedule.every().wednesday.at("13:15").do(job) schedule.every().minute.at(":17").do(job) while True: schedule.run_pending() time.sleep(1) ``` -------------------------------- ### Customize Job Logging with Decorator Source: https://schedule.readthedocs.io/en/stable/logging This example shows how to create a Python decorator to log the execution time and name of a scheduled job. The decorator wraps the job function, records the start time, executes the job, and then logs the completion time and duration. ```Python import functools import time import schedule # This decorator can be applied to any job function to log the elapsed time of each job def print_elapsed_time(func): @functools.wraps(func) def wrapper(*args, **kwargs): start_timestamp = time.time() print('LOG: Running job "%s"' % func.__name__) result = func(*args, **kwargs) print('LOG: Job "%s" completed in %d seconds' % (func.__name__, time.time() - start_timestamp)) return result return wrapper @print_elapsed_time def job(): print('Hello, Logs') time.sleep(5) schedule.every().second.do(job) schedule.run_all() ``` -------------------------------- ### Install schedule using pip3 Source: https://schedule.readthedocs.io/en/stable/faq This command installs or upgrades the 'schedule' library using pip3, ensuring compatibility with Python 3 environments. It's recommended when encountering 'ModuleNotFoundError' due to Python version mismatches. ```Shell pip3 install schedule ``` -------------------------------- ### Schedule Jobs Until a Certain Time Source: https://schedule.readthedocs.io/en/stable/examples Shows how to schedule a job to run until a specific time or for a duration. This can be done using a time string or a `timedelta` object. ```Python import schedule from datetime import datetime, timedelta, time def job(): print('Boo') # run job until a 18:30 today schedule.every(1).hours.until("18:30").do(job) # run job until a 2030-01-01 18:33 today schedule.every(1).hours.until("2030-01-01 18:33").do(job) # Schedule a job to run for the next 8 hours schedule.every(1).hours.until(timedelta(hours=8)).do(job) ``` -------------------------------- ### Pass Arguments to Scheduled Jobs Source: https://schedule.readthedocs.io/en/stable/examples Shows how to pass arguments to a scheduled job function using the `do()` method. It also demonstrates passing arguments with the `@repeat` decorator. ```Python import schedule def greet(name): print('Hello', name) schedule.every(2).seconds.do(greet, name='Alice') schedule.every(4).seconds.do(greet, name='Bob') from schedule import every, repeat @repeat(every().second, "World") @repeat(every().day, "Mars") def hello(planet): print("Hello", planet) ``` -------------------------------- ### Get Seconds Until Next Execution Source: https://schedule.readthedocs.io/en/stable/examples Calculates the number of seconds until the next scheduled job is set to run using `schedule.idle_seconds()`. Returns a negative value if the job was scheduled for the past, and `None` if no jobs are pending. ```Python import schedule import time def job(): print('Hello') schedule.every(5).seconds.do(job) while 1: n = schedule.idle_seconds() if n is None: break elif n > 0: time.sleep(n) schedule.run_pending() ``` -------------------------------- ### Fix Unicode Handling in setup.py Source: https://schedule.readthedocs.io/en/stable/changelog Resolves an issue with Unicode handling in `setup.py` that caused problems on Python 3 and Debian systems. This ensures smoother installation and compatibility. It addresses issue #27. ```Python # No direct code example, but the fix relates to the setup.py file for package installation. ``` -------------------------------- ### Run All Scheduled Jobs Immediately Source: https://schedule.readthedocs.io/en/stable/examples Executes all pending jobs immediately, regardless of their scheduled time. Jobs are re-scheduled after completion. Supports an optional delay between job executions. ```Python import schedule def job_1(): print('Foo') def job_2(): print('Bar') schedule.every().monday.at("12:40").do(job_1) schedule.every().tuesday.at("16:40").do(job_2) schedule.run_all() # Add the delay_seconds argument to run the jobs with a number # of seconds delay in between. schedule.run_all(delay_seconds=10) ``` -------------------------------- ### Run a Job Once and Cancel Source: https://schedule.readthedocs.io/en/stable/examples Explains how to make a job run only once by returning `schedule.CancelJob` from the job function. This automatically removes the job from the scheduler after execution. ```Python import schedule import time def job_that_executes_once(): # Do some work that only needs to happen once... return schedule.CancelJob schedule.every().day.at('22:30').do(job_that_executes_once) while True: schedule.run_pending() time.sleep(1) ``` -------------------------------- ### Schedule Job Until Specific Datetime Source: https://schedule.readthedocs.io/en/stable/examples Schedules a job to run every hour until a specific datetime. The job will cease execution after the provided datetime. ```Python import schedule from datetime import datetime def job(): print('This job runs until a specific datetime.') schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job) ``` -------------------------------- ### Cancel All Scheduled Jobs Source: https://schedule.readthedocs.io/en/stable/examples Demonstrates how to clear the scheduler and cancel all pending jobs using the `schedule.clear()` method. This is useful for resetting the scheduler. ```Python import schedule def greet(name): print('Hello {}'.format(name)) schedule.every().second.do(greet) schedule.clear() ``` -------------------------------- ### Run Schedule in Background Thread (Python) Source: https://schedule.readthedocs.io/en/stable/background-execution This Python code demonstrates how to run scheduled jobs in a background thread using the 'schedule' and 'threading' libraries. It defines a function to continuously execute pending jobs at specified intervals and includes an example of starting and stopping this background thread. ```Python import threading import time import schedule def run_continuously(interval=1): """Continuously run, while executing pending jobs at each elapsed time interval. @return cease_continuous_run: threading. Event which can be set to cease continuous run. Please note that it is *intended behavior that run_continuously() does not run missed jobs*. For example, if you've registered a job that should run every minute and you set a continuous run interval of one hour then your job won't be run 60 times at each interval but only once. """ cease_continuous_run = threading.Event() class ScheduleThread(threading.Thread): @classmethod def run(cls): while not cease_continuous_run.is_set(): schedule.run_pending() time.sleep(interval) continuous_thread = ScheduleThread() continuous_thread.start() return cease_continuous_run def background_job(): print('Hello from the background thread') schedule.every().second.do(background_job) # Start the background thread stop_run_continuously = run_continuously() # Do some other things... time.sleep(10) # Stop the background thread stop_run_continuously.set() ``` -------------------------------- ### Schedule Job with Timezone and DST Example Source: https://schedule.readthedocs.io/en/stable/timezones Illustrates scheduling a job at a specific time in a different timezone ('America/New_York') and shows how to retrieve the next scheduled run time. This example highlights the impact of Daylight Saving Time on the next run calculation. ```Python import schedule def job(): print("I'm working...") s = schedule.every().day.at("10:30", "America/New_York").do(job) print(s.next_run) # Expected output: 2022-03-20 15:30:00 (assuming local time is Europe/Berlin and DST is in effect in New York) print(repr(s)) # Expected output: Every 1 day at 10:30:00 do job() (last run: [never], next run: 2022-03-20 15:30:00) ``` -------------------------------- ### Schedule Job Until Specific Time Source: https://schedule.readthedocs.io/en/stable/examples Schedules a job to run every hour until a specific time (HH:MM:SS). The job will not execute after the specified deadline. ```Python import schedule from datetime import time def job(): print('This job runs until a specific time.') schedule.every(1).hours.until(time(11, 33, 42)).do(job) ``` -------------------------------- ### Cancel Jobs Filtered by Tags Source: https://schedule.readthedocs.io/en/stable/examples Shows how to cancel a group of scheduled jobs by filtering them using tags. The `schedule.clear(tag)` method removes all jobs associated with a given tag. ```Python import schedule def greet(name): print('Hello {}'.format(name)) schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend') schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend') schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer') schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest') schedule.clear('daily-tasks') ``` -------------------------------- ### Cancel a Specific Scheduled Job Source: https://schedule.readthedocs.io/en/stable/examples Demonstrates how to cancel a single scheduled job using the `schedule.cancel_job(job)` method. A job object is returned when scheduling, which can then be used for cancellation. ```Python import schedule def some_task(): print('Hello world') job = schedule.every().day.at('22:30').do(some_task) schedule.cancel_job(job) ``` -------------------------------- ### Use a job queue and worker threads for parallel execution Source: https://schedule.readthedocs.io/en/stable/parallel-execution This example shows a more controlled approach to parallel job execution using a shared job queue and multiple worker threads. It utilizes Python's 'queue' module to manage jobs. A 'worker_main' function continuously fetches jobs from the queue and executes them. The main loop schedules jobs to be put into the queue and then runs pending schedule tasks. ```Python import time import threading import schedule import queue def job(): print("I'm working") def worker_main(): while 1: job_func = jobqueue.get() job_func() jobqueue.task_done() jobqueue = queue.Queue() schedule.every(10).seconds.do(jobqueue.put, job) schedule.every(10).seconds.do(jobqueue.put, job) schedule.every(10).seconds.do(jobqueue.put, job) schedule.every(10).seconds.do(jobqueue.put, job) schedule.every(10).seconds.do(jobqueue.put, job) worker_thread = threading.Thread(target=worker_main) worker_thread.start() while 1: schedule.run_pending() time.sleep(1) ``` -------------------------------- ### Schedule Periodic Python Jobs Source: https://schedule.readthedocs.io/en/stable/index This example demonstrates how to schedule various periodic jobs using the 'schedule' library in Python. It includes scheduling jobs at fixed intervals (minutes, hours, days), at specific times, on specific days of the week, and with timezone support. The code continuously checks for and runs pending jobs. ```Python import schedule import time def job(): print("I'm working...") schedule.every(10).minutes.do(job) schedule.every().hour.do(job) schedule.every().day.at("10:30").do(job) schedule.every().monday.do(job) schedule.every().wednesday.at("13:15").do(job) schedule.every().day.at("12:42", "Europe/Amsterdam").do(job) schedule.every().minute.at(":17").do(job) while True: schedule.run_pending() time.sleep(1) ``` -------------------------------- ### Scheduler Class - Get Jobs by Tag Source: https://schedule.readthedocs.io/en/stable/reference This method retrieves jobs that have been tagged with a specific identifier. If no tag is provided, it returns all scheduled jobs. ```Python jobs = scheduler.get_jobs(tag='daily') ``` -------------------------------- ### Get Idle Seconds Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Calculates the time in seconds until the next job is scheduled to run. This function calls `idle_seconds` on the default scheduler. ```Python from typing import Optional def idle_seconds() -> Optional[float]: """Calls :meth:`idle_seconds ` on the :data:`default scheduler instance `. """ return default_scheduler.idle_seconds() ``` -------------------------------- ### Scheduler Class - Get Jobs by Tag Source: https://schedule.readthedocs.io/en/stable/_modules/schedule The `get_jobs` method retrieves scheduled jobs. It can return all jobs if no tag is provided, or filter jobs based on a specific tag. ```Python def get_jobs(self, tag: Optional[Hashable] = None) -> List["Job"]: """ Gets scheduled jobs marked with the given tag, or all jobs if tag is omitted. :param tag: An identifier used to identify a subset of jobs to retrieve """ if tag is None: return self.jobs[:] else: return [job for job in self.jobs if tag in job.tags] ``` -------------------------------- ### Get Next Run by Tag Source: https://schedule.readthedocs.io/en/stable/changelog This feature allows retrieval of the next scheduled job based on its assigned tag. This is useful for managing and executing specific sets of jobs. It addresses issue #463. ```Python schedule.get_jobs(tag='daily_report') ``` -------------------------------- ### Get Scheduled Jobs by Tag Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Retrieves a list of scheduled jobs filtered by a specific tag. This function calls the `get_jobs` method on the default scheduler instance. ```Python from typing import Optional, Hashable, List from schedule import Job def get_jobs(tag: Optional[Hashable] = None) -> List[Job]: """Calls :meth:`get_jobs ` on the :data:`default scheduler instance `. """ return default_scheduler.get_jobs(tag) ``` -------------------------------- ### Calculate Next Run Time Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Computes the next execution time for the job based on its interval and unit. Supports random intervals within a range and specific start days for weekly jobs. ```Python def _schedule_next_run(self) -> None: """ Compute the instant when this job should run next. """ if self.unit not in ("seconds", "minutes", "hours", "days", "weeks"): raise ScheduleValueError( "Invalid unit (valid units are `seconds`, `minutes`, `hours`, " "`days`, and `weeks`)" ) if self.latest is not None: if not (self.latest >= self.interval): raise ScheduleError("`latest` is greater than `interval`") interval = random.randint(self.interval, self.latest) else: interval = self.interval self.period = datetime.timedelta(**{self.unit: interval}) self.next_run = datetime.datetime.now() + self.period if self.start_day is not None: if self.unit != "weeks": raise ScheduleValueError("`unit` should be 'weeks'") weekdays = ( "monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday", ) if self.start_day not in weekdays: raise ScheduleValueError( "Invalid start day (valid start days are {})".format(weekdays) ) weekday = weekdays.index(self.start_day) days_ahead = weekday - self.next_run.weekday() ``` -------------------------------- ### Get Next Run Time Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Returns the next scheduled run time for jobs, optionally filtered by tag. This function calls `get_next_run` on the default scheduler. ```Python import datetime from typing import Optional, Hashable def next_run(tag: Optional[Hashable] = None) -> Optional[datetime.datetime]: """Calls :meth:`next_run ` on the :data:`default scheduler instance `. """ return default_scheduler.get_next_run(tag) ``` -------------------------------- ### Python Schedule: Time Unit Properties Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Provides properties to set the time unit for scheduled jobs. It includes checks to ensure the correct unit is used based on the job's interval, raising an IntervalError if mismatched. For example, accessing 'second' is only allowed if the interval is 1. ```Python @property def second(self): if self.interval != 1: raise IntervalError("Use seconds instead of second") return self.seconds @property def seconds(self): self.unit = "seconds" return self @property def minute(self): if self.interval != 1: raise IntervalError("Use minutes instead of minute") return self.minutes @property def minutes(self): self.unit = "minutes" return self @property def hour(self): if self.interval != 1: raise IntervalError("Use hours instead of hour") return self.hours @property def hours(self): self.unit = "hours" return self @property def day(self): if self.interval != 1: raise IntervalError("Use days instead of day") return self.days @property def days(self): self.unit = "days" return self @property def week(self): if self.interval != 1: raise IntervalError("Use weeks instead of week") return self.weeks @property def weeks(self): self.unit = "weeks" return self ``` -------------------------------- ### Run jobs in separate threads with schedule Source: https://schedule.readthedocs.io/en/stable/parallel-execution This code snippet demonstrates how to execute multiple 'schedule' jobs in parallel by assigning each job to its own thread. It imports necessary libraries like 'threading', 'time', and 'schedule'. The 'run_threaded' function starts a new thread for a given job function. The main loop continuously checks and runs pending scheduled jobs. ```Python import threading import time import schedule def job(): print("I'm running on thread %s" % threading.current_thread()) def run_threaded(job_func): job_thread = threading.Thread(target=job_func) job_thread.start() schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) schedule.every(10).seconds.do(run_threaded, job) while 1: schedule.run_pending() time.sleep(1) ``` -------------------------------- ### Scheduler Class - Get Next Run Time Source: https://schedule.readthedocs.io/en/stable/reference This method returns the datetime object for when the next job is scheduled to run. It can filter this by a specific tag or return the next run time for any job if no tag is specified. Returns None if no jobs are scheduled. ```Python next_run = scheduler.get_next_run(tag='important') ``` -------------------------------- ### Get Next Run Time for Scheduler Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Retrieves the datetime of the next scheduled job run. It filters jobs by an optional tag and returns the earliest next run time among the filtered jobs. Returns None if no jobs are scheduled or if the filtered list is empty. ```Python def get_next_run(self, tag: Optional[Hashable] = None ) -> Optional[datetime.datetime]: """ Datetime when the next job should run. :param tag: Filter the next run for the given tag parameter :return: A :class:`~datetime.datetime` object or None if no jobs scheduled """ if not self.jobs: return None jobs_filtered = self.get_jobs(tag) if not jobs_filtered: return None return min(jobs_filtered).next_run ``` -------------------------------- ### Compile Documentation with Sphinx Source: https://schedule.readthedocs.io/en/stable/development Compiles the project's documentation, which is written in reStructuredText and processed by Sphinx with the alabaster theme. The generated HTML files are located in `docs/_build/html`. ```Shell cd docs make html ``` -------------------------------- ### Publish New Version Source: https://schedule.readthedocs.io/en/stable/development Publishes a new version of the library. This involves updating changelog and author files, bumping the version number in configuration files, merging changes to master, and then executing the release command. ```Shell "Release X.Y.Z" git push ``` -------------------------------- ### Create Scheduler Instance Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Initializes a default Scheduler instance. This instance is used to manage and run scheduled jobs. ```Python from schedule import Scheduler default_scheduler = Scheduler() ``` -------------------------------- ### Format Code with Black Source: https://schedule.readthedocs.io/en/stable/development Formats the project's code using the black formatter, specifically version 20.8b1. This ensures consistent code style across the project. ```Shell black --version 20.8b1 ``` -------------------------------- ### Handle Job Functions Throwing Exceptions Source: https://schedule.readthedocs.io/en/stable/changelog Adds an FAQ item to address how to manage job functions that might throw exceptions. This helps users create more robust scheduling systems. This is documented in the FAQ. ```Python # Example of handling exceptions within a job: def job_with_exception(): try: # ... code that might raise an exception ... pass except Exception as e: print(f"An error occurred: {e}") # Optionally, return schedule.CancelJob here if the exception means the job should stop schedule.every().day.do(job_with_exception) ``` -------------------------------- ### Initialize Job Object Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Initializes a Job object with a specified interval and an optional scheduler. It sets up various attributes to define the job's behavior, including its execution interval, unit, function, and scheduling parameters. ```Python def __init__(self, interval: int, scheduler: Optional[Scheduler] = None): self.interval: int = interval # pause interval * unit between runs self.latest: Optional[int] = None # upper limit to the interval self.job_func: Optional[functools.partial] = None # the job job_func to run # time units, e.g. 'minutes', 'hours', ... self.unit: Optional[str] = None # optional time at which this job runs self.at_time: Optional[datetime.time] = None # optional time zone of the self.at_time field. Only relevant when at_time is not None self.at_time_zone = None # datetime of the last run self.last_run: Optional[datetime.datetime] = None # datetime of the next run self.next_run: Optional[datetime.datetime] = None # timedelta between runs, only valid for self.period: Optional[datetime.timedelta] = None # Specific day of the week to start on self.start_day: Optional[str] = None # optional time of final run self.cancel_after: Optional[datetime.datetime] = None self.tags: Set[Hashable] = set() # unique set of tags for the job self.scheduler: Optional[Scheduler] = scheduler # scheduler to register with ``` -------------------------------- ### Publish as Universal Wheel Source: https://schedule.readthedocs.io/en/stable/changelog Ensures the library is published to PyPI as a universal wheel, making it compatible with both Python 2 and Python 3 environments without requiring recompilation. This improves distribution and ease of use. ```Python # This is a build/distribution change, not directly reflected in runtime code. ``` -------------------------------- ### Run Tests with Pytest Source: https://schedule.readthedocs.io/en/stable/development Executes all tests using pytest, including coverage checks and formatting validation. This command ensures the code adheres to quality standards. ```Shell pytest ``` -------------------------------- ### Configure Schedule Logging Source: https://schedule.readthedocs.io/en/stable/logging This snippet demonstrates how to set up Python's logging module to capture messages from the 'schedule' logger at the DEBUG level. It includes basic job execution and cleanup. ```Python import schedule import logging logging.basicConfig() schedule_logger = logging.getLogger('schedule') schedule_logger.setLevel(level=logging.DEBUG) def job(): print("Hello, Logs") schedule.every().second.do(job) schedule.run_all() schedule.clear() ``` -------------------------------- ### String Representation of Job Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Provides a human-readable string representation of the Job object, including its interval, unit, associated function, arguments, and keyword arguments. ```Python def __str__(self) -> str: if hasattr(self.job_func, "__name__"): job_func_name = self.job_func.__name__ # type: ignore else: job_func_name = repr(self.job_func) return ("Job(interval={}, unit={}, do={}, args={}, kwargs={}").format( self.interval, self.unit, job_func_name, "()" if self.job_func is None else self.job_func.args, "{}" if self.job_func is None else self.job_func.keywords, ) ``` -------------------------------- ### Schedule Job Scheduling - Python Source: https://schedule.readthedocs.io/en/stable/reference Defines functions to schedule jobs using the default scheduler instance. The 'every' function allows specifying an interval for recurring tasks. ```Python import schedule # Schedule a job to run every interval schedule.every(interval=1) ``` -------------------------------- ### Schedule Job Function Specification Source: https://schedule.readthedocs.io/en/stable/reference The `do` method specifies the function to be called when the job runs. Any additional arguments provided are passed to this function. It returns the invoked job instance. ```Python def do(_job_func: Callable_ , _*args_ , _**kwargs_): """Specifies the job_func that should be called every time the job runs. Any additional arguments are passed on to job_func when the job runs. """ pass ``` -------------------------------- ### Scheduler Class - Schedule a New Periodic Job Source: https://schedule.readthedocs.io/en/stable/reference The every method initiates the scheduling of a new periodic job. It takes an interval, which is a quantity of a specific time unit, and returns an unconfigured Job object. ```Python job = scheduler.every(5).minutes ``` -------------------------------- ### Scheduler Class - Run All Jobs with Delay Source: https://schedule.readthedocs.io/en/stable/reference The run_all method executes all scheduled jobs, irrespective of their scheduled times. A specified delay in seconds is introduced between each job execution to help distribute the system load more evenly over time. ```Python scheduler.run_all(delay_seconds=5) ``` -------------------------------- ### Python Schedule: Specifying Run Time Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Allows setting a specific time for a job to run. The `at` method accepts a time string and an optional timezone. The format of the time string depends on the job's repetition frequency (daily, hourly, or minute-based). ```Python def at(self, time_str: str, tz: Optional[str] = None): """ Specify a particular time that the job should be run at. :param time_str: A string in one of the following formats: - For daily jobs -> `HH:MM:SS` or `HH:MM` - For hourly jobs -> `MM:SS` or `:MM` - For minute jobs -> `:SS` The format must make sense given how often the job is repeating; for example, a job that repeats every minute """ ``` -------------------------------- ### Schedule Job with functools.partial Source: https://schedule.readthedocs.io/en/stable/changelog Fixes an issue where scheduling a job with `functools.partial` as the job function would fail. This ensures compatibility with this common Python pattern. It addresses issue #27. ```Python from functools import partial def my_func(x, y): return x + y scheduled_job = partial(my_func, 1) schedule.every().day.do(scheduled_job, 2) ``` -------------------------------- ### Run Jobs at a Particular Time Every Hour Source: https://schedule.readthedocs.io/en/stable/changelog Updates the `at_time()` method to allow jobs to run at a specific minute within each hour. This offers more precise scheduling within hourly intervals. It addresses issue #27. ```Python every().hour.at(':15').do(job) ``` -------------------------------- ### Schedule All Job Execution - Python Source: https://schedule.readthedocs.io/en/stable/reference Allows for the execution of all scheduled jobs, with an optional delay in seconds. This is useful for batch processing or immediate execution. ```Python import schedule # Run all jobs with an optional delay schedule.run_all(delay_seconds=0) ``` -------------------------------- ### Scheduler Class - Run All Jobs Source: https://schedule.readthedocs.io/en/stable/_modules/schedule The `run_all` method executes all scheduled jobs, irrespective of their scheduled time. It includes an optional delay between job executions to help distribute system load. ```Python def run_all(self, delay_seconds: int = 0) -> None: """ Run all jobs regardless if they are scheduled to run or not. A delay of `delay` seconds is added between each job. This helps distribute system load generated by the jobs more evenly over time. :param delay_seconds: A delay added between every executed job """ logger.debug( "Running *all* %i jobs with %is delay in between", len(self.jobs), delay_seconds, ) for job in self.jobs[:]: self._run_job(job) time.sleep(delay_seconds) ``` -------------------------------- ### Scheduler Class - Schedule a New Job Source: https://schedule.readthedocs.io/en/stable/_modules/schedule The `every` method is used to schedule a new periodic job. It takes an interval and returns an unconfigured `Job` object, which can then be further configured with time units and actions. ```Python def every(self, interval: int = 1) -> "Job": """ Schedule a new periodic job. :param interval: A quantity of a certain time unit :return: An unconfigured :class:`Job ` """ job = Job(interval, self) return job ``` -------------------------------- ### Define Job Function and Arguments Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Specifies the function to be executed when the job runs and passes any additional arguments to it. It also handles scheduling the next run and associating the job with a scheduler. ```Python def do(self, job_func: Callable, *args, **kwargs): """ Specifies the job_func that should be called every time the job runs. Any additional arguments are passed on to job_func when the job runs. :param job_func: The function to be scheduled :return: The invoked job instance """ self.job_func = functools.partial(job_func, *args, **kwargs) functools.update_wrapper(self.job_func, job_func) self._schedule_next_run() if self.scheduler is None: raise ScheduleError( "Unable to a add job to schedule. " "Job is not associated with an scheduler" ) self.scheduler.jobs.append(self) return self ``` -------------------------------- ### Schedule a Job Every Interval Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Schedules a job to run at a specified interval using the default scheduler. This function acts as a convenient wrapper for the Scheduler's `every` method. ```Python from schedule import Job def every(interval: int = 1) -> Job: """Calls :meth:`every ` on the :data:`default scheduler instance `. """ return default_scheduler.every(interval) ``` -------------------------------- ### Detailed Representation of Job Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Generates a detailed string representation of the Job object, including last run time, next run time, and the call signature of the job function. It formats the output based on whether an `at_time` is specified. ```Python def __repr__(self): def format_time(t): return t.strftime("%Y-%m-%d %H:%M:%S") if t else "[never]" def is_repr(j): return not isinstance(j, Job) timestats = "(last run: %s, next run: %s)" % ( format_time(self.last_run), format_time(self.next_run), ) if hasattr(self.job_func, "__name__"): job_func_name = self.job_func.__name__ else: job_func_name = repr(self.job_func) if self.job_func is not None: args = [repr(x) if is_repr(x) else str(x) for x in self.job_func.args] kwargs = ["%s=%s" % (k, repr(v)) for k, v in self.job_func.keywords.items()] call_repr = job_func_name + "(" + ", ".join(args + kwargs) + ")" else: call_repr = "[None]" if self.at_time is not None: return "Every %s%s at %s do %s%s" % ( self.interval, self.unit[:-1] if self.interval == 1 else self.unit, self.at_time, call_repr, timestats, ) else: fmt = ( "Every %(interval)s " + ("to %(latest)s " if self.latest is not None else "") + "%(unit)s do %(call_repr)s%(timestats)s" ) return fmt % dict( interval=self.interval, ``` -------------------------------- ### Tag Jobs and Cancel Jobs by Tag Source: https://schedule.readthedocs.io/en/stable/changelog Introduces the ability to tag jobs for easier identification and management, and to cancel jobs based on these tags. This improves the control and organization of scheduled tasks. It addresses issue #404. ```Python schedule.every().day.tag('important').do(job) schedule.cancel_job_by_tag('important') ``` -------------------------------- ### Schedule Next Run Time Retrieval - Python Source: https://schedule.readthedocs.io/en/stable/reference Retrieves the next scheduled run time for all jobs or jobs matching a specific tag. Returns None if no jobs are scheduled or match the tag. ```Python import schedule import datetime import collections.abc from typing import Optional # Get the next run time for all jobs next_run_all: Optional[datetime.datetime] = schedule.next_run() # Get the next run time for jobs with a specific tag next_run_tagged: Optional[datetime.datetime] = schedule.next_run(tag='my_tag') ``` -------------------------------- ### Schedule Job to Run Only Once Source: https://schedule.readthedocs.io/en/stable/changelog Provides guidance on how to schedule a job that executes only a single time. This is a common requirement for one-off tasks. This is documented in the FAQ. ```Python # To schedule a job to run only once: schedule.every().day.do(job) # Then, after it runs, you would typically remove it or ensure it doesn't reschedule. ``` -------------------------------- ### Add Timezone Support for .at() Source: https://schedule.readthedocs.io/en/stable/changelog This change introduces timezone support for the .at() method, allowing jobs to be scheduled with specific timezones. This enhances the library's flexibility for global applications. It addresses issue #517. ```Python schedule.every().day.at("10:30").timezone("Europe/Lisbon").do(job) ``` -------------------------------- ### Python Schedule: Tagging Jobs Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Enables tagging jobs with unique identifiers to facilitate management and retrieval. The `tag` method accepts one or more hashable arguments, and any duplicate tags are automatically discarded. It returns the job instance for method chaining. ```Python def tag(self, *tags: Hashable): """ Tags the job with one or more unique identifiers. Tags must be hashable. Duplicate tags are discarded. :param tags: A unique list of ``Hashable`` tags. :return: The invoked job instance """ if not all(isinstance(tag, Hashable) for tag in tags): raise TypeError("Tags must be hashable") self.tags.update(tags) return self ``` -------------------------------- ### Compare Job Next Run Times Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Defines the less than comparison for Job objects based on their `next_run` attribute. This allows sorting jobs by their scheduled execution time. ```Python def __lt__(self, other) -> bool: """ PeriodicJobs are sortable based on the scheduled time they run next. """ return self.next_run < other.next_run ``` -------------------------------- ### Run All Scheduled Jobs Source: https://schedule.readthedocs.io/en/stable/_modules/schedule Executes all scheduled jobs, optionally with a delay. This function is a wrapper for the `run_all` method of the default scheduler. ```Python def run_all(delay_seconds: int = 0) -> None: """Calls :meth:`run_all ` on the :data:`default scheduler instance `. """ default_scheduler.run_all(delay_seconds=delay_seconds) ``` -------------------------------- ### Execute .until() Source: https://schedule.readthedocs.io/en/stable/changelog Adds the ability to schedule jobs to run until a specific time or condition is met. This provides more control over job execution lifecycles. It addresses issue #195. ```Python schedule.every().day.until("17:00").do(job) ``` -------------------------------- ### Schedule Job Retrieval - Python Source: https://schedule.readthedocs.io/en/stable/reference Enables retrieving a list of scheduled jobs, optionally filtered by a tag. This is useful for inspecting the current schedule. ```Python import schedule import collections.abc from typing import Optional, List # Get all jobs or jobs with a specific tag all_jobs: List[schedule.Job] = schedule.get_jobs() tagged_jobs: List[schedule.Job] = schedule.get_jobs(tag='my_tag') ``` -------------------------------- ### Add @repeat() Decorator Source: https://schedule.readthedocs.io/en/stable/changelog Introduces a decorator `@repeat()` that simplifies the process of scheduling jobs to run at a specified interval. This makes the code cleaner and more readable. It addresses issue #148. ```Python @schedule.repeat(schedule.every(10).minutes) def job(): print("I am called every 10 minutes") ``` -------------------------------- ### Python: Run Multiple Schedulers Concurrently Source: https://schedule.readthedocs.io/en/stable/multiple-schedulers This Python code snippet demonstrates how to create and manage multiple independent scheduler instances using the 'schedule' library. It shows adding different jobs to each scheduler and running them concurrently within a loop, ensuring that pending jobs from all schedulers are executed. ```Python import time import schedule def fooJob(): print("Foo") def barJob(): print("Bar") # Create a new scheduler scheduler1 = schedule.Scheduler() # Add jobs to the created scheduler scheduler1.every().hour.do(fooJob) scheduler1.every().hour.do(barJob) # Create as many schedulers as you need scheduler2 = schedule.Scheduler() scheduler2.every().second.do(fooJob) scheduler2.every().second.do(barJob) while True: # run_pending needs to be called on every scheduler scheduler1.run_pending() scheduler2.run_pending() time.sleep(1) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.