periodic server maintanance task added. confirmed config messages and data...

periodic server maintanance task added. confirmed config messages and data files are removed during maintanance task

env/lib/python3.5/site-packages/apscheduler/jobstores/memory.py

+from __future__ import absolute_import
+
+from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
+from apscheduler.util import datetime_to_utc_timestamp
+
+
+class MemoryJobStore(BaseJobStore):
+    """
+    Stores jobs in an array in RAM. Provides no persistence support.
+
+    Plugin alias: ``memory``
+    """
+
+    def __init__(self):
+        super(MemoryJobStore, self).__init__()
+        # list of (job, timestamp), sorted by next_run_time and job id (ascending)
+        self._jobs = []
+        self._jobs_index = {}  # id -> (job, timestamp) lookup table
+
+    def lookup_job(self, job_id):
+        return self._jobs_index.get(job_id, (None, None))[0]
+
+    def get_due_jobs(self, now):
+        now_timestamp = datetime_to_utc_timestamp(now)
+        pending = []
+        for job, timestamp in self._jobs:
+            if timestamp is None or timestamp > now_timestamp:
+                break
+            pending.append(job)
+
+        return pending
+
+    def get_next_run_time(self):
+        return self._jobs[0][0].next_run_time if self._jobs else None
+
+    def get_all_jobs(self):
+        return [j[0] for j in self._jobs]
+
+    def add_job(self, job):
+        if job.id in self._jobs_index:
+            raise ConflictingIdError(job.id)
+
+        timestamp = datetime_to_utc_timestamp(job.next_run_time)
+        index = self._get_job_index(timestamp, job.id)
+        self._jobs.insert(index, (job, timestamp))
+        self._jobs_index[job.id] = (job, timestamp)
+
+    def update_job(self, job):
+        old_job, old_timestamp = self._jobs_index.get(job.id, (None, None))
+        if old_job is None:
+            raise JobLookupError(job.id)
+
+        # If the next run time has not changed, simply replace the job in its present index.
+        # Otherwise, reinsert the job to the list to preserve the ordering.
+        old_index = self._get_job_index(old_timestamp, old_job.id)
+        new_timestamp = datetime_to_utc_timestamp(job.next_run_time)
+        if old_timestamp == new_timestamp:
+            self._jobs[old_index] = (job, new_timestamp)
+        else:
+            del self._jobs[old_index]
+            new_index = self._get_job_index(new_timestamp, job.id)
+            self._jobs.insert(new_index, (job, new_timestamp))
+
+        self._jobs_index[old_job.id] = (job, new_timestamp)
+
+    def remove_job(self, job_id):
+        job, timestamp = self._jobs_index.get(job_id, (None, None))
+        if job is None:
+            raise JobLookupError(job_id)
+
+        index = self._get_job_index(timestamp, job_id)
+        del self._jobs[index]
+        del self._jobs_index[job.id]
+
+    def remove_all_jobs(self):
+        self._jobs = []
+        self._jobs_index = {}
+
+    def shutdown(self):
+        self.remove_all_jobs()
+
+    def _get_job_index(self, timestamp, job_id):
+        """
+        Returns the index of the given job, or if it's not found, the index where the job should be
+        inserted based on the given timestamp.
+
+        :type timestamp: int
+        :type job_id: str
+
+        """
+        lo, hi = 0, len(self._jobs)
+        timestamp = float('inf') if timestamp is None else timestamp
+        while lo < hi:
+            mid = (lo + hi) // 2
+            mid_job, mid_timestamp = self._jobs[mid]
+            mid_timestamp = float('inf') if mid_timestamp is None else mid_timestamp
+            if mid_timestamp > timestamp:
+                hi = mid
+            elif mid_timestamp < timestamp:
+                lo = mid + 1
+            elif mid_job.id > job_id:
+                hi = mid
+            elif mid_job.id < job_id:
+                lo = mid + 1
+            else:
+                return mid
+
+        return lo

env/lib/python3.5/site-packages/apscheduler/jobstores/mongodb.py

+from __future__ import absolute_import
+import warnings
+
+from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
+from apscheduler.util import maybe_ref, datetime_to_utc_timestamp, utc_timestamp_to_datetime
+from apscheduler.job import Job
+
+try:
+    import cPickle as pickle
+except ImportError:  # pragma: nocover
+    import pickle
+
+try:
+    from bson.binary import Binary
+    from pymongo.errors import DuplicateKeyError
+    from pymongo import MongoClient, ASCENDING
+except ImportError:  # pragma: nocover
+    raise ImportError('MongoDBJobStore requires PyMongo installed')
+
+
+class MongoDBJobStore(BaseJobStore):
+    """
+    Stores jobs in a MongoDB database. Any leftover keyword arguments are directly passed to
+    pymongo's `MongoClient
+    <http://api.mongodb.org/python/current/api/pymongo/mongo_client.html#pymongo.mongo_client.MongoClient>`_.
+
+    Plugin alias: ``mongodb``
+
+    :param str database: database to store jobs in
+    :param str collection: collection to store jobs in
+    :param client: a :class:`~pymongo.mongo_client.MongoClient` instance to use instead of
+        providing connection arguments
+    :param int pickle_protocol: pickle protocol level to use (for serialization), defaults to the
+        highest available
+    """
+
+    def __init__(self, database='apscheduler', collection='jobs', client=None,
+                 pickle_protocol=pickle.HIGHEST_PROTOCOL, **connect_args):
+        super(MongoDBJobStore, self).__init__()
+        self.pickle_protocol = pickle_protocol
+
+        if not database:
+            raise ValueError('The "database" parameter must not be empty')
+        if not collection:
+            raise ValueError('The "collection" parameter must not be empty')
+
+        if client:
+            self.client = maybe_ref(client)
+        else:
+            connect_args.setdefault('w', 1)
+            self.client = MongoClient(**connect_args)
+
+        self.collection = self.client[database][collection]
+
+    def start(self, scheduler, alias):
+        super(MongoDBJobStore, self).start(scheduler, alias)
+        self.collection.ensure_index('next_run_time', sparse=True)
+
+    @property
+    def connection(self):
+        warnings.warn('The "connection" member is deprecated -- use "client" instead',
+                      DeprecationWarning)
+        return self.client
+
+    def lookup_job(self, job_id):
+        document = self.collection.find_one(job_id, ['job_state'])
+        return self._reconstitute_job(document['job_state']) if document else None
+
+    def get_due_jobs(self, now):
+        timestamp = datetime_to_utc_timestamp(now)
+        return self._get_jobs({'next_run_time': {'$lte': timestamp}})
+
+    def get_next_run_time(self):
+        document = self.collection.find_one({'next_run_time': {'$ne': None}},
+                                            projection=['next_run_time'],
+                                            sort=[('next_run_time', ASCENDING)])
+        return utc_timestamp_to_datetime(document['next_run_time']) if document else None
+
+    def get_all_jobs(self):
+        jobs = self._get_jobs({})
+        self._fix_paused_jobs_sorting(jobs)
+        return jobs
+
+    def add_job(self, job):
+        try:
+            self.collection.insert({
+                '_id': job.id,
+                'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+                'job_state': Binary(pickle.dumps(job.__getstate__(), self.pickle_protocol))
+            })
+        except DuplicateKeyError:
+            raise ConflictingIdError(job.id)
+
+    def update_job(self, job):
+        changes = {
+            'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+            'job_state': Binary(pickle.dumps(job.__getstate__(), self.pickle_protocol))
+        }
+        result = self.collection.update({'_id': job.id}, {'$set': changes})
+        if result and result['n'] == 0:
+            raise JobLookupError(job.id)
+
+    def remove_job(self, job_id):
+        result = self.collection.remove(job_id)
+        if result and result['n'] == 0:
+            raise JobLookupError(job_id)
+
+    def remove_all_jobs(self):
+        self.collection.remove()
+
+    def shutdown(self):
+        self.client.close()
+
+    def _reconstitute_job(self, job_state):
+        job_state = pickle.loads(job_state)
+        job = Job.__new__(Job)
+        job.__setstate__(job_state)
+        job._scheduler = self._scheduler
+        job._jobstore_alias = self._alias
+        return job
+
+    def _get_jobs(self, conditions):
+        jobs = []
+        failed_job_ids = []
+        for document in self.collection.find(conditions, ['_id', 'job_state'],
+                                             sort=[('next_run_time', ASCENDING)]):
+            try:
+                jobs.append(self._reconstitute_job(document['job_state']))
+            except BaseException:
+                self._logger.exception('Unable to restore job "%s" -- removing it',
+                                       document['_id'])
+                failed_job_ids.append(document['_id'])
+
+        # Remove all the jobs we failed to restore
+        if failed_job_ids:
+            self.collection.remove({'_id': {'$in': failed_job_ids}})
+
+        return jobs
+
+    def __repr__(self):
+        return '<%s (client=%s)>' % (self.__class__.__name__, self.client)

env/lib/python3.5/site-packages/apscheduler/jobstores/redis.py

+from __future__ import absolute_import
+from datetime import datetime
+
+from pytz import utc
+import six
+
+from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
+from apscheduler.util import datetime_to_utc_timestamp, utc_timestamp_to_datetime
+from apscheduler.job import Job
+
+try:
+    import cPickle as pickle
+except ImportError:  # pragma: nocover
+    import pickle
+
+try:
+    from redis import Redis
+except ImportError:  # pragma: nocover
+    raise ImportError('RedisJobStore requires redis installed')
+
+
+class RedisJobStore(BaseJobStore):
+    """
+    Stores jobs in a Redis database. Any leftover keyword arguments are directly passed to redis's
+    :class:`~redis.StrictRedis`.
+
+    Plugin alias: ``redis``
+
+    :param int db: the database number to store jobs in
+    :param str jobs_key: key to store jobs in
+    :param str run_times_key: key to store the jobs' run times in
+    :param int pickle_protocol: pickle protocol level to use (for serialization), defaults to the
+        highest available
+    """
+
+    def __init__(self, db=0, jobs_key='apscheduler.jobs', run_times_key='apscheduler.run_times',
+                 pickle_protocol=pickle.HIGHEST_PROTOCOL, **connect_args):
+        super(RedisJobStore, self).__init__()
+
+        if db is None:
+            raise ValueError('The "db" parameter must not be empty')
+        if not jobs_key:
+            raise ValueError('The "jobs_key" parameter must not be empty')
+        if not run_times_key:
+            raise ValueError('The "run_times_key" parameter must not be empty')
+
+        self.pickle_protocol = pickle_protocol
+        self.jobs_key = jobs_key
+        self.run_times_key = run_times_key
+        self.redis = Redis(db=int(db), **connect_args)
+
+    def lookup_job(self, job_id):
+        job_state = self.redis.hget(self.jobs_key, job_id)
+        return self._reconstitute_job(job_state) if job_state else None
+
+    def get_due_jobs(self, now):
+        timestamp = datetime_to_utc_timestamp(now)
+        job_ids = self.redis.zrangebyscore(self.run_times_key, 0, timestamp)
+        if job_ids:
+            job_states = self.redis.hmget(self.jobs_key, *job_ids)
+            return self._reconstitute_jobs(six.moves.zip(job_ids, job_states))
+        return []
+
+    def get_next_run_time(self):
+        next_run_time = self.redis.zrange(self.run_times_key, 0, 0, withscores=True)
+        if next_run_time:
+            return utc_timestamp_to_datetime(next_run_time[0][1])
+
+    def get_all_jobs(self):
+        job_states = self.redis.hgetall(self.jobs_key)
+        jobs = self._reconstitute_jobs(six.iteritems(job_states))
+        paused_sort_key = datetime(9999, 12, 31, tzinfo=utc)
+        return sorted(jobs, key=lambda job: job.next_run_time or paused_sort_key)
+
+    def add_job(self, job):
+        if self.redis.hexists(self.jobs_key, job.id):
+            raise ConflictingIdError(job.id)
+
+        with self.redis.pipeline() as pipe:
+            pipe.multi()
+            pipe.hset(self.jobs_key, job.id, pickle.dumps(job.__getstate__(),
+                                                          self.pickle_protocol))
+            if job.next_run_time:
+                pipe.zadd(self.run_times_key,
+                          {job.id: datetime_to_utc_timestamp(job.next_run_time)})
+
+            pipe.execute()
+
+    def update_job(self, job):
+        if not self.redis.hexists(self.jobs_key, job.id):
+            raise JobLookupError(job.id)
+
+        with self.redis.pipeline() as pipe:
+            pipe.hset(self.jobs_key, job.id, pickle.dumps(job.__getstate__(),
+                                                          self.pickle_protocol))
+            if job.next_run_time:
+                pipe.zadd(self.run_times_key,
+                          {job.id: datetime_to_utc_timestamp(job.next_run_time)})
+            else:
+                pipe.zrem(self.run_times_key, job.id)
+
+            pipe.execute()
+
+    def remove_job(self, job_id):
+        if not self.redis.hexists(self.jobs_key, job_id):
+            raise JobLookupError(job_id)
+
+        with self.redis.pipeline() as pipe:
+            pipe.hdel(self.jobs_key, job_id)
+            pipe.zrem(self.run_times_key, job_id)
+            pipe.execute()
+
+    def remove_all_jobs(self):
+        with self.redis.pipeline() as pipe:
+            pipe.delete(self.jobs_key)
+            pipe.delete(self.run_times_key)
+            pipe.execute()
+
+    def shutdown(self):
+        self.redis.connection_pool.disconnect()
+
+    def _reconstitute_job(self, job_state):
+        job_state = pickle.loads(job_state)
+        job = Job.__new__(Job)
+        job.__setstate__(job_state)
+        job._scheduler = self._scheduler
+        job._jobstore_alias = self._alias
+        return job
+
+    def _reconstitute_jobs(self, job_states):
+        jobs = []
+        failed_job_ids = []
+        for job_id, job_state in job_states:
+            try:
+                jobs.append(self._reconstitute_job(job_state))
+            except BaseException:
+                self._logger.exception('Unable to restore job "%s" -- removing it', job_id)
+                failed_job_ids.append(job_id)
+
+        # Remove all the jobs we failed to restore
+        if failed_job_ids:
+            with self.redis.pipeline() as pipe:
+                pipe.hdel(self.jobs_key, *failed_job_ids)
+                pipe.zrem(self.run_times_key, *failed_job_ids)
+                pipe.execute()
+
+        return jobs
+
+    def __repr__(self):
+        return '<%s>' % self.__class__.__name__

env/lib/python3.5/site-packages/apscheduler/jobstores/rethinkdb.py

+from __future__ import absolute_import
+
+from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
+from apscheduler.util import maybe_ref, datetime_to_utc_timestamp, utc_timestamp_to_datetime
+from apscheduler.job import Job
+
+try:
+    import cPickle as pickle
+except ImportError:  # pragma: nocover
+    import pickle
+
+try:
+    from rethinkdb import RethinkDB
+except ImportError:  # pragma: nocover
+    raise ImportError('RethinkDBJobStore requires rethinkdb installed')
+
+
+class RethinkDBJobStore(BaseJobStore):
+    """
+    Stores jobs in a RethinkDB database. Any leftover keyword arguments are directly passed to
+    rethinkdb's `RethinkdbClient <http://www.rethinkdb.com/api/#connect>`_.
+
+    Plugin alias: ``rethinkdb``
+
+    :param str database: database to store jobs in
+    :param str collection: collection to store jobs in
+    :param client: a :class:`rethinkdb.net.Connection` instance to use instead of providing
+        connection arguments
+    :param int pickle_protocol: pickle protocol level to use (for serialization), defaults to the
+        highest available
+    """
+
+    def __init__(self, database='apscheduler', table='jobs', client=None,
+                 pickle_protocol=pickle.HIGHEST_PROTOCOL, **connect_args):
+        super(RethinkDBJobStore, self).__init__()
+
+        if not database:
+            raise ValueError('The "database" parameter must not be empty')
+        if not table:
+            raise ValueError('The "table" parameter must not be empty')
+
+        self.database = database
+        self.table_name = table
+        self.table = None
+        self.client = client
+        self.pickle_protocol = pickle_protocol
+        self.connect_args = connect_args
+        self.r = RethinkDB()
+        self.conn = None
+
+    def start(self, scheduler, alias):
+        super(RethinkDBJobStore, self).start(scheduler, alias)
+
+        if self.client:
+            self.conn = maybe_ref(self.client)
+        else:
+            self.conn = self.r.connect(db=self.database, **self.connect_args)
+
+        if self.database not in self.r.db_list().run(self.conn):
+            self.r.db_create(self.database).run(self.conn)
+
+        if self.table_name not in self.r.table_list().run(self.conn):
+            self.r.table_create(self.table_name).run(self.conn)
+
+        if 'next_run_time' not in self.r.table(self.table_name).index_list().run(self.conn):
+            self.r.table(self.table_name).index_create('next_run_time').run(self.conn)
+
+        self.table = self.r.db(self.database).table(self.table_name)
+
+    def lookup_job(self, job_id):
+        results = list(self.table.get_all(job_id).pluck('job_state').run(self.conn))
+        return self._reconstitute_job(results[0]['job_state']) if results else None
+
+    def get_due_jobs(self, now):
+        return self._get_jobs(self.r.row['next_run_time'] <= datetime_to_utc_timestamp(now))
+
+    def get_next_run_time(self):
+        results = list(
+            self.table
+            .filter(self.r.row['next_run_time'] != None)  # noqa
+            .order_by(self.r.asc('next_run_time'))
+            .map(lambda x: x['next_run_time'])
+            .limit(1)
+            .run(self.conn)
+        )
+        return utc_timestamp_to_datetime(results[0]) if results else None
+
+    def get_all_jobs(self):
+        jobs = self._get_jobs()
+        self._fix_paused_jobs_sorting(jobs)
+        return jobs
+
+    def add_job(self, job):
+        job_dict = {
+            'id': job.id,
+            'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+            'job_state': self.r.binary(pickle.dumps(job.__getstate__(), self.pickle_protocol))
+        }
+        results = self.table.insert(job_dict).run(self.conn)
+        if results['errors'] > 0:
+            raise ConflictingIdError(job.id)
+
+    def update_job(self, job):
+        changes = {
+            'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+            'job_state': self.r.binary(pickle.dumps(job.__getstate__(), self.pickle_protocol))
+        }
+        results = self.table.get_all(job.id).update(changes).run(self.conn)
+        skipped = False in map(lambda x: results[x] == 0, results.keys())
+        if results['skipped'] > 0 or results['errors'] > 0 or not skipped:
+            raise JobLookupError(job.id)
+
+    def remove_job(self, job_id):
+        results = self.table.get_all(job_id).delete().run(self.conn)
+        if results['deleted'] + results['skipped'] != 1:
+            raise JobLookupError(job_id)
+
+    def remove_all_jobs(self):
+        self.table.delete().run(self.conn)
+
+    def shutdown(self):
+        self.conn.close()
+
+    def _reconstitute_job(self, job_state):
+        job_state = pickle.loads(job_state)
+        job = Job.__new__(Job)
+        job.__setstate__(job_state)
+        job._scheduler = self._scheduler
+        job._jobstore_alias = self._alias
+        return job
+
+    def _get_jobs(self, predicate=None):
+        jobs = []
+        failed_job_ids = []
+        query = (self.table.filter(self.r.row['next_run_time'] != None).filter(predicate)  # noqa
+                 if predicate else self.table)
+        query = query.order_by('next_run_time', 'id').pluck('id', 'job_state')
+
+        for document in query.run(self.conn):
+            try:
+                jobs.append(self._reconstitute_job(document['job_state']))
+            except Exception:
+                self._logger.exception('Unable to restore job "%s" -- removing it', document['id'])
+                failed_job_ids.append(document['id'])
+
+        # Remove all the jobs we failed to restore
+        if failed_job_ids:
+            self.r.expr(failed_job_ids).for_each(
+                lambda job_id: self.table.get_all(job_id).delete()).run(self.conn)
+
+        return jobs
+
+    def __repr__(self):
+        connection = self.conn
+        return '<%s (connection=%s)>' % (self.__class__.__name__, connection)

env/lib/python3.5/site-packages/apscheduler/jobstores/sqlalchemy.py

+from __future__ import absolute_import
+
+from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
+from apscheduler.util import maybe_ref, datetime_to_utc_timestamp, utc_timestamp_to_datetime
+from apscheduler.job import Job
+
+try:
+    import cPickle as pickle
+except ImportError:  # pragma: nocover
+    import pickle
+
+try:
+    from sqlalchemy import (
+        create_engine, Table, Column, MetaData, Unicode, Float, LargeBinary, select)
+    from sqlalchemy.exc import IntegrityError
+    from sqlalchemy.sql.expression import null
+except ImportError:  # pragma: nocover
+    raise ImportError('SQLAlchemyJobStore requires SQLAlchemy installed')
+
+
+class SQLAlchemyJobStore(BaseJobStore):
+    """
+    Stores jobs in a database table using SQLAlchemy.
+    The table will be created if it doesn't exist in the database.
+
+    Plugin alias: ``sqlalchemy``
+
+    :param str url: connection string (see
+        :ref:`SQLAlchemy documentation <sqlalchemy:database_urls>` on this)
+    :param engine: an SQLAlchemy :class:`~sqlalchemy.engine.Engine` to use instead of creating a
+        new one based on ``url``
+    :param str tablename: name of the table to store jobs in
+    :param metadata: a :class:`~sqlalchemy.schema.MetaData` instance to use instead of creating a
+        new one
+    :param int pickle_protocol: pickle protocol level to use (for serialization), defaults to the
+        highest available
+    :param str tableschema: name of the (existing) schema in the target database where the table
+        should be
+    :param dict engine_options: keyword arguments to :func:`~sqlalchemy.create_engine`
+        (ignored if ``engine`` is given)
+    """
+
+    def __init__(self, url=None, engine=None, tablename='apscheduler_jobs', metadata=None,
+                 pickle_protocol=pickle.HIGHEST_PROTOCOL, tableschema=None, engine_options=None):
+        super(SQLAlchemyJobStore, self).__init__()
+        self.pickle_protocol = pickle_protocol
+        metadata = maybe_ref(metadata) or MetaData()
+
+        if engine:
+            self.engine = maybe_ref(engine)
+        elif url:
+            self.engine = create_engine(url, **(engine_options or {}))
+        else:
+            raise ValueError('Need either "engine" or "url" defined')
+
+        # 191 = max key length in MySQL for InnoDB/utf8mb4 tables,
+        # 25 = precision that translates to an 8-byte float
+        self.jobs_t = Table(
+            tablename, metadata,
+            Column('id', Unicode(191, _warn_on_bytestring=False), primary_key=True),
+            Column('next_run_time', Float(25), index=True),
+            Column('job_state', LargeBinary, nullable=False),
+            schema=tableschema
+        )
+
+    def start(self, scheduler, alias):
+        super(SQLAlchemyJobStore, self).start(scheduler, alias)
+        self.jobs_t.create(self.engine, True)
+
+    def lookup_job(self, job_id):
+        selectable = select([self.jobs_t.c.job_state]).where(self.jobs_t.c.id == job_id)
+        job_state = self.engine.execute(selectable).scalar()
+        return self._reconstitute_job(job_state) if job_state else None
+
+    def get_due_jobs(self, now):
+        timestamp = datetime_to_utc_timestamp(now)
+        return self._get_jobs(self.jobs_t.c.next_run_time <= timestamp)
+
+    def get_next_run_time(self):
+        selectable = select([self.jobs_t.c.next_run_time]).\
+            where(self.jobs_t.c.next_run_time != null()).\
+            order_by(self.jobs_t.c.next_run_time).limit(1)
+        next_run_time = self.engine.execute(selectable).scalar()
+        return utc_timestamp_to_datetime(next_run_time)
+
+    def get_all_jobs(self):
+        jobs = self._get_jobs()
+        self._fix_paused_jobs_sorting(jobs)
+        return jobs
+
+    def add_job(self, job):
+        insert = self.jobs_t.insert().values(**{
+            'id': job.id,
+            'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+            'job_state': pickle.dumps(job.__getstate__(), self.pickle_protocol)
+        })
+        try:
+            self.engine.execute(insert)
+        except IntegrityError:
+            raise ConflictingIdError(job.id)
+
+    def update_job(self, job):
+        update = self.jobs_t.update().values(**{
+            'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+            'job_state': pickle.dumps(job.__getstate__(), self.pickle_protocol)
+        }).where(self.jobs_t.c.id == job.id)
+        result = self.engine.execute(update)
+        if result.rowcount == 0:
+            raise JobLookupError(job.id)
+
+    def remove_job(self, job_id):
+        delete = self.jobs_t.delete().where(self.jobs_t.c.id == job_id)
+        result = self.engine.execute(delete)
+        if result.rowcount == 0:
+            raise JobLookupError(job_id)
+
+    def remove_all_jobs(self):
+        delete = self.jobs_t.delete()
+        self.engine.execute(delete)
+
+    def shutdown(self):
+        self.engine.dispose()
+
+    def _reconstitute_job(self, job_state):
+        job_state = pickle.loads(job_state)
+        job_state['jobstore'] = self
+        job = Job.__new__(Job)
+        job.__setstate__(job_state)
+        job._scheduler = self._scheduler
+        job._jobstore_alias = self._alias
+        return job
+
+    def _get_jobs(self, *conditions):
+        jobs = []
+        selectable = select([self.jobs_t.c.id, self.jobs_t.c.job_state]).\
+            order_by(self.jobs_t.c.next_run_time)
+        selectable = selectable.where(*conditions) if conditions else selectable
+        failed_job_ids = set()
+        for row in self.engine.execute(selectable):
+            try:
+                jobs.append(self._reconstitute_job(row.job_state))
+            except BaseException:
+                self._logger.exception('Unable to restore job "%s" -- removing it', row.id)
+                failed_job_ids.add(row.id)
+
+        # Remove all the jobs we failed to restore
+        if failed_job_ids:
+            delete = self.jobs_t.delete().where(self.jobs_t.c.id.in_(failed_job_ids))
+            self.engine.execute(delete)
+
+        return jobs
+
+    def __repr__(self):
+        return '<%s (url=%s)>' % (self.__class__.__name__, self.engine.url)

env/lib/python3.5/site-packages/apscheduler/jobstores/zookeeper.py

+from __future__ import absolute_import
+
+import os
+from datetime import datetime
+
+from pytz import utc
+from kazoo.exceptions import NoNodeError, NodeExistsError
+
+from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
+from apscheduler.util import maybe_ref, datetime_to_utc_timestamp, utc_timestamp_to_datetime
+from apscheduler.job import Job
+
+try:
+    import cPickle as pickle
+except ImportError:  # pragma: nocover
+    import pickle
+
+try:
+    from kazoo.client import KazooClient
+except ImportError:  # pragma: nocover
+    raise ImportError('ZooKeeperJobStore requires Kazoo installed')
+
+
+class ZooKeeperJobStore(BaseJobStore):
+    """
+    Stores jobs in a ZooKeeper tree. Any leftover keyword arguments are directly passed to
+    kazoo's `KazooClient
+    <http://kazoo.readthedocs.io/en/latest/api/client.html>`_.
+
+    Plugin alias: ``zookeeper``
+
+    :param str path: path to store jobs in
+    :param client: a :class:`~kazoo.client.KazooClient` instance to use instead of
+        providing connection arguments
+    :param int pickle_protocol: pickle protocol level to use (for serialization), defaults to the
+        highest available
+    """
+
+    def __init__(self, path='/apscheduler', client=None, close_connection_on_exit=False,
+                 pickle_protocol=pickle.HIGHEST_PROTOCOL, **connect_args):
+        super(ZooKeeperJobStore, self).__init__()
+        self.pickle_protocol = pickle_protocol
+        self.close_connection_on_exit = close_connection_on_exit
+
+        if not path:
+            raise ValueError('The "path" parameter must not be empty')
+
+        self.path = path
+
+        if client:
+            self.client = maybe_ref(client)
+        else:
+            self.client = KazooClient(**connect_args)
+        self._ensured_path = False
+
+    def _ensure_paths(self):
+        if not self._ensured_path:
+            self.client.ensure_path(self.path)
+        self._ensured_path = True
+
+    def start(self, scheduler, alias):
+        super(ZooKeeperJobStore, self).start(scheduler, alias)
+        if not self.client.connected:
+            self.client.start()
+
+    def lookup_job(self, job_id):
+        self._ensure_paths()
+        node_path = os.path.join(self.path, job_id)
+        try:
+            content, _ = self.client.get(node_path)
+            doc = pickle.loads(content)
+            job = self._reconstitute_job(doc['job_state'])
+            return job
+        except BaseException:
+            return None
+
+    def get_due_jobs(self, now):
+        timestamp = datetime_to_utc_timestamp(now)
+        jobs = [job_def['job'] for job_def in self._get_jobs()
+                if job_def['next_run_time'] is not None and job_def['next_run_time'] <= timestamp]
+        return jobs
+
+    def get_next_run_time(self):
+        next_runs = [job_def['next_run_time'] for job_def in self._get_jobs()
+                     if job_def['next_run_time'] is not None]
+        return utc_timestamp_to_datetime(min(next_runs)) if len(next_runs) > 0 else None
+
+    def get_all_jobs(self):
+        jobs = [job_def['job'] for job_def in self._get_jobs()]
+        self._fix_paused_jobs_sorting(jobs)
+        return jobs
+
+    def add_job(self, job):
+        self._ensure_paths()
+        node_path = os.path.join(self.path,  str(job.id))
+        value = {
+            'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+            'job_state': job.__getstate__()
+        }
+        data = pickle.dumps(value, self.pickle_protocol)
+        try:
+            self.client.create(node_path, value=data)
+        except NodeExistsError:
+            raise ConflictingIdError(job.id)
+
+    def update_job(self, job):
+        self._ensure_paths()
+        node_path = os.path.join(self.path,  str(job.id))
+        changes = {
+            'next_run_time': datetime_to_utc_timestamp(job.next_run_time),
+            'job_state': job.__getstate__()
+        }
+        data = pickle.dumps(changes, self.pickle_protocol)
+        try:
+            self.client.set(node_path, value=data)
+        except NoNodeError:
+            raise JobLookupError(job.id)
+
+    def remove_job(self, job_id):
+        self._ensure_paths()
+        node_path = os.path.join(self.path,  str(job_id))
+        try:
+            self.client.delete(node_path)
+        except NoNodeError:
+            raise JobLookupError(job_id)
+
+    def remove_all_jobs(self):
+        try:
+            self.client.delete(self.path, recursive=True)
+        except NoNodeError:
+            pass
+        self._ensured_path = False
+
+    def shutdown(self):
+        if self.close_connection_on_exit:
+            self.client.stop()
+            self.client.close()
+
+    def _reconstitute_job(self, job_state):
+        job_state = job_state
+        job = Job.__new__(Job)
+        job.__setstate__(job_state)
+        job._scheduler = self._scheduler
+        job._jobstore_alias = self._alias
+        return job
+
+    def _get_jobs(self):
+        self._ensure_paths()
+        jobs = []
+        failed_job_ids = []
+        all_ids = self.client.get_children(self.path)
+        for node_name in all_ids:
+            try:
+                node_path = os.path.join(self.path, node_name)
+                content, _ = self.client.get(node_path)
+                doc = pickle.loads(content)
+                job_def = {
+                    'job_id': node_name,
+                    'next_run_time': doc['next_run_time'] if doc['next_run_time'] else None,
+                    'job_state': doc['job_state'],
+                    'job': self._reconstitute_job(doc['job_state']),
+                    'creation_time': _.ctime
+                }
+                jobs.append(job_def)
+            except BaseException:
+                self._logger.exception('Unable to restore job "%s" -- removing it' % node_name)
+                failed_job_ids.append(node_name)
+
+        # Remove all the jobs we failed to restore
+        if failed_job_ids:
+            for failed_id in failed_job_ids:
+                self.remove_job(failed_id)
+        paused_sort_key = datetime(9999, 12, 31, tzinfo=utc)
+        return sorted(jobs, key=lambda job_def: (job_def['job'].next_run_time or paused_sort_key,
+                                                 job_def['creation_time']))
+
+    def __repr__(self):
+        self._logger.exception('<%s (client=%s)>' % (self.__class__.__name__, self.client))
+        return '<%s (client=%s)>' % (self.__class__.__name__, self.client)

env/lib/python3.5/site-packages/apscheduler/schedulers/__init__.py

+class SchedulerAlreadyRunningError(Exception):
+    """Raised when attempting to start or configure the scheduler when it's already running."""
+
+    def __str__(self):
+        return 'Scheduler is already running'
+
+
+class SchedulerNotRunningError(Exception):
+    """Raised when attempting to shutdown the scheduler when it's not running."""
+
+    def __str__(self):
+        return 'Scheduler is not running'

env/lib/python3.5/site-packages/apscheduler/schedulers/asyncio.py

+from __future__ import absolute_import
+from functools import wraps, partial
+
+from apscheduler.schedulers.base import BaseScheduler
+from apscheduler.util import maybe_ref
+
+try:
+    import asyncio
+except ImportError:  # pragma: nocover
+    try:
+        import trollius as asyncio
+    except ImportError:
+        raise ImportError(
+            'AsyncIOScheduler requires either Python 3.4 or the asyncio package installed')
+
+
+def run_in_event_loop(func):
+    @wraps(func)
+    def wrapper(self, *args, **kwargs):
+        wrapped = partial(func, self, *args, **kwargs)
+        self._eventloop.call_soon_threadsafe(wrapped)
+    return wrapper
+
+
+class AsyncIOScheduler(BaseScheduler):
+    """
+    A scheduler that runs on an asyncio (:pep:`3156`) event loop.
+
+    The default executor can run jobs based on native coroutines (``async def``).
+
+    Extra options:
+
+    ============== =============================================================
+    ``event_loop`` AsyncIO event loop to use (defaults to the global event loop)
+    ============== =============================================================
+    """
+
+    _eventloop = None
+    _timeout = None
+
+    @run_in_event_loop
+    def shutdown(self, wait=True):
+        super(AsyncIOScheduler, self).shutdown(wait)
+        self._stop_timer()
+
+    def _configure(self, config):
+        self._eventloop = maybe_ref(config.pop('event_loop', None)) or asyncio.get_event_loop()
+        super(AsyncIOScheduler, self)._configure(config)
+
+    def _start_timer(self, wait_seconds):
+        self._stop_timer()
+        if wait_seconds is not None:
+            self._timeout = self._eventloop.call_later(wait_seconds, self.wakeup)
+
+    def _stop_timer(self):
+        if self._timeout:
+            self._timeout.cancel()
+            del self._timeout
+
+    @run_in_event_loop
+    def wakeup(self):
+        self._stop_timer()
+        wait_seconds = self._process_jobs()
+        self._start_timer(wait_seconds)
+
+    def _create_default_executor(self):
+        from apscheduler.executors.asyncio import AsyncIOExecutor
+        return AsyncIOExecutor()

env/lib/python3.5/site-packages/apscheduler/schedulers/background.py

+from __future__ import absolute_import
+
+from threading import Thread, Event
+
+from apscheduler.schedulers.base import BaseScheduler
+from apscheduler.schedulers.blocking import BlockingScheduler
+from apscheduler.util import asbool
+
+
+class BackgroundScheduler(BlockingScheduler):
+    """
+    A scheduler that runs in the background using a separate thread
+    (:meth:`~apscheduler.schedulers.base.BaseScheduler.start` will return immediately).
+
+    Extra options:
+
+    ========== =============================================================================
+    ``daemon`` Set the ``daemon`` option in the background thread (defaults to ``True``, see
+               `the documentation
+               <https://docs.python.org/3.4/library/threading.html#thread-objects>`_
+               for further details)
+    ========== =============================================================================
+    """
+
+    _thread = None
+
+    def _configure(self, config):
+        self._daemon = asbool(config.pop('daemon', True))
+        super(BackgroundScheduler, self)._configure(config)
+
+    def start(self, *args, **kwargs):
+        self._event = Event()
+        BaseScheduler.start(self, *args, **kwargs)
+        self._thread = Thread(target=self._main_loop, name='APScheduler')
+        self._thread.daemon = self._daemon
+        self._thread.start()
+
+    def shutdown(self, *args, **kwargs):
+        super(BackgroundScheduler, self).shutdown(*args, **kwargs)
+        self._thread.join()
+        del self._thread

env/lib/python3.5/site-packages/apscheduler/schedulers/base.py

+from __future__ import print_function
+
+from abc import ABCMeta, abstractmethod
+from threading import RLock
+from datetime import datetime, timedelta
+from logging import getLogger
+import warnings
+import sys
+
+from pkg_resources import iter_entry_points
+from tzlocal import get_localzone
+import six
+
+from apscheduler.schedulers import SchedulerAlreadyRunningError, SchedulerNotRunningError
+from apscheduler.executors.base import MaxInstancesReachedError, BaseExecutor
+from apscheduler.executors.pool import ThreadPoolExecutor
+from apscheduler.jobstores.base import ConflictingIdError, JobLookupError, BaseJobStore
+from apscheduler.jobstores.memory import MemoryJobStore
+from apscheduler.job import Job
+from apscheduler.triggers.base import BaseTrigger
+from apscheduler.util import (
+    asbool, asint, astimezone, maybe_ref, timedelta_seconds, undefined, TIMEOUT_MAX)
+from apscheduler.events import (
+    SchedulerEvent, JobEvent, JobSubmissionEvent, EVENT_SCHEDULER_START, EVENT_SCHEDULER_SHUTDOWN,
+    EVENT_JOBSTORE_ADDED, EVENT_JOBSTORE_REMOVED, EVENT_ALL, EVENT_JOB_MODIFIED, EVENT_JOB_REMOVED,
+    EVENT_JOB_ADDED, EVENT_EXECUTOR_ADDED, EVENT_EXECUTOR_REMOVED, EVENT_ALL_JOBS_REMOVED,
+    EVENT_JOB_SUBMITTED, EVENT_JOB_MAX_INSTANCES, EVENT_SCHEDULER_RESUMED, EVENT_SCHEDULER_PAUSED)
+
+try:
+    from collections.abc import MutableMapping
+except ImportError:
+    from collections import MutableMapping
+
+#: constant indicating a scheduler's stopped state
+STATE_STOPPED = 0
+#: constant indicating a scheduler's running state (started and processing jobs)
+STATE_RUNNING = 1
+#: constant indicating a scheduler's paused state (started but not processing jobs)
+STATE_PAUSED = 2
+
+
+class BaseScheduler(six.with_metaclass(ABCMeta)):
+    """
+    Abstract base class for all schedulers.
+
+    Takes the following keyword arguments:
+
+    :param str|logging.Logger logger: logger to use for the scheduler's logging (defaults to
+        apscheduler.scheduler)
+    :param str|datetime.tzinfo timezone: the default time zone (defaults to the local timezone)
+    :param int|float jobstore_retry_interval: the minimum number of seconds to wait between
+        retries in the scheduler's main loop if the job store raises an exception when getting
+        the list of due jobs
+    :param dict job_defaults: default values for newly added jobs
+    :param dict jobstores: a dictionary of job store alias -> job store instance or configuration
+        dict
+    :param dict executors: a dictionary of executor alias -> executor instance or configuration
+        dict
+
+    :ivar int state: current running state of the scheduler (one of the following constants from
+        ``apscheduler.schedulers.base``: ``STATE_STOPPED``, ``STATE_RUNNING``, ``STATE_PAUSED``)
+
+    .. seealso:: :ref:`scheduler-config`
+    """
+
+    _trigger_plugins = dict((ep.name, ep) for ep in iter_entry_points('apscheduler.triggers'))
+    _trigger_classes = {}
+    _executor_plugins = dict((ep.name, ep) for ep in iter_entry_points('apscheduler.executors'))
+    _executor_classes = {}
+    _jobstore_plugins = dict((ep.name, ep) for ep in iter_entry_points('apscheduler.jobstores'))
+    _jobstore_classes = {}
+
+    #
+    # Public API
+    #
+
+    def __init__(self, gconfig={}, **options):
+        super(BaseScheduler, self).__init__()
+        self._executors = {}
+        self._executors_lock = self._create_lock()
+        self._jobstores = {}
+        self._jobstores_lock = self._create_lock()
+        self._listeners = []
+        self._listeners_lock = self._create_lock()
+        self._pending_jobs = []
+        self.state = STATE_STOPPED
+        self.configure(gconfig, **options)
+
+    def configure(self, gconfig={}, prefix='apscheduler.', **options):
+        """
+        Reconfigures the scheduler with the given options.
+
+        Can only be done when the scheduler isn't running.
+
+        :param dict gconfig: a "global" configuration dictionary whose values can be overridden by
+            keyword arguments to this method
+        :param str|unicode prefix: pick only those keys from ``gconfig`` that are prefixed with
+            this string (pass an empty string or ``None`` to use all keys)
+        :raises SchedulerAlreadyRunningError: if the scheduler is already running
+
+        """
+        if self.state != STATE_STOPPED:
+            raise SchedulerAlreadyRunningError
+
+        # If a non-empty prefix was given, strip it from the keys in the
+        # global configuration dict
+        if prefix:
+            prefixlen = len(prefix)
+            gconfig = dict((key[prefixlen:], value) for key, value in six.iteritems(gconfig)
+                           if key.startswith(prefix))
+
+        # Create a structure from the dotted options
+        # (e.g. "a.b.c = d" -> {'a': {'b': {'c': 'd'}}})
+        config = {}
+        for key, value in six.iteritems(gconfig):
+            parts = key.split('.')
+            parent = config
+            key = parts.pop(0)
+            while parts:
+                parent = parent.setdefault(key, {})
+                key = parts.pop(0)
+            parent[key] = value
+
+        # Override any options with explicit keyword arguments
+        config.update(options)
+        self._configure(config)
+
+    def start(self, paused=False):
+        """
+        Start the configured executors and job stores and begin processing scheduled jobs.
+
+        :param bool paused: if ``True``, don't start job processing until :meth:`resume` is called
+        :raises SchedulerAlreadyRunningError: if the scheduler is already running
+        :raises RuntimeError: if running under uWSGI with threads disabled
+
+        """
+        if self.state != STATE_STOPPED:
+            raise SchedulerAlreadyRunningError
+
+        self._check_uwsgi()
+
+        with self._executors_lock:
+            # Create a default executor if nothing else is configured
+            if 'default' not in self._executors:
+                self.add_executor(self._create_default_executor(), 'default')
+
+            # Start all the executors
+            for alias, executor in six.iteritems(self._executors):
+                executor.start(self, alias)
+
+        with self._jobstores_lock:
+            # Create a default job store if nothing else is configured
+            if 'default' not in self._jobstores:
+                self.add_jobstore(self._create_default_jobstore(), 'default')
+
+            # Start all the job stores
+            for alias, store in six.iteritems(self._jobstores):
+                store.start(self, alias)
+
+            # Schedule all pending jobs
+            for job, jobstore_alias, replace_existing in self._pending_jobs:
+                self._real_add_job(job, jobstore_alias, replace_existing)
+            del self._pending_jobs[:]
+
+        self.state = STATE_PAUSED if paused else STATE_RUNNING
+        self._logger.info('Scheduler started')
+        self._dispatch_event(SchedulerEvent(EVENT_SCHEDULER_START))
+
+        if not paused:
+            self.wakeup()
+
+    @abstractmethod
+    def shutdown(self, wait=True):
+        """
+        Shuts down the scheduler, along with its executors and job stores.
+
+        Does not interrupt any currently running jobs.
+
+        :param bool wait: ``True`` to wait until all currently executing jobs have finished
+        :raises SchedulerNotRunningError: if the scheduler has not been started yet
+
+        """
+        if self.state == STATE_STOPPED:
+            raise SchedulerNotRunningError
+
+        self.state = STATE_STOPPED
+
+        # Shut down all executors
+        with self._executors_lock:
+            for executor in six.itervalues(self._executors):
+                executor.shutdown(wait)
+
+        # Shut down all job stores
+        with self._jobstores_lock:
+            for jobstore in six.itervalues(self._jobstores):
+                jobstore.shutdown()
+
+        self._logger.info('Scheduler has been shut down')
+        self._dispatch_event(SchedulerEvent(EVENT_SCHEDULER_SHUTDOWN))
+
+    def pause(self):
+        """
+        Pause job processing in the scheduler.
+
+        This will prevent the scheduler from waking up to do job processing until :meth:`resume`
+        is called. It will not however stop any already running job processing.
+
+        """
+        if self.state == STATE_STOPPED:
+            raise SchedulerNotRunningError
+        elif self.state == STATE_RUNNING:
+            self.state = STATE_PAUSED
+            self._logger.info('Paused scheduler job processing')
+            self._dispatch_event(SchedulerEvent(EVENT_SCHEDULER_PAUSED))
+
+    def resume(self):
+        """Resume job processing in the scheduler."""
+        if self.state == STATE_STOPPED:
+            raise SchedulerNotRunningError
+        elif self.state == STATE_PAUSED:
+            self.state = STATE_RUNNING
+            self._logger.info('Resumed scheduler job processing')
+            self._dispatch_event(SchedulerEvent(EVENT_SCHEDULER_RESUMED))
+            self.wakeup()
+
+    @property
+    def running(self):
+        """
+        Return ``True`` if the scheduler has been started.
+
+        This is a shortcut for ``scheduler.state != STATE_STOPPED``.
+
+        """
+        return self.state != STATE_STOPPED
+
+    def add_executor(self, executor, alias='default', **executor_opts):
+        """
+        Adds an executor to this scheduler.
+
+        Any extra keyword arguments will be passed to the executor plugin's constructor, assuming
+        that the first argument is the name of an executor plugin.
+
+        :param str|unicode|apscheduler.executors.base.BaseExecutor executor: either an executor
+            instance or the name of an executor plugin
+        :param str|unicode alias: alias for the scheduler
+        :raises ValueError: if there is already an executor by the given alias
+
+        """
+        with self._executors_lock:
+            if alias in self._executors:
+                raise ValueError('This scheduler already has an executor by the alias of "%s"' %
+                                 alias)
+
+            if isinstance(executor, BaseExecutor):
+                self._executors[alias] = executor
+            elif isinstance(executor, six.string_types):
+                self._executors[alias] = executor = self._create_plugin_instance(
+                    'executor', executor, executor_opts)
+            else:
+                raise TypeError('Expected an executor instance or a string, got %s instead' %
+                                executor.__class__.__name__)
+
+            # Start the executor right away if the scheduler is running
+            if self.state != STATE_STOPPED:
+                executor.start(self, alias)
+
+        self._dispatch_event(SchedulerEvent(EVENT_EXECUTOR_ADDED, alias))
+
+    def remove_executor(self, alias, shutdown=True):
+        """
+        Removes the executor by the given alias from this scheduler.
+
+        :param str|unicode alias: alias of the executor
+        :param bool shutdown: ``True`` to shut down the executor after
+            removing it
+
+        """
+        with self._executors_lock:
+            executor = self._lookup_executor(alias)
+            del self._executors[alias]
+
+        if shutdown:
+            executor.shutdown()
+
+        self._dispatch_event(SchedulerEvent(EVENT_EXECUTOR_REMOVED, alias))
+
+    def add_jobstore(self, jobstore, alias='default', **jobstore_opts):
+        """
+        Adds a job store to this scheduler.
+
+        Any extra keyword arguments will be passed to the job store plugin's constructor, assuming
+        that the first argument is the name of a job store plugin.
+
+        :param str|unicode|apscheduler.jobstores.base.BaseJobStore jobstore: job store to be added
+        :param str|unicode alias: alias for the job store
+        :raises ValueError: if there is already a job store by the given alias
+
+        """
+        with self._jobstores_lock:
+            if alias in self._jobstores:
+                raise ValueError('This scheduler already has a job store by the alias of "%s"' %
+                                 alias)
+
+            if isinstance(jobstore, BaseJobStore):
+                self._jobstores[alias] = jobstore
+            elif isinstance(jobstore, six.string_types):
+                self._jobstores[alias] = jobstore = self._create_plugin_instance(
+                    'jobstore', jobstore, jobstore_opts)
+            else:
+                raise TypeError('Expected a job store instance or a string, got %s instead' %
+                                jobstore.__class__.__name__)
+
+            # Start the job store right away if the scheduler isn't stopped
+            if self.state != STATE_STOPPED:
+                jobstore.start(self, alias)
+
+        # Notify listeners that a new job store has been added
+        self._dispatch_event(SchedulerEvent(EVENT_JOBSTORE_ADDED, alias))
+
+        # Notify the scheduler so it can scan the new job store for jobs
+        if self.state != STATE_STOPPED:
+            self.wakeup()
+
+    def remove_jobstore(self, alias, shutdown=True):
+        """
+        Removes the job store by the given alias from this scheduler.
+
+        :param str|unicode alias: alias of the job store
+        :param bool shutdown: ``True`` to shut down the job store after removing it
+
+        """
+        with self._jobstores_lock:
+            jobstore = self._lookup_jobstore(alias)
+            del self._jobstores[alias]
+
+        if shutdown:
+            jobstore.shutdown()
+
+        self._dispatch_event(SchedulerEvent(EVENT_JOBSTORE_REMOVED, alias))
+
+    def add_listener(self, callback, mask=EVENT_ALL):
+        """
+        add_listener(callback, mask=EVENT_ALL)
+
+        Adds a listener for scheduler events.
+
+        When a matching event  occurs, ``callback`` is executed with the event object as its
+        sole argument. If the ``mask`` parameter is not provided, the callback will receive events
+        of all types.
+
+        :param callback: any callable that takes one argument
+        :param int mask: bitmask that indicates which events should be
+            listened to
+
+        .. seealso:: :mod:`apscheduler.events`
+        .. seealso:: :ref:`scheduler-events`
+
+        """
+        with self._listeners_lock:
+            self._listeners.append((callback, mask))
+
+    def remove_listener(self, callback):
+        """Removes a previously added event listener."""
+
+        with self._listeners_lock:
+            for i, (cb, _) in enumerate(self._listeners):
+                if callback == cb:
+                    del self._listeners[i]
+
+    def add_job(self, func, trigger=None, args=None, kwargs=None, id=None, name=None,
+                misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,
+                next_run_time=undefined, jobstore='default', executor='default',
+                replace_existing=False, **trigger_args):
+        """
+        add_job(func, trigger=None, args=None, kwargs=None, id=None, \
+            name=None, misfire_grace_time=undefined, coalesce=undefined, \
+            max_instances=undefined, next_run_time=undefined, \
+            jobstore='default', executor='default', \
+            replace_existing=False, **trigger_args)
+
+        Adds the given job to the job list and wakes up the scheduler if it's already running.
+
+        Any option that defaults to ``undefined`` will be replaced with the corresponding default
+        value when the job is scheduled (which happens when the scheduler is started, or
+        immediately if the scheduler is already running).
+
+        The ``func`` argument can be given either as a callable object or a textual reference in
+        the ``package.module:some.object`` format, where the first half (separated by ``:``) is an
+        importable module and the second half is a reference to the callable object, relative to
+        the module.
+
+        The ``trigger`` argument can either be:
+          #. the alias name of the trigger (e.g. ``date``, ``interval`` or ``cron``), in which case
+            any extra keyword arguments to this method are passed on to the trigger's constructor
+          #. an instance of a trigger class
+
+        :param func: callable (or a textual reference to one) to run at the given time
+        :param str|apscheduler.triggers.base.BaseTrigger trigger: trigger that determines when
+            ``func`` is called
+        :param list|tuple args: list of positional arguments to call func with
+        :param dict kwargs: dict of keyword arguments to call func with
+        :param str|unicode id: explicit identifier for the job (for modifying it later)
+        :param str|unicode name: textual description of the job
+        :param int misfire_grace_time: seconds after the designated runtime that the job is still
+            allowed to be run
+        :param bool coalesce: run once instead of many times if the scheduler determines that the
+            job should be run more than once in succession
+        :param int max_instances: maximum number of concurrently running instances allowed for this
+            job
+        :param datetime next_run_time: when to first run the job, regardless of the trigger (pass
+            ``None`` to add the job as paused)
+        :param str|unicode jobstore: alias of the job store to store the job in
+        :param str|unicode executor: alias of the executor to run the job with
+        :param bool replace_existing: ``True`` to replace an existing job with the same ``id``
+            (but retain the number of runs from the existing one)
+        :rtype: Job
+
+        """
+        job_kwargs = {
+            'trigger': self._create_trigger(trigger, trigger_args),
+            'executor': executor,
+            'func': func,
+            'args': tuple(args) if args is not None else (),
+            'kwargs': dict(kwargs) if kwargs is not None else {},
+            'id': id,
+            'name': name,
+            'misfire_grace_time': misfire_grace_time,
+            'coalesce': coalesce,
+            'max_instances': max_instances,
+            'next_run_time': next_run_time
+        }
+        job_kwargs = dict((key, value) for key, value in six.iteritems(job_kwargs) if
+                          value is not undefined)
+        job = Job(self, **job_kwargs)
+
+        # Don't really add jobs to job stores before the scheduler is up and running
+        with self._jobstores_lock:
+            if self.state == STATE_STOPPED:
+                self._pending_jobs.append((job, jobstore, replace_existing))
+                self._logger.info('Adding job tentatively -- it will be properly scheduled when '
+                                  'the scheduler starts')
+            else:
+                self._real_add_job(job, jobstore, replace_existing)
+
+        return job
+
+    def scheduled_job(self, trigger, args=None, kwargs=None, id=None, name=None,
+                      misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,
+                      next_run_time=undefined, jobstore='default', executor='default',
+                      **trigger_args):
+        """
+        scheduled_job(trigger, args=None, kwargs=None, id=None, \
+            name=None, misfire_grace_time=undefined, \
+            coalesce=undefined, max_instances=undefined, \
+            next_run_time=undefined, jobstore='default', \
+            executor='default',**trigger_args)
+
+        A decorator version of :meth:`add_job`, except that ``replace_existing`` is always
+        ``True``.
+
+        .. important:: The ``id`` argument must be given if scheduling a job in a persistent job
+        store. The scheduler cannot, however, enforce this requirement.
+
+        """
+        def inner(func):
+            self.add_job(func, trigger, args, kwargs, id, name, misfire_grace_time, coalesce,
+                         max_instances, next_run_time, jobstore, executor, True, **trigger_args)
+            return func
+        return inner
+
+    def modify_job(self, job_id, jobstore=None, **changes):
+        """
+        Modifies the properties of a single job.
+
+        Modifications are passed to this method as extra keyword arguments.
+
+        :param str|unicode job_id: the identifier of the job
+        :param str|unicode jobstore: alias of the job store that contains the job
+        :return Job: the relevant job instance
+
+        """
+        with self._jobstores_lock:
+            job, jobstore = self._lookup_job(job_id, jobstore)
+            job._modify(**changes)
+            if jobstore:
+                self._lookup_jobstore(jobstore).update_job(job)
+
+        self._dispatch_event(JobEvent(EVENT_JOB_MODIFIED, job_id, jobstore))
+
+        # Wake up the scheduler since the job's next run time may have been changed
+        if self.state == STATE_RUNNING:
+            self.wakeup()
+
+        return job
+
+    def reschedule_job(self, job_id, jobstore=None, trigger=None, **trigger_args):
+        """
+        Constructs a new trigger for a job and updates its next run time.
+
+        Extra keyword arguments are passed directly to the trigger's constructor.
+
+        :param str|unicode job_id: the identifier of the job
+        :param str|unicode jobstore: alias of the job store that contains the job
+        :param trigger: alias of the trigger type or a trigger instance
+        :return Job: the relevant job instance
+
+        """
+        trigger = self._create_trigger(trigger, trigger_args)
+        now = datetime.now(self.timezone)
+        next_run_time = trigger.get_next_fire_time(None, now)
+        return self.modify_job(job_id, jobstore, trigger=trigger, next_run_time=next_run_time)
+
+    def pause_job(self, job_id, jobstore=None):
+        """
+        Causes the given job not to be executed until it is explicitly resumed.
+
+        :param str|unicode job_id: the identifier of the job
+        :param str|unicode jobstore: alias of the job store that contains the job
+        :return Job: the relevant job instance
+
+        """
+        return self.modify_job(job_id, jobstore, next_run_time=None)
+
+    def resume_job(self, job_id, jobstore=None):
+        """
+        Resumes the schedule of the given job, or removes the job if its schedule is finished.
+
+        :param str|unicode job_id: the identifier of the job
+        :param str|unicode jobstore: alias of the job store that contains the job
+        :return Job|None: the relevant job instance if the job was rescheduled, or ``None`` if no
+            next run time could be calculated and the job was removed
+
+        """
+        with self._jobstores_lock:
+            job, jobstore = self._lookup_job(job_id, jobstore)
+            now = datetime.now(self.timezone)
+            next_run_time = job.trigger.get_next_fire_time(None, now)
+            if next_run_time:
+                return self.modify_job(job_id, jobstore, next_run_time=next_run_time)
+            else:
+                self.remove_job(job.id, jobstore)
+
+    def get_jobs(self, jobstore=None, pending=None):
+        """
+        Returns a list of pending jobs (if the scheduler hasn't been started yet) and scheduled
+        jobs, either from a specific job store or from all of them.
+
+        If the scheduler has not been started yet, only pending jobs can be returned because the
+        job stores haven't been started yet either.
+
+        :param str|unicode jobstore: alias of the job store
+        :param bool pending: **DEPRECATED**
+        :rtype: list[Job]
+
+        """
+        if pending is not None:
+            warnings.warn('The "pending" option is deprecated -- get_jobs() always returns '
+                          'scheduled jobs if the scheduler has been started and pending jobs '
+                          'otherwise', DeprecationWarning)
+
+        with self._jobstores_lock:
+            jobs = []
+            if self.state == STATE_STOPPED:
+                for job, alias, replace_existing in self._pending_jobs:
+                    if jobstore is None or alias == jobstore:
+                        jobs.append(job)
+            else:
+                for alias, store in six.iteritems(self._jobstores):
+                    if jobstore is None or alias == jobstore:
+                        jobs.extend(store.get_all_jobs())
+
+            return jobs
+
+    def get_job(self, job_id, jobstore=None):
+        """
+        Returns the Job that matches the given ``job_id``.
+
+        :param str|unicode job_id: the identifier of the job
+        :param str|unicode jobstore: alias of the job store that most likely contains the job
+        :return: the Job by the given ID, or ``None`` if it wasn't found
+        :rtype: Job
+
+        """
+        with self._jobstores_lock:
+            try:
+                return self._lookup_job(job_id, jobstore)[0]
+            except JobLookupError:
+                return
+
+    def remove_job(self, job_id, jobstore=None):
+        """
+        Removes a job, preventing it from being run any more.
+
+        :param str|unicode job_id: the identifier of the job
+        :param str|unicode jobstore: alias of the job store that contains the job
+        :raises JobLookupError: if the job was not found
+
+        """
+        jobstore_alias = None
+        with self._jobstores_lock:
+            # Check if the job is among the pending jobs
+            if self.state == STATE_STOPPED:
+                for i, (job, alias, replace_existing) in enumerate(self._pending_jobs):
+                    if job.id == job_id and jobstore in (None, alias):
+                        del self._pending_jobs[i]
+                        jobstore_alias = alias
+                        break
+            else:
+                # Otherwise, try to remove it from each store until it succeeds or we run out of
+                # stores to check
+                for alias, store in six.iteritems(self._jobstores):
+                    if jobstore in (None, alias):
+                        try:
+                            store.remove_job(job_id)
+                            jobstore_alias = alias
+                            break
+                        except JobLookupError:
+                            continue
+
+        if jobstore_alias is None:
+            raise JobLookupError(job_id)
+
+        # Notify listeners that a job has been removed
+        event = JobEvent(EVENT_JOB_REMOVED, job_id, jobstore_alias)
+        self._dispatch_event(event)
+
+        self._logger.info('Removed job %s', job_id)
+
+    def remove_all_jobs(self, jobstore=None):
+        """
+        Removes all jobs from the specified job store, or all job stores if none is given.
+
+        :param str|unicode jobstore: alias of the job store
+
+        """
+        with self._jobstores_lock:
+            if self.state == STATE_STOPPED:
+                if jobstore:
+                    self._pending_jobs = [pending for pending in self._pending_jobs if
+                                          pending[1] != jobstore]
+                else:
+                    self._pending_jobs = []
+            else:
+                for alias, store in six.iteritems(self._jobstores):
+                    if jobstore in (None, alias):
+                        store.remove_all_jobs()
+
+        self._dispatch_event(SchedulerEvent(EVENT_ALL_JOBS_REMOVED, jobstore))
+
+    def print_jobs(self, jobstore=None, out=None):
+        """
+        print_jobs(jobstore=None, out=sys.stdout)
+
+        Prints out a textual listing of all jobs currently scheduled on either all job stores or
+        just a specific one.
+
+        :param str|unicode jobstore: alias of the job store, ``None`` to list jobs from all stores
+        :param file out: a file-like object to print to (defaults to  **sys.stdout** if nothing is
+            given)
+
+        """
+        out = out or sys.stdout
+        with self._jobstores_lock:
+            if self.state == STATE_STOPPED:
+                print(u'Pending jobs:', file=out)
+                if self._pending_jobs:
+                    for job, jobstore_alias, replace_existing in self._pending_jobs:
+                        if jobstore in (None, jobstore_alias):
+                            print(u'    %s' % job, file=out)
+                else:
+                    print(u'    No pending jobs', file=out)
+            else:
+                for alias, store in sorted(six.iteritems(self._jobstores)):
+                    if jobstore in (None, alias):
+                        print(u'Jobstore %s:' % alias, file=out)
+                        jobs = store.get_all_jobs()
+                        if jobs:
+                            for job in jobs:
+                                print(u'    %s' % job, file=out)
+                        else:
+                            print(u'    No scheduled jobs', file=out)
+
+    @abstractmethod
+    def wakeup(self):
+        """
+        Notifies the scheduler that there may be jobs due for execution.
+        Triggers :meth:`_process_jobs` to be run in an implementation specific manner.
+        """
+
+    #
+    # Private API
+    #
+
+    def _configure(self, config):
+        # Set general options
+        self._logger = maybe_ref(config.pop('logger', None)) or getLogger('apscheduler.scheduler')
+        self.timezone = astimezone(config.pop('timezone', None)) or get_localzone()
+        self.jobstore_retry_interval = float(config.pop('jobstore_retry_interval', 10))
+
+        # Set the job defaults
+        job_defaults = config.get('job_defaults', {})
+        self._job_defaults = {
+            'misfire_grace_time': asint(job_defaults.get('misfire_grace_time', 1)),
+            'coalesce': asbool(job_defaults.get('coalesce', True)),
+            'max_instances': asint(job_defaults.get('max_instances', 1))
+        }
+
+        # Configure executors
+        self._executors.clear()
+        for alias, value in six.iteritems(config.get('executors', {})):
+            if isinstance(value, BaseExecutor):
+                self.add_executor(value, alias)
+            elif isinstance(value, MutableMapping):
+                executor_class = value.pop('class', None)
+                plugin = value.pop('type', None)
+                if plugin:
+                    executor = self._create_plugin_instance('executor', plugin, value)
+                elif executor_class:
+                    cls = maybe_ref(executor_class)
+                    executor = cls(**value)
+                else:
+                    raise ValueError(
+                        'Cannot create executor "%s" -- either "type" or "class" must be defined' %
+                        alias)
+
+                self.add_executor(executor, alias)
+            else:
+                raise TypeError(
+                    "Expected executor instance or dict for executors['%s'], got %s instead" %
+                    (alias, value.__class__.__name__))
+
+        # Configure job stores
+        self._jobstores.clear()
+        for alias, value in six.iteritems(config.get('jobstores', {})):
+            if isinstance(value, BaseJobStore):
+                self.add_jobstore(value, alias)
+            elif isinstance(value, MutableMapping):
+                jobstore_class = value.pop('class', None)
+                plugin = value.pop('type', None)
+                if plugin:
+                    jobstore = self._create_plugin_instance('jobstore', plugin, value)
+                elif jobstore_class:
+                    cls = maybe_ref(jobstore_class)
+                    jobstore = cls(**value)
+                else:
+                    raise ValueError(
+                        'Cannot create job store "%s" -- either "type" or "class" must be '
+                        'defined' % alias)
+
+                self.add_jobstore(jobstore, alias)
+            else:
+                raise TypeError(
+                    "Expected job store instance or dict for jobstores['%s'], got %s instead" %
+                    (alias, value.__class__.__name__))
+
+    def _create_default_executor(self):
+        """Creates a default executor store, specific to the particular scheduler type."""
+        return ThreadPoolExecutor()
+
+    def _create_default_jobstore(self):
+        """Creates a default job store, specific to the particular scheduler type."""
+        return MemoryJobStore()
+
+    def _lookup_executor(self, alias):
+        """
+        Returns the executor instance by the given name from the list of executors that were added
+        to this scheduler.
+
+        :type alias: str
+        :raises KeyError: if no executor by the given alias is not found
+
+        """
+        try:
+            return self._executors[alias]
+        except KeyError:
+            raise KeyError('No such executor: %s' % alias)
+
+    def _lookup_jobstore(self, alias):
+        """
+        Returns the job store instance by the given name from the list of job stores that were
+        added to this scheduler.
+
+        :type alias: str
+        :raises KeyError: if no job store by the given alias is not found
+
+        """
+        try:
+            return self._jobstores[alias]
+        except KeyError:
+            raise KeyError('No such job store: %s' % alias)
+
+    def _lookup_job(self, job_id, jobstore_alias):
+        """
+        Finds a job by its ID.
+
+        :type job_id: str
+        :param str jobstore_alias: alias of a job store to look in
+        :return tuple[Job, str]: a tuple of job, jobstore alias (jobstore alias is None in case of
+            a pending job)
+        :raises JobLookupError: if no job by the given ID is found.
+
+        """
+        if self.state == STATE_STOPPED:
+            # Check if the job is among the pending jobs
+            for job, alias, replace_existing in self._pending_jobs:
+                if job.id == job_id:
+                    return job, None
+        else:
+            # Look in all job stores
+            for alias, store in six.iteritems(self._jobstores):
+                if jobstore_alias in (None, alias):
+                    job = store.lookup_job(job_id)
+                    if job is not None:
+                        return job, alias
+
+        raise JobLookupError(job_id)
+
+    def _dispatch_event(self, event):
+        """
+        Dispatches the given event to interested listeners.
+
+        :param SchedulerEvent event: the event to send
+
+        """
+        with self._listeners_lock:
+            listeners = tuple(self._listeners)
+
+        for cb, mask in listeners:
+            if event.code & mask:
+                try:
+                    cb(event)
+                except BaseException:
+                    self._logger.exception('Error notifying listener')
+
+    def _check_uwsgi(self):
+        """Check if we're running under uWSGI with threads disabled."""
+        uwsgi_module = sys.modules.get('uwsgi')
+        if not getattr(uwsgi_module, 'has_threads', True):
+            raise RuntimeError('The scheduler seems to be running under uWSGI, but threads have '
+                               'been disabled. You must run uWSGI with the --enable-threads '
+                               'option for the scheduler to work.')
+
+    def _real_add_job(self, job, jobstore_alias, replace_existing):
+        """
+        :param Job job: the job to add
+        :param bool replace_existing: ``True`` to use update_job() in case the job already exists
+            in the store
+
+        """
+        # Fill in undefined values with defaults
+        replacements = {}
+        for key, value in six.iteritems(self._job_defaults):
+            if not hasattr(job, key):
+                replacements[key] = value
+
+        # Calculate the next run time if there is none defined
+        if not hasattr(job, 'next_run_time'):
+            now = datetime.now(self.timezone)
+            replacements['next_run_time'] = job.trigger.get_next_fire_time(None, now)
+
+        # Apply any replacements
+        job._modify(**replacements)
+
+        # Add the job to the given job store
+        store = self._lookup_jobstore(jobstore_alias)
+        try:
+            store.add_job(job)
+        except ConflictingIdError:
+            if replace_existing:
+                store.update_job(job)
+            else:
+                raise
+
+        # Mark the job as no longer pending
+        job._jobstore_alias = jobstore_alias
+
+        # Notify listeners that a new job has been added
+        event = JobEvent(EVENT_JOB_ADDED, job.id, jobstore_alias)
+        self._dispatch_event(event)
+
+        self._logger.info('Added job "%s" to job store "%s"', job.name, jobstore_alias)
+
+        # Notify the scheduler about the new job
+        if self.state == STATE_RUNNING:
+            self.wakeup()
+
+    def _create_plugin_instance(self, type_, alias, constructor_kwargs):
+        """Creates an instance of the given plugin type, loading the plugin first if necessary."""
+        plugin_container, class_container, base_class = {
+            'trigger': (self._trigger_plugins, self._trigger_classes, BaseTrigger),
+            'jobstore': (self._jobstore_plugins, self._jobstore_classes, BaseJobStore),
+            'executor': (self._executor_plugins, self._executor_classes, BaseExecutor)
+        }[type_]
+
+        try:
+            plugin_cls = class_container[alias]
+        except KeyError:
+            if alias in plugin_container:
+                plugin_cls = class_container[alias] = plugin_container[alias].load()
+                if not issubclass(plugin_cls, base_class):
+                    raise TypeError('The {0} entry point does not point to a {0} class'.
+                                    format(type_))
+            else:
+                raise LookupError('No {0} by the name "{1}" was found'.format(type_, alias))
+
+        return plugin_cls(**constructor_kwargs)
+
+    def _create_trigger(self, trigger, trigger_args):
+        if isinstance(trigger, BaseTrigger):
+            return trigger
+        elif trigger is None:
+            trigger = 'date'
+        elif not isinstance(trigger, six.string_types):
+            raise TypeError('Expected a trigger instance or string, got %s instead' %
+                            trigger.__class__.__name__)
+
+        # Use the scheduler's time zone if nothing else is specified
+        trigger_args.setdefault('timezone', self.timezone)
+
+        # Instantiate the trigger class
+        return self._create_plugin_instance('trigger', trigger, trigger_args)
+
+    def _create_lock(self):
+        """Creates a reentrant lock object."""
+        return RLock()
+
+    def _process_jobs(self):
+        """
+        Iterates through jobs in every jobstore, starts jobs that are due and figures out how long
+        to wait for the next round.
+
+        If the ``get_due_jobs()`` call raises an exception, a new wakeup is scheduled in at least
+        ``jobstore_retry_interval`` seconds.
+
+        """
+        if self.state == STATE_PAUSED:
+            self._logger.debug('Scheduler is paused -- not processing jobs')
+            return None
+
+        self._logger.debug('Looking for jobs to run')
+        now = datetime.now(self.timezone)
+        next_wakeup_time = None
+        events = []
+
+        with self._jobstores_lock:
+            for jobstore_alias, jobstore in six.iteritems(self._jobstores):
+                try:
+                    due_jobs = jobstore.get_due_jobs(now)
+                except Exception as e:
+                    # Schedule a wakeup at least in jobstore_retry_interval seconds
+                    self._logger.warning('Error getting due jobs from job store %r: %s',
+                                         jobstore_alias, e)
+                    retry_wakeup_time = now + timedelta(seconds=self.jobstore_retry_interval)
+                    if not next_wakeup_time or next_wakeup_time > retry_wakeup_time:
+                        next_wakeup_time = retry_wakeup_time
+
+                    continue
+
+                for job in due_jobs:
+                    # Look up the job's executor
+                    try:
+                        executor = self._lookup_executor(job.executor)
+                    except BaseException:
+                        self._logger.error(
+                            'Executor lookup ("%s") failed for job "%s" -- removing it from the '
+                            'job store', job.executor, job)
+                        self.remove_job(job.id, jobstore_alias)
+                        continue
+
+                    run_times = job._get_run_times(now)
+                    run_times = run_times[-1:] if run_times and job.coalesce else run_times
+                    if run_times:
+                        try:
+                            executor.submit_job(job, run_times)
+                        except MaxInstancesReachedError:
+                            self._logger.warning(
+                                'Execution of job "%s" skipped: maximum number of running '
+                                'instances reached (%d)', job, job.max_instances)
+                            event = JobSubmissionEvent(EVENT_JOB_MAX_INSTANCES, job.id,
+                                                       jobstore_alias, run_times)
+                            events.append(event)
+                        except BaseException:
+                            self._logger.exception('Error submitting job "%s" to executor "%s"',
+                                                   job, job.executor)
+                        else:
+                            event = JobSubmissionEvent(EVENT_JOB_SUBMITTED, job.id, jobstore_alias,
+                                                       run_times)
+                            events.append(event)
+
+                        # Update the job if it has a next execution time.
+                        # Otherwise remove it from the job store.
+                        job_next_run = job.trigger.get_next_fire_time(run_times[-1], now)
+                        if job_next_run:
+                            job._modify(next_run_time=job_next_run)
+                            jobstore.update_job(job)
+                        else:
+                            self.remove_job(job.id, jobstore_alias)
+
+                # Set a new next wakeup time if there isn't one yet or
+                # the jobstore has an even earlier one
+                jobstore_next_run_time = jobstore.get_next_run_time()
+                if jobstore_next_run_time and (next_wakeup_time is None or
+                                               jobstore_next_run_time < next_wakeup_time):
+                    next_wakeup_time = jobstore_next_run_time.astimezone(self.timezone)
+
+        # Dispatch collected events
+        for event in events:
+            self._dispatch_event(event)
+
+        # Determine the delay until this method should be called again
+        if self.state == STATE_PAUSED:
+            wait_seconds = None
+            self._logger.debug('Scheduler is paused; waiting until resume() is called')
+        elif next_wakeup_time is None:
+            wait_seconds = None
+            self._logger.debug('No jobs; waiting until a job is added')
+        else:
+            wait_seconds = min(max(timedelta_seconds(next_wakeup_time - now), 0), TIMEOUT_MAX)
+            self._logger.debug('Next wakeup is due at %s (in %f seconds)', next_wakeup_time,
+                               wait_seconds)
+
+        return wait_seconds

env/lib/python3.5/site-packages/apscheduler/schedulers/blocking.py

+from __future__ import absolute_import
+
+from threading import Event
+
+from apscheduler.schedulers.base import BaseScheduler, STATE_STOPPED
+from apscheduler.util import TIMEOUT_MAX
+
+
+class BlockingScheduler(BaseScheduler):
+    """
+    A scheduler that runs in the foreground
+    (:meth:`~apscheduler.schedulers.base.BaseScheduler.start` will block).
+    """
+    _event = None
+
+    def start(self, *args, **kwargs):
+        self._event = Event()
+        super(BlockingScheduler, self).start(*args, **kwargs)
+        self._main_loop()
+
+    def shutdown(self, wait=True):
+        super(BlockingScheduler, self).shutdown(wait)
+        self._event.set()
+
+    def _main_loop(self):
+        wait_seconds = TIMEOUT_MAX
+        while self.state != STATE_STOPPED:
+            self._event.wait(wait_seconds)
+            self._event.clear()
+            wait_seconds = self._process_jobs()
+
+    def wakeup(self):
+        self._event.set()