<s> from __future__ import absolute_import <EOL> from datetime import datetime <EOL> from pytz import utc <EOL> import six <EOL> from apscheduler . jobstores . base import BaseJobStore , JobLookupError , ConflictingIdError <EOL> from apscheduler . util import datetime_to_utc_timestamp , utc_timestamp_to_datetime <EOL> from apscheduler . job import Job <EOL> try : <EOL> import cPickle as pickle <EOL> except ImportError : <EOL> import pickle <EOL> try : <EOL> from redis import StrictRedis <EOL> except ImportError : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> class RedisJobStore ( BaseJobStore ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db = <NUM_LIT:0> , jobs_key = '<STR_LIT>' , run_times_key = '<STR_LIT>' , <EOL> pickle_protocol = pickle . HIGHEST_PROTOCOL , ** connect_args ) : <EOL> super ( RedisJobStore , self ) . __init__ ( ) <EOL> if db is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not jobs_key : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not run_times_key : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . pickle_protocol = pickle_protocol <EOL> self . jobs_key = jobs_key <EOL> self . run_times_key = run_times_key <EOL> self . redis = StrictRedis ( db = int ( db ) , ** connect_args ) <EOL> def lookup_job ( self , job_id ) : <EOL> job_state = self . redis . hget ( self . jobs_key , job_id ) <EOL> return self . _reconstitute_job ( job_state ) if job_state else None <EOL> def get_due_jobs ( self , now ) : <EOL> timestamp = datetime_to_utc_timestamp ( now ) <EOL> job_ids = self . redis . zrangebyscore ( self . run_times_key , <NUM_LIT:0> , timestamp ) <EOL> if job_ids : <EOL> job_states = self . redis . hmget ( self . jobs_key , * job_ids ) <EOL> return self . _reconstitute_jobs ( six . moves . zip ( job_ids , job_states ) ) <EOL> return [ ] <EOL> def get_next_run_time ( self ) : <EOL> next_run_time = self . redis . zrange ( self . run_times_key , <NUM_LIT:0> , <NUM_LIT:0> , withscores = True ) <EOL> if next_run_time : <EOL> return utc_timestamp_to_datetime ( next_run_time [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> def get_all_jobs ( self ) : <EOL> job_states = self . redis . hgetall ( self . jobs_key ) <EOL> jobs = self . _reconstitute_jobs ( six . iteritems ( job_states ) ) <EOL> paused_sort_key = datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT> , tzinfo = utc ) <EOL> return sorted ( jobs , key = lambda job : job . next_run_time or paused_sort_key ) <EOL> def add_job ( self , job ) : <EOL> if self . redis . hexists ( self . jobs_key , job . id ) : <EOL> raise ConflictingIdError ( job . id ) <EOL> with self . redis . pipeline ( ) as pipe : <EOL> pipe . multi ( ) <EOL> pipe . hset ( self . jobs_key , job . id , pickle . dumps ( job . __getstate__ ( ) , <EOL> self . pickle_protocol ) ) <EOL> if job . next_run_time : <EOL> pipe . zadd ( self . run_times_key , datetime_to_utc_timestamp ( job . next_run_time ) , job . id ) <EOL> pipe . execute ( ) <EOL> def update_job ( self , job ) : <EOL> if not self . redis . hexists ( self . jobs_key , job . id ) : <EOL> raise JobLookupError ( job . id ) <EOL> with self . redis . pipeline ( ) as pipe : <EOL> pipe . hset ( self . jobs_key , job . id , pickle . dumps ( job . __getstate__ ( ) , <EOL> self . pickle_protocol ) ) <EOL> if job . next_run_time : <EOL> pipe . zadd ( self . run_times_key , datetime_to_utc_timestamp ( job . next_run_time ) , job . id ) <EOL> else : <EOL> pipe . zrem ( self . run_times_key , job . id ) <EOL> pipe . execute ( ) <EOL> def remove_job ( self , job_id ) : <EOL> if not self . redis . hexists ( self . jobs_key , job_id ) : <EOL> raise JobLookupError ( job_id ) <EOL> with self . redis . pipeline ( ) as pipe : <EOL> pipe . hdel ( self . jobs_key , job_id ) <EOL> pipe . zrem ( self . run_times_key , job_id ) <EOL> pipe . execute ( ) <EOL> def remove_all_jobs ( self ) : <EOL> with self . redis . pipeline ( ) as pipe : <EOL> pipe . delete ( self . jobs_key ) <EOL> pipe . delete ( self . run_times_key ) <EOL> pipe . execute ( ) <EOL> def shutdown ( self ) : <EOL> self . redis . connection_pool . disconnect ( ) <EOL> def _reconstitute_job ( self , job_state ) : <EOL> job_state = pickle . loads ( job_state ) <EOL> job = Job . __new__ ( Job ) <EOL> job . __setstate__ ( job_state ) <EOL> job . _scheduler = self . _scheduler <EOL> job . _jobstore_alias = self . _alias <EOL> return job <EOL> def _reconstitute_jobs ( self , job_states ) : <EOL> jobs = [ ] <EOL> failed_job_ids = [ ] <EOL> for job_id , job_state in job_states : <EOL> try : <EOL> jobs . append ( self . _reconstitute_job ( job_state ) ) <EOL> except : <EOL> self . _logger . exception ( '<STR_LIT>' , job_id ) <EOL> failed_job_ids . append ( job_id ) <EOL> if failed_job_ids : <EOL> with self . redis . pipeline ( ) as pipe : <EOL> pipe . hdel ( self . jobs_key , * failed_job_ids ) <EOL> pipe . zrem ( self . run_times_key , * failed_job_ids ) <EOL> pipe . execute ( ) <EOL> return jobs <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % self . __class__ . __name__ </s>
