Исходный код Python для изучения расписания

Python
Исходный код Python для изучения расписания

обо мне
Небольшой программист в мире программирования, в настоящее время работает руководителем команды в предпринимательской команде.Стек технологий включает Android, Python, Java и Go, который также является основным стеком технологий нашей команды. Контакт: hylinux1024@gmail.com

Предыдущийвводит простойPythonИспользование планировщика, а то я пролистал егоисходный код, я был удивлен, обнаружив, что основная библиотека представляет собой всего один файл, а объем кода составляет менее 700 строк. Это отличный учебный материал.
Я был вне себя от радости, что автор этой библиотеки является автором недавно прочитанной книги "Python Tricks"! Теперь давайте посмотрим на реализацию Великого Бога.

0x00 готов

адрес проекта

GitHub.com/много людей/садись в машину…

кодcheckoutк местному

окружающая обстановка

PyCharm+venv+Python3

0x01 использование

Это также представлено в предыдущей статье, очень просто

import schedule

# 定义需要执行的方法
def job():
    print("a simple scheduler in python.")

# 设置调度的参数，这里是每2秒执行一次
schedule.every(2).seconds.do(job)

if __name__ == '__main__':
    while True:
        schedule.run_pending()

# 执行结果
a simple scheduler in python.
a simple scheduler in python.
a simple scheduler in python.
...

Документация этой библиотеки также очень подробная, вы можете просмотретьschedule.readthedocs.io/Понимание общего использования библиотеки

0x02 Структура проекта

(venv) ➜  schedule git:(master) tree -L 2
.
...
├── requirements-dev.txt
├── schedule
│   └── __init__.py
├── setup.py
├── test_schedule.py
├── tox.ini
└── venv
    ├── bin
    ├── include
    ├── lib
    ├── pip-selfcheck.json
    └── pyvenv.cfg

8 directories, 18 files
  • scheduleодин в каталоге__init__.pyфайл, здесь нам нужно сосредоточиться на обучении.
  • setup.pyфайл - это файл конфигурации для опубликованного проекта
  • test_schedule.pyЭто файл модульного теста.В дополнение к чтению документации в начале вы также можете начать с модульного теста, чтобы понять использование этой библиотеки.
  • requirements-dev.txtФайлы библиотеки зависимостей среды разработки, если основная библиотека не требует сторонних зависимостей, но требует модульного тестирования
  • venvэто яcheckoutСоздан позже, исходный проект не существует

0x03 schedule

мы знаем__init__.pyэто определениеPythonПакет необходимых файлов. Методы и классы, определенные в этом файле, можно использовать вimportкоманду в проект, а затем использовать.

исходный код расписания

Ниже приведеныscheduleМодули, которые будут использоваться,Pythonвнутренние модули.

import collections
import datetime
import functools
import logging
import random
import re
import time

logger = logging.getLogger('schedule')

Затем определите экземпляр средства печати журнала.

Далее идет система структуры, которая определяет три класса исключений модуля, которые определяютсяExceptionпроизводные, соответственноScheduleError,ScheduleValueErrorа такжеIntervalError

class ScheduleError(Exception):
    """Base schedule exception"""
    pass

class ScheduleValueError(ScheduleError):
    """Base schedule value error"""
    pass

class IntervalError(ScheduleValueError):
    """An improper interval was used"""
    pass

также определяетCancelJobКласс, используемый для отмены продолжения планировщика

class CancelJob(object):
    """
    Can be returned from a job to unschedule itself.
    """
    pass

Например, верните это в пользовательском методе, который необходимо отправить.CancelJobКлассы могут реализовывать разовые задачи

# 定义需要执行的方法
def job():
    print("a simple scheduler in python.")
    # 返回CancelJob可以停止调度器的后续执行
    return schedule.CancelJob

Затем есть два основных класса этой библиотекиSchedulerа такжеJob.

class Scheduler(object):
    """
    Objects instantiated by the :class:`Scheduler <Scheduler>` are
    factories to create jobs, keep record of scheduled jobs and
    handle their execution.
    """
    
class Job(object):
    """
    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`.
    """

Scheduler— это класс реализации планировщика, который отвечает за планирование задач (job) создание и исполнение.

JobЭто абстракция задачи, которую необходимо выполнить.

Эти два класса являются ядром этой библиотеки, и позже мы увидим их подробный анализ.
Далее идет планировщик по умолчаниюdefault_schedulerи список задачjobsсоздание.

# The following methods are shortcuts for not having to
# create a Scheduler instance:

#: 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()?

в исполненииimport scheduleПосле этого он создается по умолчаниюdefault_scheduler. а такжеSchedulerМетод строительства

def __init__(self):
    self.jobs = []

При выполнении инициализации планировщик создает пустой список задач.

В конце файла определены некоторые связанные методы вызова, которые также очень удобны для пользователя и заслуживают изучения. Все методы здесь определены в модуле, и все они инкапсулированы.default_schedulerвызов экземпляра.

def every(interval=1):
    """Calls :meth:`every <Scheduler.every>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.every(interval)


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


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


def clear(tag=None):
    """Calls :meth:`clear <Scheduler.clear>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    default_scheduler.clear(tag)


def cancel_job(job):
    """Calls :meth:`cancel_job <Scheduler.cancel_job>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    default_scheduler.cancel_job(job)


def next_run():
    """Calls :meth:`next_run <Scheduler.next_run>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.next_run


def idle_seconds():
    """Calls :meth:`idle_seconds <Scheduler.idle_seconds>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.idle_seconds

Давайте посмотрим на метод входаrun_pending(), с начала этой статьиDemoВы можете знать, что это метод запуска планировщика. Здесь он выполняетсяdefault_schedulerметод в .

default_scheduler.run_pending()

Итак, мы нацелились наSchedulerсоответствующий метод класса

def run_pending(self):
    """
    Run all jobs that are scheduled to run.

    Please note that it is *intended behavior that run_pending()
    does not run missed jobs*. For example, if you've registered a job
    that should run every minute and you only call run_pending()
    in one hour increments then your job won't be run 60 times in
    between but only once.
    """
    runnable_jobs = (job for job in self.jobs if job.should_run)
    for job in sorted(runnable_jobs):
        self._run_job(job)

В этом методе первыйjobsСписок фильтрует задачи, которые необходимо выполнить, и помещает их вrunnable_jobslist, затем отсортируйте его и выполните внутреннюю_run_job(job)метод

def _run_job(self, job):
    ret = job.run()
    if isinstance(ret, CancelJob) or ret is CancelJob:
        self.cancel_job(job)

существует_run_jobметод называетсяjobв классеrunметод и решить, нужно ли отменить задачу, в соответствии с возвращаемым значением.

Теперь мы увидимJobЛогика реализации класса.

Сначала мы смотрим наJobКогда он был создан. или изDemoНачните с кода в

schedule.every(2).seconds.do(job)

Выполнено здесьschedule.every()метод

def every(interval=1):
    """Calls :meth:`every <Scheduler.every>` on the
    :data:`default scheduler instance <default_scheduler>`.
    """
    return default_scheduler.every(interval)

Этот методschedulerв классеeveryметод

def every(self, interval=1):
    """
    Schedule a new periodic job.

    :param interval: A quantity of a certain time unit
    :return: An unconfigured :class:`Job <Job>`
    """
    job = Job(interval, self)
    return job

Создал задачу здесьjob, и установите параметрintervalа такжеschedulerЭкземпляр передается в конструктор и, наконец, возвращаетсяjobЭкземпляры используются для реализации связанных вызовов.

перенаправить наJobметод строительства

def __init__(self, interval, scheduler=None):
    self.interval = interval  # pause interval * unit between runs
    self.latest = None  # upper limit to the interval
    self.job_func = None  # the job job_func to run
    self.unit = None  # time units, e.g. 'minutes', 'hours', ...
    self.at_time = None  # optional time at which this job runs
    self.last_run = None  # datetime of the last run
    self.next_run = None  # datetime of the next run
    self.period = None  # timedelta between runs, only valid for
    self.start_day = None  # Specific day of the week to start on
    self.tags = set()  # unique set of tags for the job
    self.scheduler = scheduler  # scheduler to register with

В основном он инициализирует конфигурацию интервала времени, методы, которые должны выполняться, различные единицы времени планировщика и т. д.

воплощать в жизньeveryметод вызывается послеsecondsэтот метод свойства

@property
def seconds(self):
    self.unit = 'seconds'
    return self

Единица времени установлена, это устанавливает секунды, и, конечно же, есть другие подобные методы свойств.minutes,hours,daysи т.п.

Наконец он выполняетсяdoметод

def do(self, job_func, *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)
    try:
        functools.update_wrapper(self.job_func, job_func)
    except AttributeError:
        # job_funcs already wrapped by functools.partial won't have
        # __name__, __module__ or __doc__ and the update_wrapper()
        # call will fail.
        pass
    self._schedule_next_run()
    self.scheduler.jobs.append(self)
    return self

использовать здесьfunctoolsЧастичные функции в инструментахpartialИнкапсулируйте наш пользовательский метод в вызываемый объект

тогда позвони_schedule_next_runметод, это в основном анализ времени, в зависимости от времениjobСортировка, я думаю, что этот метод является техническим моментом в этом проекте, и логика немного сложна, вы можете понять это, если внимательно прочитаете, в основном из-за времени.datetimeиспользование. Из-за недостатка места код не будет размещен здесь.

Миссия выполнена здесьjobдополнение. затем звонюrun_pendingспособ выполнения задачи.

0x04 Суммировать

scheduleБиблиотека определяет два основных классаSchedulerа такжеJob. Пакет создается по умолчанию при импорте пакетаSchedulerобъект и инициализировать список задач.
scheduleМодуль предоставляет интерфейс для объединения вызовов, в конфигурацииscheduleпараметр, создается объект задачиjob, и воляjobдобавлен в список задач и окончательно выполненrun_pendingбудет вызван наш пользовательский метод. Основная идея этой библиотеки состоит в том, чтобы использовать объектно-ориентированные методы для точного абстрагирования.Его общая логика не сложна, и это хороший пример изучения исходного кода.

0x05 Учебные материалы

Категории