<s> """<STR_LIT>""" <EOL> from . util import AzkabanError , Config , Adapter , MultipartForm , flatten <EOL> from getpass import getpass , getuser <EOL> from os . path import basename , exists <EOL> from requests . exceptions import HTTPError <EOL> from six import string_types <EOL> from six . moves . configparser import NoOptionError , NoSectionError <EOL> from six . moves . urllib . parse import urlparse <EOL> from time import sleep <EOL> from warnings import warn <EOL> import logging as lg <EOL> import requests as rq <EOL> import re <EOL> _logger = lg . getLogger ( __name__ ) <EOL> def _azkaban_request ( method , url , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> response = rq . request ( url = url , method = method , ** kwargs ) <EOL> except rq . ConnectionError as err : <EOL> raise AzkabanError ( '<STR_LIT>' , url , err ) <EOL> except rq . exceptions . MissingSchema : <EOL> raise AzkabanError ( '<STR_LIT>' , url ) <EOL> else : <EOL> return response <EOL> def _extract_json ( response ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> json = response . json ( ) <EOL> except ValueError as err : <EOL> _logger . error ( '<STR_LIT>' , response . text ) <EOL> raise err <EOL> else : <EOL> if '<STR_LIT:error>' in json : <EOL> raise AzkabanError ( json [ '<STR_LIT:error>' ] ) <EOL> elif json . get ( '<STR_LIT:status>' ) == '<STR_LIT:error>' : <EOL> raise AzkabanError ( json [ '<STR_LIT:message>' ] ) <EOL> else : <EOL> return json <EOL> def _parse_url ( url ) : <EOL> """<STR_LIT>""" <EOL> if not re . match ( r'<STR_LIT>' , url ) and not re . search ( r'<STR_LIT>' , url ) : <EOL> url = '<STR_LIT>' + url <EOL> if re . search ( r'<STR_LIT>' , url ) : <EOL> splitted = url . rstrip ( '<STR_LIT:/>' ) . split ( '<STR_LIT:@>' ) <EOL> if len ( splitted ) == <NUM_LIT:1> : <EOL> address = splitted [ <NUM_LIT:0> ] <EOL> user = None <EOL> password = None <EOL> elif len ( splitted ) == <NUM_LIT:2> : <EOL> address = splitted [ <NUM_LIT:1> ] <EOL> creds = splitted [ <NUM_LIT:0> ] . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> if len ( creds ) == <NUM_LIT:1> : <EOL> user = creds [ <NUM_LIT:0> ] <EOL> password = None <EOL> else : <EOL> user , password = creds <EOL> else : <EOL> raise AzkabanError ( '<STR_LIT>' % ( url , ) ) <EOL> return user , password , address <EOL> else : <EOL> parsed = urlparse ( url ) <EOL> return ( parsed . username , parsed . password , <EOL> '<STR_LIT>' % ( parsed . scheme , parsed . hostname , parsed . port ) ) <EOL> class Session ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , url = None , alias = None , config = None , attempts = <NUM_LIT:3> , verify = True <EOL> ) : <EOL> self . attempts = attempts <EOL> self . verify = verify <EOL> self . config = config <EOL> if not url : <EOL> warn ( DeprecationWarning ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> ) ) <EOL> config = config or Config ( ) <EOL> alias = alias or config . get_option ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> url = config . parser . get ( '<STR_LIT>' , alias ) <EOL> except ( NoOptionError , NoSectionError ) : <EOL> raise AzkabanError ( '<STR_LIT>' , alias ) <EOL> self . user , self . password , self . url = _parse_url ( url ) <EOL> if not self . user : <EOL> self . user = getuser ( ) <EOL> self . id = None <EOL> if self . config : <EOL> try : <EOL> key = str ( self ) . replace ( '<STR_LIT::>' , '<STR_LIT:.>' ) <EOL> self . id = self . config . parser . get ( '<STR_LIT>' , key ) <EOL> except ( NoOptionError , NoSectionError ) : <EOL> pass <EOL> self . _logger = Adapter ( repr ( self ) , _logger ) <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , self . user , self . url <EOL> ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . user , self . url ) <EOL> def is_valid ( self , response = None ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> if not self . id : <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> return False <EOL> if response is None : <EOL> self . _logger . debug ( '<STR_LIT>' , self . id ) <EOL> response = _azkaban_request ( <EOL> '<STR_LIT:POST>' , <EOL> '<STR_LIT>' % ( self . url , ) , <EOL> data = { '<STR_LIT>' : self . id } , <EOL> verify = self . verify , <EOL> ) <EOL> if ( <EOL> '<STR_LIT>' in response . text or <EOL> '<STR_LIT>' in response . text or <EOL> '<STR_LIT>' in response . text <EOL> ) : <EOL> self . _logger . debug ( '<STR_LIT>' , self . id , response . text ) <EOL> return False <EOL> else : <EOL> self . _logger . debug ( '<STR_LIT>' , self . id ) <EOL> return True <EOL> def get_workflow_executions ( self , project , flow , start = <NUM_LIT:0> , length = <NUM_LIT:10> ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , project , flow ) <EOL> res = self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : project , <EOL> '<STR_LIT>' : flow , <EOL> '<STR_LIT:start>' : start , <EOL> '<STR_LIT>' : length <EOL> } , <EOL> ) <EOL> if not res . text : <EOL> raise AzkabanError ( <EOL> '<STR_LIT>' , project <EOL> ) <EOL> else : <EOL> return _extract_json ( res ) <EOL> def get_running_workflows ( self , project , flow ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , project , flow ) <EOL> return _extract_json ( self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : project , <EOL> '<STR_LIT>' : flow , <EOL> } , <EOL> ) ) <EOL> def get_execution_status ( self , exec_id ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , exec_id ) <EOL> return _extract_json ( self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : exec_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> ) ) <EOL> def get_execution_logs ( self , exec_id , offset = <NUM_LIT:0> , limit = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , exec_id ) <EOL> return _extract_json ( self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : exec_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : offset , <EOL> '<STR_LIT>' : limit , <EOL> } , <EOL> ) ) <EOL> def get_job_logs ( self , exec_id , job , offset = <NUM_LIT:0> , limit = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , exec_id , job ) <EOL> return _extract_json ( self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : exec_id , <EOL> '<STR_LIT>' : job , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : offset , <EOL> '<STR_LIT>' : limit , <EOL> } , <EOL> ) ) <EOL> def cancel_execution ( self , exec_id ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , exec_id ) <EOL> res = _extract_json ( self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : exec_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> ) ) <EOL> if '<STR_LIT:error>' in res : <EOL> raise AzkabanError ( '<STR_LIT>' , exec_id ) <EOL> else : <EOL> self . _logger . info ( '<STR_LIT>' , exec_id ) <EOL> return res <EOL> def create_project ( self , name , description ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , name ) <EOL> return _extract_json ( self . _request ( <EOL> method = '<STR_LIT:POST>' , <EOL> endpoint = '<STR_LIT>' , <EOL> data = { <EOL> '<STR_LIT:action>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : name , <EOL> '<STR_LIT:description>' : description , <EOL> } , <EOL> ) ) <EOL> def delete_project ( self , name ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , name ) <EOL> res = self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> } , <EOL> ) <EOL> msg = "<STR_LIT>" % ( name , ) <EOL> if not msg in res . text : <EOL> raise AzkabanError ( '<STR_LIT>' ) <EOL> return res <EOL> def run_workflow ( self , name , flow , jobs = None , disabled_jobs = None , <EOL> concurrent = True , properties = None , on_failure = '<STR_LIT>' , notify_early = False , <EOL> emails = None ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , name , flow ) <EOL> request_data = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT>' : flow <EOL> } <EOL> request_data . update ( self . _run_options ( <EOL> name , <EOL> flow , <EOL> jobs = jobs , <EOL> disabled_jobs = disabled_jobs , <EOL> concurrent = concurrent , <EOL> properties = properties , <EOL> on_failure = on_failure , <EOL> notify_early = notify_early , <EOL> emails = emails <EOL> ) ) <EOL> res = _extract_json ( self . _request ( <EOL> method = '<STR_LIT:POST>' , <EOL> endpoint = '<STR_LIT>' , <EOL> include_session = '<STR_LIT>' , <EOL> data = request_data , <EOL> ) ) <EOL> self . _logger . info ( '<STR_LIT>' , name , flow ) <EOL> return res <EOL> def schedule_workflow ( self , name , flow , date , time , period = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , flow , name ) <EOL> request_data = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT>' : self . _get_project_id ( name ) , <EOL> '<STR_LIT>' : flow , <EOL> '<STR_LIT>' : date , <EOL> '<STR_LIT>' : time , <EOL> '<STR_LIT>' : '<STR_LIT>' if period else '<STR_LIT>' , <EOL> } <EOL> if period : <EOL> request_data [ '<STR_LIT>' ] = period <EOL> request_data . update ( self . _run_options ( name , flow , ** kwargs ) ) <EOL> res = _extract_json ( self . _request ( <EOL> method = '<STR_LIT:POST>' , <EOL> endpoint = '<STR_LIT>' , <EOL> data = request_data , <EOL> ) ) <EOL> self . _logger . info ( '<STR_LIT>' , name , flow ) <EOL> return res <EOL> def unschedule_workflow ( self , name , flow ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , flow , name ) <EOL> request_data = { <EOL> '<STR_LIT:action>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . get_schedule ( name , flow ) [ '<STR_LIT>' ] , <EOL> } <EOL> res = _extract_json ( self . _request ( <EOL> method = '<STR_LIT:POST>' , <EOL> endpoint = '<STR_LIT>' , <EOL> data = request_data , <EOL> ) ) <EOL> self . _logger . info ( '<STR_LIT>' , name , flow ) <EOL> return res <EOL> def get_schedule ( self , name , flow ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( <EOL> '<STR_LIT>' , flow , name <EOL> ) <EOL> res = _extract_json ( self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . _get_project_id ( name ) , <EOL> '<STR_LIT>' : flow , <EOL> } , <EOL> ) ) <EOL> self . _logger . info ( <EOL> '<STR_LIT>' , name , flow <EOL> ) <EOL> if '<STR_LIT>' not in res : <EOL> raise AzkabanError ( <EOL> '<STR_LIT>' <EOL> ) <EOL> return res [ '<STR_LIT>' ] <EOL> def _get_project_id ( self , name ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , name ) <EOL> try : <EOL> res = _extract_json ( self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : name , <EOL> } , <EOL> ) ) <EOL> except ValueError : <EOL> raise AzkabanError ( <EOL> '<STR_LIT>' <EOL> ) <EOL> else : <EOL> project_id = res [ '<STR_LIT>' ] <EOL> self . _logger . info ( '<STR_LIT>' , name , project_id ) <EOL> return project_id <EOL> def upload_project ( self , name , path , archive_name = None , callback = None ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' , path , name ) <EOL> if not exists ( path ) : <EOL> raise AzkabanError ( '<STR_LIT>' % ( path , ) ) <EOL> if not self . is_valid ( ) : <EOL> self . _refresh ( ) <EOL> archive_name = archive_name or basename ( path ) <EOL> form = MultipartForm ( <EOL> files = [ { <EOL> '<STR_LIT:path>' : path , <EOL> '<STR_LIT:name>' : archive_name , <EOL> '<STR_LIT:type>' : '<STR_LIT>' <EOL> } ] , <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT>' : self . id , <EOL> } , <EOL> callback = callback <EOL> ) <EOL> res = _extract_json ( self . _request ( <EOL> method = '<STR_LIT:POST>' , <EOL> endpoint = '<STR_LIT>' , <EOL> include_session = False , <EOL> headers = form . headers , <EOL> data = form , <EOL> ) ) <EOL> self . _logger . info ( <EOL> '<STR_LIT>' , path , name , archive_name <EOL> ) <EOL> return res <EOL> def get_workflow_info ( self , name , flow ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( <EOL> '<STR_LIT>' , flow , name <EOL> ) <EOL> try : <EOL> res = self . _request ( <EOL> method = '<STR_LIT:GET>' , <EOL> endpoint = '<STR_LIT>' , <EOL> params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT>' : flow , <EOL> } , <EOL> ) <EOL> except HTTPError : <EOL> raise AzkabanError ( '<STR_LIT>' , flow , name ) <EOL> else : <EOL> try : <EOL> return _extract_json ( res ) <EOL> except ValueError : <EOL> raise AzkabanError ( '<STR_LIT>' , name ) <EOL> def _refresh ( self , password = None ) : <EOL> """<STR_LIT>""" <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> attempts = self . attempts <EOL> password = password or self . password <EOL> while True : <EOL> password = password or getpass ( '<STR_LIT>' % ( self , ) ) <EOL> try : <EOL> res = _extract_json ( _azkaban_request ( <EOL> '<STR_LIT:POST>' , <EOL> self . url , <EOL> data = { <EOL> '<STR_LIT:action>' : '<STR_LIT>' , <EOL> '<STR_LIT:username>' : self . user , <EOL> '<STR_LIT:password>' : password , <EOL> } , <EOL> verify = self . verify , <EOL> ) ) <EOL> except AzkabanError as err : <EOL> if not '<STR_LIT>' in err . message : <EOL> raise err <EOL> self . _logger . warning ( '<STR_LIT>' ) <EOL> attempts -= <NUM_LIT:1> <EOL> password = None <EOL> if attempts <= <NUM_LIT:0> : <EOL> raise AzkabanError ( '<STR_LIT>' ) <EOL> else : <EOL> break <EOL> self . id = res [ '<STR_LIT>' ] <EOL> if self . config : <EOL> if not self . config . parser . has_section ( '<STR_LIT>' ) : <EOL> self . config . parser . add_section ( '<STR_LIT>' ) <EOL> self . config . parser . set ( <EOL> '<STR_LIT>' , <EOL> str ( self ) . replace ( '<STR_LIT::>' , '<STR_LIT:.>' ) , <EOL> self . id <EOL> ) <EOL> self . config . save ( ) <EOL> self . _logger . info ( '<STR_LIT>' ) <EOL> def _run_options ( self , name , flow , jobs = None , disabled_jobs = None , <EOL> concurrent = True , properties = None , on_failure = '<STR_LIT>' , notify_early = False , <EOL> emails = None ) : <EOL> """<STR_LIT>""" <EOL> if jobs and disabled_jobs : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not jobs : <EOL> if not disabled_jobs : <EOL> disabled = '<STR_LIT>' <EOL> else : <EOL> disabled = '<STR_LIT>' % ( '<STR_LIT:U+002C>' . join ( '<STR_LIT>' % ( n , ) for n in disabled_jobs ) , ) <EOL> else : <EOL> all_names = set ( <EOL> n [ '<STR_LIT:id>' ] <EOL> for n in self . get_workflow_info ( name , flow ) [ '<STR_LIT>' ] <EOL> ) <EOL> run_names = set ( jobs ) <EOL> missing_names = run_names - all_names <EOL> if missing_names : <EOL> raise AzkabanError ( <EOL> '<STR_LIT>' % <EOL> ( flow , '<STR_LIT:U+002CU+0020>' . join ( missing_names ) ) <EOL> ) <EOL> else : <EOL> disabled = ( <EOL> '<STR_LIT>' <EOL> % ( '<STR_LIT:U+002C>' . join ( '<STR_LIT>' % ( n , ) for n in all_names - run_names ) , ) <EOL> ) <EOL> try : <EOL> failure_action = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } [ on_failure ] <EOL> except KeyError : <EOL> raise ValueError ( '<STR_LIT>' % ( on_failure , ) ) <EOL> request_data = { <EOL> '<STR_LIT>' : disabled , <EOL> '<STR_LIT>' : '<STR_LIT>' if concurrent else '<STR_LIT>' , <EOL> '<STR_LIT>' : failure_action , <EOL> '<STR_LIT>' : '<STR_LIT:true>' if notify_early else '<STR_LIT:false>' , <EOL> } <EOL> if properties : <EOL> request_data . update ( dict ( <EOL> ( '<STR_LIT>' % ( key , ) , value ) <EOL> for key , value in flatten ( properties ) . items ( ) <EOL> ) ) <EOL> if emails : <EOL> if isinstance ( emails [ <NUM_LIT:0> ] , string_types ) : <EOL> failure_emails = '<STR_LIT:U+002C>' . join ( emails ) <EOL> success_emails = failure_emails <EOL> else : <EOL> failure_emails = '<STR_LIT:U+002C>' . join ( emails [ <NUM_LIT:0> ] ) <EOL> success_emails = '<STR_LIT:U+002C>' . join ( emails [ <NUM_LIT:1> ] ) <EOL> request_data . update ( { <EOL> '<STR_LIT>' : failure_emails , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : success_emails , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> } ) <EOL> return request_data <EOL> def _request ( self , method , endpoint , include_session = '<STR_LIT>' , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> full_url = '<STR_LIT>' % ( self . url , endpoint . lstrip ( '<STR_LIT:/>' ) ) <EOL> if not self . id : <EOL> self . _logger . debug ( '<STR_LIT>' ) <EOL> self . _refresh ( ) <EOL> def _send_request ( ) : <EOL> """<STR_LIT>""" <EOL> if include_session == '<STR_LIT>' : <EOL> kwargs . setdefault ( '<STR_LIT>' , { } ) [ '<STR_LIT>' ] = self . id <EOL> elif include_session == '<STR_LIT>' : <EOL> kwargs . setdefault ( '<STR_LIT:data>' , { } ) [ '<STR_LIT>' ] = self . id <EOL> elif include_session : <EOL> raise ValueError ( '<STR_LIT>' % ( include_session , ) ) <EOL> return _azkaban_request ( method , full_url , verify = self . verify , ** kwargs ) <EOL> response = _send_request ( ) <EOL> if not self . is_valid ( response ) : <EOL> self . _refresh ( ) <EOL> response = _send_request ( ) <EOL> if not self . is_valid ( response ) : <EOL> raise AzkabanError ( '<STR_LIT>' ) <EOL> try : <EOL> response . raise_for_status ( ) <EOL> except HTTPError as err : <EOL> self . _logger . warning ( <EOL> '<STR_LIT>' , <EOL> response . request . url , response . content <EOL> ) <EOL> raise err <EOL> else : <EOL> return response <EOL> @ classmethod <EOL> def from_alias ( cls , alias , config = None ) : <EOL> """<STR_LIT>""" <EOL> config = config or Config ( ) <EOL> section_name = '<STR_LIT>' % ( alias , ) <EOL> try : <EOL> url = config . parser . get ( section_name , '<STR_LIT:url>' ) <EOL> except NoSectionError : <EOL> raise AzkabanError ( '<STR_LIT>' % ( alias , ) ) <EOL> except NoOptionError : <EOL> raise AzkabanError ( '<STR_LIT>' % ( alias , ) ) <EOL> else : <EOL> opts = { '<STR_LIT:url>' : url , '<STR_LIT>' : config } <EOL> if config . parser . has_option ( section_name , '<STR_LIT>' ) : <EOL> opts [ '<STR_LIT>' ] = config . parser . getboolean ( section_name , '<STR_LIT>' ) <EOL> if config . parser . has_option ( section_name , '<STR_LIT>' ) : <EOL> opts [ '<STR_LIT>' ] = config . parser . getint ( section_name , '<STR_LIT>' ) <EOL> return Session ( ** opts ) <EOL> class Execution ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , session , exec_id ) : <EOL> self . _session = session <EOL> self . exec_id = exec_id <EOL> @ property <EOL> def status ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _session . get_execution_status ( self . exec_id ) <EOL> @ property <EOL> def url ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( self . _session . url , self . exec_id ) <EOL> def cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _session . cancel_execution ( self . exec_id ) <EOL> def logs ( self , delay = <NUM_LIT:5> ) : <EOL> """<STR_LIT>""" <EOL> finishing = False <EOL> offset = <NUM_LIT:0> <EOL> while True : <EOL> logs = self . _session . get_execution_logs ( <EOL> exec_id = self . exec_id , <EOL> offset = offset , <EOL> ) <EOL> if logs [ '<STR_LIT>' ] : <EOL> offset += logs [ '<STR_LIT>' ] <EOL> lines = ( e for e in logs [ '<STR_LIT:data>' ] . split ( '<STR_LIT:\n>' ) if e ) <EOL> for line in lines : <EOL> yield line <EOL> elif finishing : <EOL> break <EOL> else : <EOL> if self . status [ '<STR_LIT:status>' ] != '<STR_LIT>' : <EOL> finishing = True <EOL> sleep ( delay ) <EOL> def job_logs ( self , job , delay = <NUM_LIT:5> ) : <EOL> """<STR_LIT>""" <EOL> finishing = False <EOL> offset = <NUM_LIT:0> <EOL> while True : <EOL> try : <EOL> logs = self . _session . get_job_logs ( <EOL> exec_id = self . exec_id , <EOL> job = job , <EOL> offset = offset , <EOL> ) <EOL> except HTTPError as err : <EOL> preparing = False <EOL> while True : <EOL> sleep ( delay ) <EOL> preparing_jobs = set ( <EOL> e [ '<STR_LIT:id>' ] <EOL> for e in self . status [ '<STR_LIT>' ] <EOL> if e [ '<STR_LIT:status>' ] == '<STR_LIT>' <EOL> ) <EOL> if job in preparing_jobs : <EOL> if not preparing : <EOL> preparing = True <EOL> _logger . debug ( <EOL> '<STR_LIT>' , job , self . exec_id <EOL> ) <EOL> else : <EOL> break <EOL> if not preparing : <EOL> raise err <EOL> else : <EOL> if logs [ '<STR_LIT>' ] : <EOL> offset += logs [ '<STR_LIT>' ] <EOL> lines = ( e for e in logs [ '<STR_LIT:data>' ] . split ( '<STR_LIT:\n>' ) if e ) <EOL> for line in lines : <EOL> yield line <EOL> elif finishing : <EOL> break <EOL> else : <EOL> running_jobs = set ( <EOL> e [ '<STR_LIT:id>' ] <EOL> for e in self . status [ '<STR_LIT>' ] <EOL> if e [ '<STR_LIT:status>' ] == '<STR_LIT>' <EOL> ) <EOL> if job not in running_jobs : <EOL> finishing = True <EOL> sleep ( delay ) <EOL> @ classmethod <EOL> def start ( cls , session , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> res = session . run_workflow ( * args , ** kwargs ) <EOL> return cls ( session , res [ '<STR_LIT>' ] ) </s>
