<s> from collections import OrderedDict <EOL> import uuid <EOL> from django . conf import settings <EOL> from django . db import models , transaction <EOL> from django . utils . timezone import now <EOL> from picklefield . fields import PickledObjectField <EOL> from six import integer_types <EOL> from . queue import Queue <EOL> from . job import Job <EOL> PQ_DEFAULT_JOB_TIMEOUT = getattr ( settings , '<STR_LIT>' , <NUM_LIT> ) <EOL> class FlowQueue ( Queue ) : <EOL> class Meta : <EOL> proxy = True <EOL> def enqueue_job ( self , job , timeout = None , set_meta_data = True , async = True ) : <EOL> """<STR_LIT>""" <EOL> if set_meta_data : <EOL> job . origin = self . name <EOL> if timeout : <EOL> job . timeout = timeout <EOL> else : <EOL> job . timeout = PQ_DEFAULT_JOB_TIMEOUT <EOL> job . uuid = uuid . uuid4 ( ) <EOL> if self . jobs : <EOL> try : <EOL> prior_job = self . jobs . values ( ) [ - <NUM_LIT:1> ] <EOL> except TypeError : <EOL> prior_job = self . jobs . popitem ( ) [ <NUM_LIT:1> ] <EOL> prior_job . if_result = job . uuid <EOL> self . jobs [ prior_job . uuid ] = prior_job <EOL> else : <EOL> job . queue_id = self . name <EOL> self . jobs [ job . uuid ] = job <EOL> return job <EOL> class FlowStore ( models . Model ) : <EOL> """<STR_LIT>""" <EOL> QUEUED = <NUM_LIT:1> <EOL> FINISHED = <NUM_LIT:2> <EOL> FAILED = <NUM_LIT:3> <EOL> STATUS_CHOICES = ( <EOL> ( QUEUED , '<STR_LIT>' ) , <EOL> ( FINISHED , '<STR_LIT>' ) , <EOL> ( FAILED , '<STR_LIT>' ) , <EOL> ) <EOL> name = models . CharField ( max_length = <NUM_LIT:100> , default = '<STR_LIT>' ) <EOL> queue = models . ForeignKey ( '<STR_LIT>' , blank = True , null = True ) <EOL> enqueued_at = models . DateTimeField ( null = True , blank = True ) <EOL> ended_at = models . DateTimeField ( null = True , blank = True ) <EOL> expired_at = models . DateTimeField ( '<STR_LIT>' , null = True , blank = True ) <EOL> status = models . PositiveIntegerField ( null = True , <EOL> blank = True , choices = STATUS_CHOICES ) <EOL> jobs = PickledObjectField ( blank = True ) <EOL> class Meta : <EOL> verbose_name = '<STR_LIT>' <EOL> verbose_name_plural = '<STR_LIT>' <EOL> def __unicode__ ( self ) : <EOL> if self . id and self . name : <EOL> return '<STR_LIT>' % ( self . id , self . name ) <EOL> elif self . id : <EOL> return str ( self . id ) <EOL> else : <EOL> return self . name <EOL> def enqueue ( self , func , * args , ** kwargs ) : <EOL> return self . queue . enqueue ( func , * args , ** kwargs ) <EOL> def enqueue_call ( self , func , * args , ** kwargs ) : <EOL> return self . queue . enqueue_call ( func , * args , ** kwargs ) <EOL> def schedule ( self , * args , ** kwargs ) : <EOL> return self . queue . schedule ( * args , ** kwargs ) <EOL> def schedule_call ( self , * args , ** kwargs ) : <EOL> return self . queue . schedule_call ( * args , ** kwargs ) <EOL> @ classmethod <EOL> def delete_expired_ttl ( cls , connection ) : <EOL> """<STR_LIT>""" <EOL> with transaction . commit_on_success ( using = connection ) : <EOL> FlowStore . objects . using ( connection ) . filter ( <EOL> status = FlowStore . FINISHED , expired_at__lte = now ( ) ) . delete ( ) <EOL> def save ( self , * args , ** kwargs ) : <EOL> self . queue . save_queue ( ) <EOL> super ( FlowStore , self ) . save ( * args , ** kwargs ) <EOL> class Flow ( object ) : <EOL> def __init__ ( self , queue , name = '<STR_LIT>' ) : <EOL> queue = FlowQueue . create ( name = queue . name ) <EOL> self . flowstore = FlowStore ( name = name ) <EOL> self . flowstore . jobs = [ ] <EOL> self . flowstore . queue = queue <EOL> self . flowstore . save ( ) <EOL> queue . jobs = OrderedDict ( ) <EOL> self . queue = queue <EOL> self . name = name <EOL> self . async = queue . _async <EOL> def __enter__ ( self ) : <EOL> return self . flowstore <EOL> def __exit__ ( self , type , value , traceback ) : <EOL> for i , job in enumerate ( self . queue . jobs . values ( ) ) : <EOL> job . flow = self . flowstore <EOL> if not job . queue_id : <EOL> job . status = Job . FLOW <EOL> else : <EOL> job . status = Job . QUEUED <EOL> if self . async : <EOL> job . save ( ) <EOL> if i == <NUM_LIT:0> : <EOL> self . flowstore . enqueued_at = job . enqueued_at <EOL> self . flowstore . jobs . append ( job . id ) <EOL> if job . queue_id : <EOL> self . queue . notify ( job . id ) <EOL> else : <EOL> job . perform ( ) <EOL> job . save ( ) <EOL> if self . async and self . queue . jobs : <EOL> self . flowstore . status = FlowStore . QUEUED <EOL> self . flowstore . save ( ) <EOL> @ classmethod <EOL> def get ( cls , id_or_name ) : <EOL> if isinstance ( id_or_name , integer_types ) : <EOL> return FlowStore . objects . get ( pk = id_or_name ) <EOL> else : <EOL> return FlowStore . objects . filter ( name = id_or_name ) <EOL> @ classmethod <EOL> def handle_result ( cls , job , queue ) : <EOL> """<STR_LIT>""" <EOL> if job . if_result : <EOL> next_job = Job . objects . get ( uuid = job . if_result ) <EOL> next_job . queue_id = queue . name <EOL> next_job . enqueued_at = now ( ) <EOL> next_job . status = Job . QUEUED <EOL> next_job . save ( ) <EOL> queue . notify ( next_job . id ) <EOL> else : <EOL> fs = FlowStore . objects . get ( pk = job . flow_id ) <EOL> if fs . jobs [ - <NUM_LIT:1> ] == job . id and job . expired_at < now ( ) : <EOL> fs . delete ( ) <EOL> elif fs . jobs [ - <NUM_LIT:1> ] == job . id : <EOL> fs . ended_at = job . ended_at <EOL> fs . expired_at = job . expired_at <EOL> fs . status = FlowStore . FINISHED <EOL> fs . save ( ) <EOL> @ classmethod <EOL> def handle_failed ( cls , job , queue ) : <EOL> """<STR_LIT>""" <EOL> if job . if_failed : <EOL> next_job = Job . objects . get ( uuid = job . if_failed ) <EOL> next_job . queue_id = queue . name <EOL> next_job . enqueued_at = now ( ) <EOL> next_job . status = Job . QUEUED <EOL> next_job . save ( ) <EOL> queue . notify ( next_job . id ) <EOL> fs = FlowStore . objects . get ( pk = job . flow_id ) <EOL> fs . status = FlowStore . FAILED <EOL> fs . save ( ) </s>
