<commit_before>from __future__ import absolute_import

from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
from apscheduler.util import maybe_ref, datetime_to_utc_timestamp
from apscheduler.job import Job

try:
    import cPickle as pickle
except ImportError:  # pragma: nocover
    import pickle

try:
    import rethinkdb as r
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 rethink'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.Connectio` 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__()
        self.pickle_protocol = pickle_protocol

        if not database:
            raise ValueError('The "database" parameter must not be empty')
        if not table:
            raise ValueError('The "table" parameter must not be empty')

        if client:
            self.conn = maybe_ref(client)
        else:
            self.conn = r.connect(db=database, **connect_args)

        if database not in r.db_list().run(self.conn):
            r.db_create(database).run(self.conn)

        if table not in r.table_list().run(self.conn):
            r.table_create(table).run(self.conn)

        if 'next_run_time' not in r.table(table).index_list().run(self.conn):
            r.table(table).index_create('next_run_time').run(self.conn)

        self.table = r.db(database).table(table)

    def lookup_job(self, job_id):
        document = self.table.get(job_id).pluck('job_state').run(self.conn)
        return self._reconstitute_job(document['job_state']) if document else None

    def get_due_jobs(self, now):
        timestamp = r.epoch_time(datetime_to_utc_timestamp(now))
        search = (lambda x: x['next_run_time'] <= timestamp)
        return self._get_jobs(search)

    def get_next_run_time(self):
        document = (
            self.table
            .filter(
                lambda x:
                x['next_run_time'] != None
            )
            .pluck('next_run_time')
            .order_by(r.asc('next_run_time'))
            .limit(1)
            .run(self.conn)
        )
        if document:
            document = document[0]['next_run_time']
        else:
            document = None

        return document

    def get_all_jobs(self):
        return self._get_jobs()

    def add_job(self, job):
        job_exist = self.table.get(job.id).run(self.conn)
        if not job_exist:
            job_dict = {}
            job_dict['id'] = job.id
            job_dict['job_state'] = (
                pickle
                .dumps(job.__getstate__(), self.pickle_protocol)
                .encode("zip")
                .encode("base64")
                .strip()
            )
            job_dict['next_run_time'] = r.epoch_time(
                float(job_dict['next_run_time'].strftime('%s'))
            )
            self.table.insert(job_dict).run(self.conn)
        else:
            raise ConflictingIdError(job.id)

    def update_job(self, job):
        document = {}
        next_run_time = r.epoch_time(
            float(job.next_run_time.strftime('%s'))
        )
        document['job_state'] = (
            pickle
            .dumps(job.__getstate__(), self.pickle_protocol)
            .encode("zip")
            .encode("base64")
            .strip()
        )
        document['next_run_time'] = next_run_time
        results = self.table.get(job.id).update(document).run(self.conn)
        if results['skipped'] > 0:
            raise JobLookupError(id)

    def remove_job(self, job):
        results = self.table.get(job.id).delete().run(self.conn)
        if results['deleted'] == 0:
            raise JobLookupError(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.decode("base64").decode("zip"))
        job = Job.__new__(Job)
        job.__setstate__(job_state)
        job._scheduler = self._scheduler
        job._jobstore_alias = self._alias
        return job<commit_msg>Fixed next_run_time parser<commit_after>from __future__ import absolute_import

from apscheduler.jobstores.base import BaseJobStore, JobLookupError, ConflictingIdError
from apscheduler.util import maybe_ref, datetime_to_utc_timestamp
from apscheduler.job import Job

try:
    import cPickle as pickle
except ImportError:  # pragma: nocover
    import pickle

try:
    import rethinkdb as r
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 rethink'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.Connectio` 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__()
        self.pickle_protocol = pickle_protocol

        if not database:
            raise ValueError('The "database" parameter must not be empty')
        if not table:
            raise ValueError('The "table" parameter must not be empty')

        if client:
            self.conn = maybe_ref(client)
        else:
            self.conn = r.connect(db=database, **connect_args)

        if database not in r.db_list().run(self.conn):
            r.db_create(database).run(self.conn)

        if table not in r.table_list().run(self.conn):
            r.table_create(table).run(self.conn)

        if 'next_run_time' not in r.table(table).index_list().run(self.conn):
            r.table(table).index_create('next_run_time').run(self.conn)

        self.table = r.db(database).table(table)

    def lookup_job(self, job_id):
        document = self.table.get(job_id).pluck('job_state').run(self.conn)
        return self._reconstitute_job(document['job_state']) if document else None

    def get_due_jobs(self, now):
        timestamp = r.epoch_time(datetime_to_utc_timestamp(now))
        search = (lambda x: x['next_run_time'] <= timestamp)
        return self._get_jobs(search)

    def get_next_run_time(self):
        document = (
            self.table
            .filter(
                lambda x:
                x['next_run_time'] != None
            )
            .pluck('next_run_time')
            .order_by(r.asc('next_run_time'))
            .limit(1)
            .run(self.conn)
        )
        if document:
            document = document[0]['next_run_time']
        else:
            document = None

        return document

    def get_all_jobs(self):
        return self._get_jobs()

    def add_job(self, job):
        job_exist = self.table.get(job.id).run(self.conn)
        if not job_exist:
            job_dict = {}
            job_dict['id'] = job.id
            job_dict['job_state'] = (
                pickle
                .dumps(job.__getstate__(), self.pickle_protocol)
                .encode("zip")
                .encode("base64")
                .strip()
            )
            job_dict['next_run_time'] = r.epoch_time(
                datetime_to_utc_timestamp(job.next_run_time)
            )
            self.table.insert(job_dict).run(self.conn)
        else:
            raise ConflictingIdError(job.id)

    def update_job(self, job):
        document = {}
        next_run_time = r.epoch_time(
            datetime_to_utc_timestamp(job.next_run_time)
        )
        document['job_state'] = (
            pickle
            .dumps(job.__getstate__(), self.pickle_protocol)
            .encode("zip")
            .encode("base64")
            .strip()
        )
        document['next_run_time'] = next_run_time
        results = self.table.get(job.id).update(document).run(self.conn)
        if results['skipped'] > 0:
            raise JobLookupError(id)

    def remove_job(self, job):
        results = self.table.get(job.id).delete().run(self.conn)
        if results['deleted'] == 0:
            raise JobLookupError(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.decode("base64").decode("zip"))
        job = Job.__new__(Job)
        job.__setstate__(job_state)
        job._scheduler = self._scheduler
        job._jobstore_alias = self._alias
        return job
{'fraction_non_alphanumeric': 0.10690121786197564, 'fraction_numerical': 0.0020818153429790777, 'mean_word_length': 2.283663704716336, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5430269', 'n_tokens_mistral': 2925, 'n_tokens_neox': 2846, 'n_words': 595}