<s> from collections import defaultdict <EOL> from datetime import datetime <EOL> from botocore . client import Config <EOL> import arrow <EOL> import boto3 <EOL> from mass . scheduler . swf import config <EOL> def workflows_to_jobs ( workflows ) : <EOL> workflows = sorted ( workflows , key = lambda w : w [ '<STR_LIT>' ] ) <EOL> workflows = [ defaultdict ( list , w ) for w in workflows ] <EOL> jobs = [ ] <EOL> def insert ( workflow , roots ) : <EOL> if '<STR_LIT>' not in workflow : <EOL> jobs . append ( workflow ) <EOL> return <EOL> parent = [ w for w in roots if w [ '<STR_LIT>' ] [ '<STR_LIT>' ] == workflow [ '<STR_LIT>' ] [ '<STR_LIT>' ] ] <EOL> if parent : <EOL> parent [ <NUM_LIT:0> ] [ '<STR_LIT>' ] . append ( workflow ) <EOL> return True <EOL> else : <EOL> for root in roots : <EOL> if insert ( workflow , root [ '<STR_LIT>' ] ) : <EOL> return True <EOL> return False <EOL> for w in workflows : <EOL> insert ( w , jobs ) <EOL> return jobs <EOL> def datetime_to_timestamp ( data ) : <EOL> if isinstance ( data , datetime ) : <EOL> return arrow . get ( data ) . timestamp <EOL> elif isinstance ( data , dict ) : <EOL> for k , v in data . items ( ) : <EOL> data [ k ] = datetime_to_timestamp ( v ) <EOL> return data <EOL> elif isinstance ( data , list ) : <EOL> return [ datetime_to_timestamp ( d ) for d in data ] <EOL> else : <EOL> return data <EOL> def list_workflow_executions ( region , domain , ** kwargs ) : <EOL> client = boto3 . client ( <EOL> '<STR_LIT>' , <EOL> region_name = self . region , <EOL> config = Config ( connect_timeout = config . CONNECT_TIMEOUT , <EOL> read_timeout = config . READ_TIMEOUT ) ) <EOL> paginator = client . get_paginator ( '<STR_LIT>' ) <EOL> for res in paginator . paginate ( domain = domain , ** kwargs ) : <EOL> for workflow in res . get ( '<STR_LIT>' , [ ] ) : <EOL> yield workflow <EOL> paginator = client . get_paginator ( '<STR_LIT>' ) <EOL> for res in paginator . paginate ( domain = domain , ** kwargs ) : <EOL> for workflow in res . get ( '<STR_LIT>' , [ ] ) : <EOL> yield workflow <EOL> def describe_workflow_execution ( region , domain , ** kwargs ) : <EOL> client = boto3 . client ( <EOL> '<STR_LIT>' , <EOL> region_name = self . region , <EOL> config = Config ( connect_timeout = config . CONNECT_TIMEOUT , <EOL> read_timeout = config . READ_TIMEOUT ) ) <EOL> res = client . describe_workflow_execution ( domain = domain , ** kwargs ) <EOL> return res <EOL> def get_workflow_execution_history ( region , domain , ** kwargs ) : <EOL> client = boto3 . client ( <EOL> '<STR_LIT>' , <EOL> region_name = self . region , <EOL> config = Config ( connect_timeout = config . CONNECT_TIMEOUT , <EOL> read_timeout = config . READ_TIMEOUT ) ) <EOL> paginator = client . get_paginator ( '<STR_LIT>' ) <EOL> for res in paginator . paginate ( domain = domain , ** kwargs ) : <EOL> for event in res [ '<STR_LIT>' ] : <EOL> yield event <EOL> def list_jobs ( region , domain , oldest ) : <EOL> jobs = [ ] <EOL> for workflow in list_workflow_executions ( <EOL> region = region , <EOL> domain = domain , <EOL> startTimeFilter = { <EOL> '<STR_LIT>' : oldest <EOL> } , <EOL> typeFilter = config . WORKFLOW_TYPE_FOR_JOB ) : <EOL> keys = [ k for k in workflow . keys ( ) ] <EOL> for k , v in workflow . items ( ) : <EOL> if isinstance ( v , datetime ) : <EOL> workflow [ k ] = arrow . get ( v ) . timestamp <EOL> jobs . append ( workflow ) <EOL> return jobs <EOL> def retrieve_jobs ( region , domain , workflow_id , oldest ) : <EOL> roots = list_workflow_executions ( <EOL> region = region , <EOL> domain = domain , <EOL> executionFilter = { <EOL> '<STR_LIT>' : workflow_id <EOL> } , <EOL> startTimeFilter = { <EOL> '<STR_LIT>' : oldest <EOL> } ) <EOL> try : <EOL> start_time = min ( map ( lambda r : r [ '<STR_LIT>' ] , roots ) ) <EOL> except ValueError : <EOL> return [ ] <EOL> workflows = list_workflow_executions ( <EOL> region = region , <EOL> domain = domain , <EOL> startTimeFilter = { <EOL> '<STR_LIT>' : start_time <EOL> } , <EOL> tagFilter = { <EOL> '<STR_LIT>' : workflow_id <EOL> } ) <EOL> jobs = workflows_to_jobs ( workflows ) <EOL> jobs = datetime_to_timestamp ( jobs ) <EOL> return jobs <EOL> def retrieve_job_history ( region , domain , workflow_id , run_id , reverse = False ) : <EOL> for event in get_workflow_execution_history ( <EOL> region = region , <EOL> domain = domain , <EOL> execution = { <EOL> '<STR_LIT>' : workflow_id , <EOL> '<STR_LIT>' : run_id <EOL> } , <EOL> reverseOrder = reverse ) : <EOL> yield datetime_to_timestamp ( event ) </s>
