Schedule

https://schedule.readthedocs.io
Admin
Schedule is a lightweight Python job scheduling library that enables developers to run Python...
Tokens:14,572
Snippets:157
Trust Score:9.6
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install development dependencies

Source: https://schedule.readthedocs.io/en/stable/development.html

Install all required tooling and libraries for development.

```bash
pip install -r requirements-dev.txt
```

--------------------------------

### Install via PIP

Source: https://schedule.readthedocs.io/en/stable/installation.html

The recommended method for installing the schedule package.

```bash
$ pip install schedule
```

--------------------------------

### Install Missing Dependencies

Source: https://schedule.readthedocs.io/en/stable/faq.html

Commands to install the schedule package or the required pytz dependency.

```bash
pip3 install schedule
```

```bash
pip install pytz
```

--------------------------------

### Install pytz dependency

Source: https://schedule.readthedocs.io/en/stable/timezones.html

Install the required pytz library to enable timezone support in schedule.

```bash
pip install pytz
```

--------------------------------

### Validate Installation

Source: https://schedule.readthedocs.io/en/stable/faq.html

Commands to verify if the schedule package is installed via pip or apt.

```bash
pip3 list | grep schedule
```

```bash
dpkg -l | grep python3-schedule
```

--------------------------------

### Install via Ubuntu package manager

Source: https://schedule.readthedocs.io/en/stable/installation.html

Installs the schedule package on Ubuntu systems; note that this version may be outdated.

```bash
$ apt-get install python3-schedule
```

--------------------------------

### Install via Arch Linux AUR

Source: https://schedule.readthedocs.io/en/stable/installation.html

Installs the python-schedule package using the yay AUR helper.

```bash
$ yay -S python-schedule
```

--------------------------------

### Install via Debian package manager

Source: https://schedule.readthedocs.io/en/stable/installation.html

Installs the schedule package on Debian systems; note that this version may be outdated.

```bash
$ apt-get install python3 schedule
```

--------------------------------

### Schedule job with specific timezone

Source: https://schedule.readthedocs.io/en/stable/timezones.html

Example of scheduling a daily job at a specific time in the America/New_York timezone.

```python
s = every().day.at("10:30", "America/New_York").do(job)
```

--------------------------------

### Internal Schedule Logic

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Internal logic for calculating the next run time based on intervals, start days, and specific time constraints.

```python
        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()
            if days_ahead <= 0:  # Target day already happened this week
                days_ahead += 7
            self.next_run += datetime.timedelta(days_ahead) - self.period
        if self.at_time is not None:
            if self.unit not in ("days", "hours", "minutes") and self.start_day is None:
                raise ScheduleValueError("Invalid unit without specifying start day")
            kwargs = {"second": self.at_time.second, "microsecond": 0}
            if self.unit == "days" or self.start_day is not None:
                kwargs["hour"] = self.at_time.hour
            if self.unit in ["days", "hours"] or self.start_day is not None:
                kwargs["minute"] = self.at_time.minute
            self.next_run = self.next_run.replace(**kwargs)  # type: ignore

            if self.at_time_zone is not None:
                # Convert next_run from the expected timezone into the local time
                # self.next_run is a naive datetime so after conversion remove tzinfo
                self.next_run = (
                    self.at_time_zone.localize(self.next_run)
                    .astimezone()
                    .replace(tzinfo=None)
                )

            # Make sure we run at the specified time *today* (or *this hour*)
            # as well. This accounts for when a job takes so long it finished
            # in the next period.
            if not self.last_run or (self.next_run - self.last_run) > self.period:
                now = datetime.datetime.now()
                if (
                    self.unit == "days"
                    and self.at_time > now.time()
                    and self.interval == 1
                ):
                    self.next_run = self.next_run - datetime.timedelta(days=1)
                elif self.unit == "hours" and (
                    self.at_time.minute > now.minute
                    or (
                        self.at_time.minute == now.minute
                        and self.at_time.second > now.second
                    )
                ):
                    self.next_run = self.next_run - datetime.timedelta(hours=1)
                elif self.unit == "minutes" and self.at_time.second > now.second:
                    self.next_run = self.next_run - datetime.timedelta(minutes=1)
        if self.start_day is not None and self.at_time is not None:
            # Let's see if we will still make that time we specified today
            if (self.next_run - datetime.datetime.now()).days >= 7:
                self.next_run -= self.period
```

--------------------------------

### Get all jobs

Source: https://schedule.readthedocs.io/en/stable/examples.html

Retrieves a list of all currently scheduled jobs using get_jobs().

```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.html

Retrieves a subset of jobs by filtering them based on assigned tags.

```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')
```

--------------------------------

### Get next run time

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Returns the datetime of the next scheduled run for the default scheduler, optionally filtered by tag.

```python
def next_run(tag: Optional[Hashable] = None) -> Optional[datetime.datetime]:
    """Calls :meth:`next_run <Scheduler.next_run>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.get_next_run(tag)
```

--------------------------------

### Get idle time

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Returns the number of seconds until the next job is scheduled to run.

```python
def idle_seconds() -> Optional[float]:
    """Calls :meth:`idle_seconds <Scheduler.idle_seconds>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.idle_seconds
```

--------------------------------

### Publish a new version

Source: https://schedule.readthedocs.io/en/stable/development.html

Commands to tag the release, build the distribution packages, and upload them to PyPI using twine.

```bash
git tag X.Y.Z -m "Release X.Y.Z"
git push --tags

pip install --upgrade setuptools twine wheel
python setup.py sdist bdist_wheel --universal
twine upload --repository schedule dist/*
```

--------------------------------

### Compile documentation

Source: https://schedule.readthedocs.io/en/stable/development.html

Generate HTML documentation using Sphinx within the docs directory.

```bash
cd docs
make html
```

--------------------------------

### run()

Source: https://schedule.readthedocs.io/en/stable/reference.html

Run the job and immediately reschedule it.

```APIDOC
## run()

### Description
Run the job and immediately reschedule it. If the job's deadline is reached, the job is not run and CancelJob is returned.

### Returns
- The return value of the job_func, or CancelJob if the deadline is reached.
```

--------------------------------

### Schedule jobs with various intervals and times

Source: https://schedule.readthedocs.io/en/stable/examples.html

Demonstrates scheduling functions to run at specific intervals, at specific minutes/seconds, and on specific days of the week.

```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)
```

--------------------------------

### Initialize and run multiple schedulers

Source: https://schedule.readthedocs.io/en/stable/multiple-schedulers.html

Create distinct Scheduler instances and execute their pending jobs within a single loop.

```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)
```

--------------------------------

### Scheduler.every

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Initiates the creation of a new periodic job.

```APIDOC
## Scheduler.every(interval=1)

### Description
Schedule a new periodic job.

### Parameters
- **interval** (int) - Optional - A quantity of a certain time unit

### Returns
- **Job** - An unconfigured Job instance
```

--------------------------------

### at(time_str: str, tz: Optional[str] = None)

Source: https://schedule.readthedocs.io/en/stable/reference.html

Specifies a particular time that the job should be run at.

```APIDOC
## at(time_str, tz=None)

### Description
Specify a particular time that the job should be run at.

### Parameters
- **time_str** (str) - Required - A string representing the time (e.g., HH:MM:SS, HH:MM, MM:SS, :SS).
- **tz** (Optional[str]) - Optional - The timezone that this timestamp refers to, parsable by pytz.timezone().

### Returns
- The invoked job instance.
```

--------------------------------

### Default Scheduler Instance

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Global default scheduler instance and job list access.

```python
#: Default :class:`Scheduler <Scheduler>` object
default_scheduler = Scheduler()

#: Default :class:`Jobs <Job>` list
jobs = default_scheduler.jobs  # todo: should this be a copy, e.g. jobs()?
```

--------------------------------

### Run tests with coverage and linting

Source: https://schedule.readthedocs.io/en/stable/development.html

Execute the test suite using pytest with flake8 and coverage reporting enabled.

```bash
py.test test_schedule.py --flake8 schedule -v --cov schedule --cov-report term-missing
```

--------------------------------

### Execute job with run()

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Runs the job function and handles rescheduling or cancellation based on deadlines.

```python
def run(self):
        """
        Run the job and immediately reschedule it.
        If the job's deadline is reached (configured using .until()), the job is not
        run and CancelJob is returned immediately. If the next scheduled run exceeds
        the job's deadline, CancelJob is returned after the execution. In this latter
        case CancelJob takes priority over any other returned value.

        :return: The return value returned by the `job_func`, or CancelJob if the job's
                 deadline is reached.

        """
        if self._is_overdue(datetime.datetime.now()):
            logger.debug("Cancelling job %s", self)
            return CancelJob

        logger.debug("Running job %s", self)
        ret = self.job_func()
        self.last_run = datetime.datetime.now()
        self._schedule_next_run()

        if self._is_overdue(self.next_run):
            logger.debug("Cancelling job %s", self)
            return CancelJob
        return ret
```

--------------------------------

### Format code with black

Source: https://schedule.readthedocs.io/en/stable/development.html

Apply code formatting using the black tool. Ensure version 20.8b1 is used.

```bash
black .
```

--------------------------------

### Pass arguments to a job

Source: https://schedule.readthedocs.io/en/stable/examples.html

Demonstrates passing arguments to scheduled functions using the do() method or 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)
```

--------------------------------

### Scheduler Shortcut Functions

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Convenience functions that wrap the default scheduler instance.

```python
[docs]def every(interval: int = 1) -> Job:
    """Calls :meth:`every <Scheduler.every>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.every(interval)



[docs]def run_pending() -> None:
    """Calls :meth:`run_pending <Scheduler.run_pending>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    default_scheduler.run_pending()
```

--------------------------------

### do(job_func: Callable, *args, **kwargs)

Source: https://schedule.readthedocs.io/en/stable/reference.html

Specifies the job_func that should be called every time the job runs.

```APIDOC
## do(job_func, *args, **kwargs)

### Description
Specifies the function that should be called every time the job runs, passing any additional arguments provided.

### Parameters
- **job_func** (Callable) - Required - The function to be scheduled.
- ***args** - Optional - Positional arguments for the job function.
- ****kwargs** - Optional - Keyword arguments for the job function.

### Returns
- The invoked job instance.
```

--------------------------------

### Define job function with do()

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Specifies the function to execute when the job runs, including support for arguments and keyword arguments.

```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 jobs with timezones

Source: https://schedule.readthedocs.io/en/stable/timezones.html

Configure job execution times using timezone strings or pytz timezone objects within the .at() method.

```python
# Pass a timezone as a string
schedule.every().day.at("12:42", "Europe/Amsterdam").do(job)

# Pass an pytz timezone object
from pytz import timezone
schedule.every().friday.at("12:42", timezone("Africa/Lagos")).do(job)
```

--------------------------------

### until(until_time: Union[datetime.datetime, datetime.timedelta, datetime.time, str])

Source: https://schedule.readthedocs.io/en/stable/reference.html

Schedule job to run until the specified moment.

```APIDOC
## until(until_time)

### Description
Schedule job to run until the specified moment. The job is canceled if the next run is after the until_time.

### Parameters
- **until_time** (Union[datetime.datetime, datetime.timedelta, datetime.time, str]) - Required - A moment in the future representing the latest time a job can be run.

### Returns
- The invoked job instance.
```

--------------------------------

### Scheduling at Specific Times

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Configures a job to run at a specific time, supporting daily, hourly, and minutely intervals with optional timezone support.

```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
            should not be given a string in the form `HH:MM:SS`. The
            difference between `:MM` and `:SS` is inferred from the
            selected time-unit (e.g. `every().hour.at(':30')` vs.
            `every().minute.at(':30')`).

        :param tz: The timezone that this timestamp refers to. Can be
            a string that can be parsed by pytz.timezone(), or a pytz.BaseTzInfo object

        :return: The invoked job instance
        """
        if self.unit not in ("days", "hours", "minutes") and not self.start_day:
            raise ScheduleValueError(
                "Invalid unit (valid units are `days`, `hours`, and `minutes`)"
            )

        if tz is not None:
            import pytz

            if isinstance(tz, str):
                self.at_time_zone = pytz.timezone(tz)  # type: ignore
            elif isinstance(tz, pytz.BaseTzInfo):
                self.at_time_zone = tz
            else:
                raise ScheduleValueError(
                    "Timezone must be string or pytz.timezone object"
                )

        if not isinstance(time_str, str):
            raise TypeError("at() should be passed a string")
        if self.unit == "days" or self.start_day:
            if not re.match(r"^[0-2]\d:[0-5]\d(:[0-5]\d)?$", time_str):
                raise ScheduleValueError(
                    "Invalid time format for a daily job (valid format is HH:MM(:SS)?)"
                )
        if self.unit == "hours":
            if not re.match(r"^([0-5]\d)?:[0-5]\d$", time_str):
                raise ScheduleValueError(
                    "Invalid time format for an hourly job (valid format is (MM)?:SS)"
                )

        if self.unit == "minutes":
            if not re.match(r"^:[0-5]\d$", time_str):
                raise ScheduleValueError(
                    "Invalid time format for a minutely job (valid format is :SS)"
                )
        time_values = time_str.split(":")
        hour: Union[str, int]
        minute: Union[str, int]
        second: Union[str, int]
        if len(time_values) == 3:
            hour, minute, second = time_values
        elif len(time_values) == 2 and self.unit == "minutes":
            hour = 0
            minute = 0
            _, second = time_values
        elif len(time_values) == 2 and self.unit == "hours" and len(time_values[0]):
            hour = 0
            minute, second = time_values
        else:
            hour, minute = time_values
            second = 0
        if self.unit == "days" or self.start_day:
            hour = int(hour)
            if not (0 <= hour <= 23):
                raise ScheduleValueError(
                    "Invalid number of hours ({} is not between 0 and 23)"
                )
        elif self.unit == "hours":
            hour = 0
        elif self.unit == "minutes":
            hour = 0
            minute = 0
        hour = int(hour)
        minute = int(minute)
        second = int(second)
        self.at_time = datetime.time(hour, minute, second)
        return self
```

--------------------------------

### schedule.run_all(delay_seconds: int = 0)

Source: https://schedule.readthedocs.io/en/stable/reference.html

Executes all registered jobs regardless of their schedule.

```APIDOC
## schedule.run_all(delay_seconds: int = 0)

### Description
Run all jobs regardless if they are scheduled to run or not. A delay is added between each job to distribute system load.

### Parameters
- **delay_seconds** (int) - Optional - A delay added between every executed job.
```

--------------------------------

### Run all scheduled jobs

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Executes all jobs currently in the default scheduler instance with an optional delay.

```python
def run_all(delay_seconds: int = 0) -> None:
    """Calls :meth:`run_all <Scheduler.run_all>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    default_scheduler.run_all(delay_seconds=delay_seconds)
```

--------------------------------

### next_run(tag: Optional[Hashable] = None)

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Returns the datetime of the next scheduled run for the default scheduler.

```APIDOC
## next_run(tag: Optional[Hashable] = None)

### Description
Calculates and returns the next scheduled run time for jobs on the default scheduler, optionally filtered by tag.

### Parameters
- **tag** (Hashable) - Optional - The tag used to filter jobs for the next run calculation.
```

--------------------------------

### Configure job deadline with until()

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Sets a deadline for a job using various time formats. Throws ScheduleValueError if the time is in the past or the format is invalid.

```python
if isinstance(until_time, datetime.datetime):
            self.cancel_after = until_time
        elif isinstance(until_time, datetime.timedelta):
            self.cancel_after = datetime.datetime.now() + until_time
        elif isinstance(until_time, datetime.time):
            self.cancel_after = datetime.datetime.combine(
                datetime.datetime.now(), until_time
            )
        elif isinstance(until_time, str):
            cancel_after = self._decode_datetimestr(
                until_time,
                [
                    "%Y-%m-%d %H:%M:%S",
                    "%Y-%m-%d %H:%M",
                    "%Y-%m-%d",
                    "%H:%M:%S",
                    "%H:%M",
                ],
            )
            if cancel_after is None:
                raise ScheduleValueError("Invalid string format for until()")
            if "-" not in until_time:
                # the until_time is a time-only format. Set the date to today
                now = datetime.datetime.now()
                cancel_after = cancel_after.replace(
                    year=now.year, month=now.month, day=now.day
                )
            self.cancel_after = cancel_after
        else:
            raise TypeError(
                "until() takes a string, datetime.datetime, datetime.timedelta, "
                "datetime.time parameter"
            )
        if self.cancel_after < datetime.datetime.now():
            raise ScheduleValueError(
                "Cannot schedule a job to run until a time in the past"
            )
        return self
```

--------------------------------

### Schedule jobs at random intervals

Source: https://schedule.readthedocs.io/en/stable/examples.html

The to method allows defining a range for the interval, where the job executes at a random time between the two specified values.

```python
def my_job():
    print('Foo')

# Run every 5 to 10 seconds.
schedule.every(5).to(10).seconds.do(my_job)
```

--------------------------------

### to(latest: int)

Source: https://schedule.readthedocs.io/en/stable/reference.html

Schedule the job to run at an irregular (randomized) interval.

```APIDOC
## to(latest)

### Description
Schedule the job to run at an irregular (randomized) interval between the base interval and the latest value.

### Parameters
- **latest** (int) - Required - Maximum interval between randomized job runs.

### Returns
- The invoked job instance.
```

--------------------------------

### Job Class Definition and Initialization

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

The Job class manages periodic task configuration, including intervals, time units, and scheduling metadata.

```python
[docs]class Job:
    """
    A periodic job as used by :class:`Scheduler`.

    :param interval: A quantity of a certain time unit
    :param scheduler: The :class:`Scheduler <Scheduler>` instance that
                      this job will register itself with once it has
                      been fully configured in :meth:`Job.do()`.

    Every job runs at a given fixed time interval that is defined by:

    * a :meth:`time unit <Job.second>`
    * a quantity of `time units` defined by `interval`

    A job is usually created and returned by :meth:`Scheduler.every`
    method, which also defines its `interval`.
    """

    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

    def __lt__(self, other) -> bool:
        """
        PeriodicJobs are sortable based on the scheduled time they
        run next.
        """
        return self.next_run < other.next_run

    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,
        )

    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]"
```

--------------------------------

### schedule.every(interval: int = 1)

Source: https://schedule.readthedocs.io/en/stable/reference.html

Schedules a new periodic job using the default scheduler instance.

```APIDOC
## schedule.every(interval: int = 1)

### Description
Schedules a new periodic job. Returns an unconfigured Job instance.

### Parameters
- **interval** (int) - Optional - A quantity of a certain time unit.
```

--------------------------------

### schedule.get_jobs(tag: Optional[Hashable] = None)

Source: https://schedule.readthedocs.io/en/stable/reference.html

Retrieves a list of scheduled jobs, optionally filtered by tag.

```APIDOC
## schedule.get_jobs(tag: Optional[Hashable] = None)

### Description
Gets scheduled jobs marked with the given tag, or all jobs if tag is omitted.

### Parameters
- **tag** (Hashable) - Optional - An identifier used to identify a subset of jobs to retrieve.
```

--------------------------------

### Schedule Periodic Jobs

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Basic usage of the schedule library to define and run periodic tasks.

```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)
```

--------------------------------

### Internal Helper Methods

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Internal methods for checking overdue status and parsing datetime strings.

```python
    def _is_overdue(self, when: datetime.datetime):
        return self.cancel_after is not None and when > self.cancel_after

    def _decode_datetimestr(
        self, datetime_str: str, formats: List[str]
    ) -> Optional[datetime.datetime]:
        for f in formats:
            try:
                return datetime.datetime.strptime(datetime_str, f)
            except ValueError:
                pass
        return None
```

--------------------------------

### Inspect scheduled job runtime

Source: https://schedule.readthedocs.io/en/stable/timezones.html

Verify the next scheduled run time and job representation in the local timezone.

```python
print(s.next_run)
# 2022-03-20 15:30:00

print(repr(s))
# Every 1 day at 10:30:00 do job() (last run: [never], next run: 2022-03-20 15:30:00)
```

--------------------------------

### Execute all jobs immediately

Source: https://schedule.readthedocs.io/en/stable/examples.html

Trigger all scheduled jobs regardless of their timing using run_all. An optional delay_seconds argument can be provided to space out execution.

```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)
```

--------------------------------

### Scheduler Job Execution and Timing

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Internal methods for running jobs, calculating the next scheduled run time, and determining idle time.

```python
    def _run_job(self, job: "Job") -> None:
        ret = job.run()
        if isinstance(ret, CancelJob) or ret is CancelJob:
            self.cancel_job(job)


[docs]    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


    next_run = property(get_next_run)

    @property
    def idle_seconds(self) -> Optional[float]:
        """
        :return: Number of seconds until
                 :meth:`next_run <Scheduler.next_run>`
                 or None if no jobs are scheduled
        """
        if not self.next_run:
            return None
        return (self.next_run - datetime.datetime.now()).total_seconds()
```

--------------------------------

### Format job string representation

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Logic for generating a human-readable string representation of a scheduled job based on its interval and timing.

```python
        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,
                latest=self.latest,
                unit=(self.unit[:-1] if self.interval == 1 else self.unit),
                call_repr=call_repr,
                timestats=timestats,
            )
```

--------------------------------

### Configure Schedule Logging

Source: https://schedule.readthedocs.io/en/stable/logging.html

Sets the logging level to DEBUG for the 'schedule' logger to capture internal library events.

```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()
```

--------------------------------

### Define day-of-week properties

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Properties for scheduling jobs on specific days of the week, restricted to weekly intervals.

```python
    @property
    def monday(self):
        if self.interval != 1:
            raise IntervalError(
                "Scheduling .monday() jobs is only allowed for weekly jobs. "
                "Using .monday() on a job scheduled to run every 2 or more weeks "
                "is not supported."
            )
        self.start_day = "monday"
        return self.weeks

    @property
    def tuesday(self):
        if self.interval != 1:
            raise IntervalError(
                "Scheduling .tuesday() jobs is only allowed for weekly jobs. "
                "Using .tuesday() on a job scheduled to run every 2 or more weeks "
                "is not supported."
            )
        self.start_day = "tuesday"
        return self.weeks

    @property
    def wednesday(self):
        if self.interval != 1:
            raise IntervalError(
                "Scheduling .wednesday() jobs is only allowed for weekly jobs. "
                "Using .wednesday() on a job scheduled to run every 2 or more weeks "
                "is not supported."
            )
        self.start_day = "wednesday"
        return self.weeks

    @property
    def thursday(self):
        if self.interval != 1:
            raise IntervalError(
                "Scheduling .thursday() jobs is only allowed for weekly jobs. "
                "Using .thursday() on a job scheduled to run every 2 or more weeks "
                "is not supported."
            )
        self.start_day = "thursday"
        return self.weeks

    @property
    def friday(self):
        if self.interval != 1:
            raise IntervalError(
                "Scheduling .friday() jobs is only allowed for weekly jobs. "
                "Using .friday() on a job scheduled to run every 2 or more weeks "
                "is not supported."
            )
        self.start_day = "friday"
        return self.weeks

    @property
    def saturday(self):
        if self.interval != 1:
            raise IntervalError(
                "Scheduling .saturday() jobs is only allowed for weekly jobs. "
                "Using .saturday() on a job scheduled to run every 2 or more weeks "
                "is not supported."
            )
        self.start_day = "saturday"
        return self.weeks

    @property
    def sunday(self):
        if self.interval != 1:
            raise IntervalError(
                "Scheduling .sunday() jobs is only allowed for weekly jobs. "
                "Using .sunday() on a job scheduled to run every 2 or more weeks "
                "is not supported."
            )
        self.start_day = "sunday"
        return self.weeks
```

--------------------------------

### Troubleshoot AttributeError

Source: https://schedule.readthedocs.io/en/stable/faq.html

Displays the error message encountered when a local file named schedule.py conflicts with the library.

```text
AttributeError: 'module' object has no attribute 'every'
```

--------------------------------

### schedule.run_pending()

Source: https://schedule.readthedocs.io/en/stable/reference.html

Executes all jobs that are scheduled to run at the current time.

```APIDOC
## schedule.run_pending()

### Description
Run all jobs that are scheduled to run. Note that this does not run missed jobs.
```

--------------------------------

### Tag a job

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Method to assign unique, hashable tags to a job.

```python
    [docs]
    def tag(self, *tags: Hashable):
        """
        Tags the job with one or more unique identifiers.

        Tags must be hashable. Duplicate tags are discarded.

```

--------------------------------

### until(until_time)

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Sets a deadline for the job. If the deadline is in the past, a ScheduleValueError is raised.

```APIDOC
## until(until_time)

### Description
Specifies the latest time a job can be run. If only a time is provided, the date defaults to today.

### Parameters
- **until_time** (datetime.datetime | datetime.timedelta | datetime.time | str) - Required - The deadline for the job. String formats accepted: "%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d", "%H:%M:%S", "%H:%M".

### Returns
- **job** (instance) - The invoked job instance.
```

--------------------------------

### Implement Exception Handling Decorator

Source: https://schedule.readthedocs.io/en/stable/exception-handling.html

Wraps job functions to catch exceptions, print stack traces, and optionally return schedule.CancelJob to stop future executions.

```python
import functools

def catch_exceptions(cancel_on_failure=False):
    def catch_exceptions_decorator(job_func):
        @functools.wraps(job_func)
        def wrapper(*args, **kwargs):
            try:
                return job_func(*args, **kwargs)
            except:
                import traceback
                print(traceback.format_exc())
                if cancel_on_failure:
                    return schedule.CancelJob
        return wrapper
    return catch_exceptions_decorator

@catch_exceptions(cancel_on_failure=True)
def bad_task():
    return 1 / 0

schedule.every(5).minutes.do(bad_task)
```

--------------------------------

### Setting Job Expiration

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Defines a deadline for job execution, after which the job will be canceled.

```python
    def until(
        self,
        until_time: Union[datetime.datetime, datetime.timedelta, datetime.time, str],
    ):
        """
        Schedule job to run until the specified moment.

        The job is canceled whenever the next run is calculated and it turns out the
        next run is after the until_time. The job is also canceled right before it runs,
        if the current time is after until_time. This latter case can happen when the
        the job was scheduled to run before until_time, but runs after until_time.
```

--------------------------------

### Schedule periodic jobs with a decorator

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Uses a decorator to register a function as a periodic job on a provided job object.

```python
def repeat(job, *args, **kwargs):
    """
    Decorator to schedule a new periodic job.

    Any additional arguments are passed on to the decorated function
    when the job runs.

    :param job: a :class:`Jobs <Job>`
    """

    def _schedule_decorator(decorated_function):
        job.do(decorated_function, *args, **kwargs)
        return decorated_function

    return _schedule_decorator
```

--------------------------------

### Schedule Periodic Jobs

Source: https://schedule.readthedocs.io/en/stable/index.html

Define and run periodic tasks using various time intervals and constraints, executed within a loop.

```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)
```

--------------------------------

### Retrieve scheduled jobs

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Fetches a list of jobs from the default scheduler, optionally filtered by a tag.

```python
def get_jobs(tag: Optional[Hashable] = None) -> List[Job]:
    """Calls :meth:`get_jobs <Scheduler.get_jobs>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.get_jobs(tag)
```

--------------------------------

### Job.tag(*tags)

Source: https://schedule.readthedocs.io/en/stable/reference.html

Adds tags to a job instance for identification and filtering.

```APIDOC
## Job.tag(*tags)

### Description
Tags the job with one or more unique identifiers. Tags must be hashable.

### Parameters
- **tags** (Hashable) - Required - A unique list of Hashable tags.
```

--------------------------------

### Run a job once

Source: https://schedule.readthedocs.io/en/stable/examples.html

Executes a job exactly once by returning schedule.CancelJob from the task function.

```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)
```

--------------------------------

### Expected Log Output

Source: https://schedule.readthedocs.io/en/stable/logging.html

The output generated by the library when DEBUG logging is enabled.

```text
DEBUG:schedule:Running *all* 1 jobs with 0s delay in between
DEBUG:schedule:Running job Job(interval=1, unit=seconds, do=job, args=(), kwargs={})
Hello, Logs
DEBUG:schedule:Deleting *all* jobs
```

--------------------------------

### Scheduler.get_next_run

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Retrieves the datetime when the next job is scheduled to run, optionally filtered by a tag.

```APIDOC
## Scheduler.get_next_run(tag=None)

### Description
Returns the datetime object representing when the next job should run. If no jobs are scheduled, it returns None.

### Parameters
#### Query Parameters
- **tag** (Hashable) - Optional - Filter the next run for the given tag parameter.

### Response
- **datetime.datetime** - The scheduled time for the next job or None.
```

--------------------------------

### Compute next run time

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Calculates the next execution time based on the job unit.

```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`)"
            )
```

--------------------------------

### Define time unit properties

Source: https://schedule.readthedocs.io/en/stable/_modules/schedule.html

Properties for setting the time unit of a job, raising an IntervalError if the interval is not 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
```

--------------------------------

### Custom Logging Output

Source: https://schedule.readthedocs.io/en/stable/logging.html

The output produced when using the custom elapsed time decorator.

```text
LOG: Running job "job"
Hello, Logs
LOG: Job "job" completed in 5 seconds
```

--------------------------------

### Set job deadlines with until

Source: https://schedule.readthedocs.io/en/stable/examples.html

The until method accepts strings, datetime objects, time objects, or timedeltas to define when a job should stop running.

```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)

# Run my_job until today 11:33:42
schedule.every(1).hours.until(time(11, 33, 42)).do(job)

# run job until a specific datetime
schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job)
```

--------------------------------

### Execute jobs in parallel using threading

Source: https://schedule.readthedocs.io/en/stable/parallel-execution.html

Run each job in its own thread to bypass the default serial execution behavior.

```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)
```
Tokens:
Raw