<s> from __future__ import absolute_import <EOL> from apscheduler . jobstores . base import BaseJobStore , JobLookupError , ConflictingIdError <EOL> from apscheduler . util import maybe_ref , 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 sqlalchemy import ( <EOL> create_engine , Table , Column , MetaData , Unicode , Float , LargeBinary , select ) <EOL> from sqlalchemy . exc import IntegrityError <EOL> from sqlalchemy . sql . expression import null <EOL> except ImportError : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> class SQLAlchemyJobStore ( BaseJobStore ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , url = None , engine = None , tablename = '<STR_LIT>' , metadata = None , <EOL> pickle_protocol = pickle . HIGHEST_PROTOCOL ) : <EOL> super ( SQLAlchemyJobStore , self ) . __init__ ( ) <EOL> self . pickle_protocol = pickle_protocol <EOL> metadata = maybe_ref ( metadata ) or MetaData ( ) <EOL> if engine : <EOL> self . engine = maybe_ref ( engine ) <EOL> elif url : <EOL> self . engine = create_engine ( url ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . jobs_t = Table ( <EOL> tablename , metadata , <EOL> Column ( '<STR_LIT:id>' , Unicode ( <NUM_LIT> , _warn_on_bytestring = False ) , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , Float ( <NUM_LIT> ) , index = True ) , <EOL> Column ( '<STR_LIT>' , LargeBinary , nullable = False ) <EOL> ) <EOL> def start ( self , scheduler , alias ) : <EOL> super ( SQLAlchemyJobStore , self ) . start ( scheduler , alias ) <EOL> self . jobs_t . create ( self . engine , True ) <EOL> def lookup_job ( self , job_id ) : <EOL> selectable = select ( [ self . jobs_t . c . job_state ] ) . where ( self . jobs_t . c . id == job_id ) <EOL> job_state = self . engine . execute ( selectable ) . scalar ( ) <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> return self . _get_jobs ( self . jobs_t . c . next_run_time <= timestamp ) <EOL> def get_next_run_time ( self ) : <EOL> 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 ( <NUM_LIT:1> ) <EOL> next_run_time = self . engine . execute ( selectable ) . scalar ( ) <EOL> return utc_timestamp_to_datetime ( next_run_time ) <EOL> def get_all_jobs ( self ) : <EOL> jobs = self . _get_jobs ( ) <EOL> self . _fix_paused_jobs_sorting ( jobs ) <EOL> return jobs <EOL> def add_job ( self , job ) : <EOL> insert = self . jobs_t . insert ( ) . values ( ** { <EOL> '<STR_LIT:id>' : job . id , <EOL> '<STR_LIT>' : datetime_to_utc_timestamp ( job . next_run_time ) , <EOL> '<STR_LIT>' : pickle . dumps ( job . __getstate__ ( ) , self . pickle_protocol ) <EOL> } ) <EOL> try : <EOL> self . engine . execute ( insert ) <EOL> except IntegrityError : <EOL> raise ConflictingIdError ( job . id ) <EOL> def update_job ( self , job ) : <EOL> update = self . jobs_t . update ( ) . values ( ** { <EOL> '<STR_LIT>' : datetime_to_utc_timestamp ( job . next_run_time ) , <EOL> '<STR_LIT>' : pickle . dumps ( job . __getstate__ ( ) , self . pickle_protocol ) <EOL> } ) . where ( self . jobs_t . c . id == job . id ) <EOL> result = self . engine . execute ( update ) <EOL> if result . rowcount == <NUM_LIT:0> : <EOL> raise JobLookupError ( id ) <EOL> def remove_job ( self , job_id ) : <EOL> delete = self . jobs_t . delete ( ) . where ( self . jobs_t . c . id == job_id ) <EOL> result = self . engine . execute ( delete ) <EOL> if result . rowcount == <NUM_LIT:0> : <EOL> raise JobLookupError ( job_id ) <EOL> def remove_all_jobs ( self ) : <EOL> delete = self . jobs_t . delete ( ) <EOL> self . engine . execute ( delete ) <EOL> def shutdown ( self ) : <EOL> self . engine . dispose ( ) <EOL> def _reconstitute_job ( self , job_state ) : <EOL> job_state = pickle . loads ( job_state ) <EOL> job_state [ '<STR_LIT>' ] = self <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 _get_jobs ( self , * conditions ) : <EOL> jobs = [ ] <EOL> selectable = select ( [ self . jobs_t . c . id , self . jobs_t . c . job_state ] ) . order_by ( self . jobs_t . c . next_run_time ) <EOL> selectable = selectable . where ( * conditions ) if conditions else selectable <EOL> failed_job_ids = set ( ) <EOL> for row in self . engine . execute ( selectable ) : <EOL> try : <EOL> jobs . append ( self . _reconstitute_job ( row . job_state ) ) <EOL> except : <EOL> self . _logger . exception ( '<STR_LIT>' , row . id ) <EOL> failed_job_ids . add ( row . id ) <EOL> if failed_job_ids : <EOL> delete = self . jobs_t . delete ( ) . where ( self . jobs_t . c . id . in_ ( failed_job_ids ) ) <EOL> self . engine . execute ( delete ) <EOL> return jobs <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . engine . url ) </s>
