signature:
def _predict ( self , my_task , seen = None , looked_ahead = 0 ) :
    """Updates the branch such that all possible future routes are added .
 Should NOT be overwritten ! Instead , overwrite _ predict _ hook ( ) .
 : type my _ task : Task
 : param my _ task : The associated task in the task tree .
 : type seen : list [ taskspec ]
 : param seen : A list of already visited tasks .
 : type looked _ ahead : integer
 : param looked _ ahead : The depth of the predicted path so far ."""

implementation:
if my_task . _is_finished ( ) :
    return
if seen is None :
    seen = [ ]
elif self in seen :
    return
if not my_task . _is_finished ( ) :
    self . _predict_hook ( my_task )
if not my_task . _is_definite ( ) :
    if looked_ahead + 1 >= self . lookahead :
        return
    seen . append ( self )
for child in my_task . children :
    child . task_spec . _predict ( child , seen [ : ] , looked_ahead + 1 )


signature:
def _update_hook ( self , my_task ) :
    """Typically this method should perform the following actions : :
 - Update the state of the corresponding task .
 - Update the predictions for its successors .
 Returning non - False will cause the task to go into READY .
 Returning any other value will cause no action ."""

implementation:
if my_task . _is_predicted ( ) :
    self . _predict ( my_task )
LOG . debug ( "'%s'._update_hook says parent (%s, state=%s) " "is_finished=%s" % ( self . name , my_task . parent . get_name ( ) , my_task . parent . get_state_name ( ) , my_task . parent . _is_finished ( ) ) )
if not my_task . parent . _is_finished ( ) :
    return
self . entered_event . emit ( my_task . workflow , my_task )
my_task . _ready ( )


signature:
def _on_ready ( self , my_task ) :
    """Return True on success , False otherwise .
 : type my _ task : Task
 : param my _ task : The associated task in the task tree ."""

implementation:
assert my_task is not None
self . test ( )
# Acquire locks , if any .
for lock in self . locks :
    mutex = my_task . workflow . _get_mutex ( lock )
    if not mutex . testandset ( ) :
        return
# Assign variables , if so requested .
for assignment in self . pre_assign :
    assignment . assign ( my_task , my_task )
# Run task - specific code .
self . _on_ready_before_hook ( my_task )
self . reached_event . emit ( my_task . workflow , my_task )
self . _on_ready_hook ( my_task )
# Run user code , if any .
if self . ready_event . emit ( my_task . workflow , my_task ) : # Assign variables , if so requested .
    for assignment in self . post_assign :
        assignment . assign ( my_task , my_task )
# Release locks , if any .
for lock in self . locks :
    mutex = my_task . workflow . _get_mutex ( lock )
    mutex . unlock ( )
self . finished_event . emit ( my_task . workflow , my_task )


signature:
def _on_complete ( self , my_task ) :
    """Return True on success , False otherwise . Should not be overwritten ,
 overwrite _ on _ complete _ hook ( ) instead .
 : type my _ task : Task
 : param my _ task : The associated task in the task tree .
 : rtype : boolean
 : returns : True on success , False otherwise ."""

implementation:
assert my_task is not None
if my_task . workflow . debug :
    print ( "Executing %s: %s (%s)" % ( my_task . task_spec . __class__ . __name__ , my_task . get_name ( ) , my_task . get_description ( ) ) )
self . _on_complete_hook ( my_task )
# Notify the Workflow .
my_task . workflow . _task_completed_notify ( my_task )
if my_task . workflow . debug :
    if hasattr ( my_task . workflow , "outer_workflow" ) :
        my_task . workflow . outer_workflow . task_tree . dump ( )
self . completed_event . emit ( my_task . workflow , my_task )
return True


signature:
def create_package ( self ) :
    """Creates the package , writing the data out to the provided file - like
 object ."""

implementation:
# Check that all files exist ( and calculate the longest shared path
# prefix ) :
self . input_path_prefix = None
for filename in self . input_files :
    if not os . path . isfile ( filename ) :
        raise ValueError ( '%s does not exist or is not a file' % filename )
    if self . input_path_prefix :
        full = os . path . abspath ( os . path . dirname ( filename ) )
        while not ( full . startswith ( self . input_path_prefix ) and self . input_path_prefix ) :
            self . input_path_prefix = self . input_path_prefix [ : - 1 ]
    else :
        self . input_path_prefix = os . path . abspath ( os . path . dirname ( filename ) )
# Parse all of the XML :
self . bpmn = { }
for filename in self . input_files :
    bpmn = ET . parse ( filename )
    self . bpmn [ os . path . abspath ( filename ) ] = bpmn
# Now run through pre - parsing and validation :
for filename , bpmn in list ( self . bpmn . items ( ) ) :
    bpmn = self . pre_parse_and_validate ( bpmn , filename )
    self . bpmn [ os . path . abspath ( filename ) ] = bpmn
# Now check that we can parse it fine :
for filename , bpmn in list ( self . bpmn . items ( ) ) :
    self . parser . add_bpmn_xml ( bpmn , filename = filename )
self . wf_spec = self . parser . get_spec ( self . entry_point_process )
# Now package everything :
self . package_zip = zipfile . ZipFile ( self . package_file , "w" , compression = zipfile . ZIP_DEFLATED )
done_files = set ( )
for spec in self . wf_spec . get_specs_depth_first ( ) :
    filename = spec . file
    if filename not in done_files :
        done_files . add ( filename )
        bpmn = self . bpmn [ os . path . abspath ( filename ) ]
        self . write_to_package_zip ( "%s.bpmn" % spec . name , ET . tostring ( bpmn . getroot ( ) ) )
        self . write_file_to_package_zip ( "src/" + self . _get_zip_path ( filename ) , filename )
        self . _call_editor_hook ( 'package_for_editor' , spec , filename )
self . write_meta_data ( )
self . write_manifest ( )
self . package_zip . close ( )


signature:
def write_file_to_package_zip ( self , filename , src_filename ) :
    """Writes a local file in to the zip file and adds it to the manifest
 dictionary
 : param filename : The zip file name
 : param src _ filename : the local file name"""

implementation:
f = open ( src_filename )
with f :
    data = f . read ( )
self . manifest [ filename ] = md5hash ( data )
self . package_zip . write ( src_filename , filename )


signature:
def write_to_package_zip ( self , filename , data ) :
    """Writes data to the zip file and adds it to the manifest dictionary
 : param filename : The zip file name
 : param data : the data"""

implementation:
self . manifest [ filename ] = md5hash ( data )
self . package_zip . writestr ( filename , data )


signature:
def write_manifest ( self ) :
    """Write the manifest content to the zip file . It must be a predictable
 order ."""

implementation:
config = configparser . ConfigParser ( )
config . add_section ( 'Manifest' )
for f in sorted ( self . manifest . keys ( ) ) :
    config . set ( 'Manifest' , f . replace ( '\\' , '/' ) . lower ( ) , self . manifest [ f ] )
ini = StringIO ( )
config . write ( ini )
self . manifest_data = ini . getvalue ( )
self . package_zip . writestr ( self . MANIFEST_FILE , self . manifest_data )


signature:
def pre_parse_and_validate ( self , bpmn , filename ) :
    """A subclass can override this method to provide additional parseing or
 validation . It should call the parent method first .
 : param bpmn : an lxml tree of the bpmn content
 : param filename : the source file name
 This must return the updated bpmn object ( or a replacement )"""

implementation:
bpmn = self . _call_editor_hook ( 'pre_parse_and_validate' , bpmn , filename ) or bpmn
return bpmn


signature:
def pre_parse_and_validate_signavio ( self , bpmn , filename ) :
    """This is the Signavio specific editor hook for pre - parsing and
 validation .
 A subclass can override this method to provide additional parseing or
 validation . It should call the parent method first .
 : param bpmn : an lxml tree of the bpmn content
 : param filename : the source file name
 This must return the updated bpmn object ( or a replacement )"""

implementation:
self . _check_for_disconnected_boundary_events_signavio ( bpmn , filename )
self . _fix_call_activities_signavio ( bpmn , filename )
return bpmn


signature:
def _fix_call_activities_signavio ( self , bpmn , filename ) :
    """Signavio produces slightly invalid BPMN for call activity nodes . . . It
 is supposed to put a reference to the id of the called process in to
 the calledElement attribute . Instead it stores a string ( which is the
 name of the process - not its ID , in our interpretation ) in an
 extension tag .
 This code gets the name of the ' subprocess reference ' , finds a process
 with a matching name , and sets the calledElement attribute to the id of
 the process ."""

implementation:
for node in xpath_eval ( bpmn ) ( ".//bpmn:callActivity" ) :
    calledElement = node . get ( 'calledElement' , None )
    if not calledElement :
        signavioMetaData = xpath_eval ( node , extra_ns = { 'signavio' : SIGNAVIO_NS } ) ( './/signavio:signavioMetaData[@metaKey="entry"]' )
        if not signavioMetaData :
            raise ValidationException ( 'No Signavio "Subprocess reference" specified.' , node = node , filename = filename )
        subprocess_reference = one ( signavioMetaData ) . get ( 'metaValue' )
        matches = [ ]
        for b in list ( self . bpmn . values ( ) ) :
            for p in xpath_eval ( b ) ( ".//bpmn:process" ) :
                if ( p . get ( 'name' , p . get ( 'id' , None ) ) == subprocess_reference ) :
                    matches . append ( p )
        if not matches :
            raise ValidationException ( "No matching process definition found for '%s'." % subprocess_reference , node = node , filename = filename )
        if len ( matches ) != 1 :
            raise ValidationException ( "More than one matching process definition " " found for '%s'." % subprocess_reference , node = node , filename = filename )
        node . set ( 'calledElement' , matches [ 0 ] . get ( 'id' ) )


signature:
def package_for_editor_signavio ( self , spec , filename ) :
    """Adds the SVG files to the archive for this BPMN file ."""

implementation:
signavio_file = filename [ : - len ( '.bpmn20.xml' ) ] + '.signavio.xml'
if os . path . exists ( signavio_file ) :
    self . write_file_to_package_zip ( "src/" + self . _get_zip_path ( signavio_file ) , signavio_file )
    f = open ( signavio_file , 'r' )
    try :
        signavio_tree = ET . parse ( f )
    finally :
        f . close ( )
    svg_node = one ( signavio_tree . findall ( './/svg-representation' ) )
    self . write_to_package_zip ( "%s.svg" % spec . name , svg_node . text )


signature:
def write_meta_data ( self ) :
    """Writes the metadata . ini file to the archive ."""

implementation:
config = configparser . ConfigParser ( )
config . add_section ( 'MetaData' )
config . set ( 'MetaData' , 'entry_point_process' , self . wf_spec . name )
if self . editor :
    config . set ( 'MetaData' , 'editor' , self . editor )
for k , v in self . meta_data :
    config . set ( 'MetaData' , k , v )
if not self . PARSER_CLASS == BpmnParser :
    config . set ( 'MetaData' , 'parser_class_module' , inspect . getmodule ( self . PARSER_CLASS ) . __name__ )
    config . set ( 'MetaData' , 'parser_class' , self . PARSER_CLASS . __name__ )
ini = StringIO ( )
config . write ( ini )
self . write_to_package_zip ( self . METADATA_FILE , ini . getvalue ( ) )


signature:
def add_main_options ( cls , parser ) :
    """Override in subclass if required ."""

implementation:
parser . add_option ( "-o" , "--output" , dest = "package_file" , help = "create the BPMN package in the specified file" )
parser . add_option ( "-p" , "--process" , dest = "entry_point_process" , help = "specify the entry point process" )
parser . add_option ( "-c" , "--config-file" , dest = "config_file" , help = "specify a config file to use" )
parser . add_option ( "-i" , "--initialise-config-file" , action = "store_true" , dest = "init_config_file" , default = False , help = "create a new config file from the specified options" )
group = OptionGroup ( parser , "BPMN Editor Options" , "These options are not required, but may be " " provided to activate special features of " "supported BPMN editors." )
group . add_option ( "--editor" , dest = "editor" , help = "editors with special support: signavio" )
parser . add_option_group ( group )


signature:
def add_additional_options ( cls , parser ) :
    """Override in subclass if required ."""

implementation:
group = OptionGroup ( parser , "Target Engine Options" , "These options are not required, but may be " "provided if a specific " "BPMN application engine is targeted." )
group . add_option ( "-e" , "--target-engine" , dest = "target_engine" , help = "target the specified BPMN application engine" )
group . add_option ( "-t" , "--target-version" , dest = "target_engine_version" , help = "target the specified version of the BPMN application engine" )
parser . add_option_group ( group )


signature:
def check_args ( cls , config , options , args , parser , package_file = None ) :
    """Override in subclass if required ."""

implementation:
if not args :
    parser . error ( "no input files specified" )
if not ( package_file or options . package_file ) :
    parser . error ( "no package file specified" )
if not options . entry_point_process :
    parser . error ( "no entry point process specified" )


signature:
def merge_options_and_config ( cls , config , options , args ) :
    """Override in subclass if required ."""

implementation:
if args :
    config . set ( CONFIG_SECTION_NAME , 'input_files' , ',' . join ( args ) )
elif config . has_option ( CONFIG_SECTION_NAME , 'input_files' ) :
    for i in config . get ( CONFIG_SECTION_NAME , 'input_files' ) . split ( ',' ) :
        if not os . path . isabs ( i ) :
            i = os . path . abspath ( os . path . join ( os . path . dirname ( options . config_file ) , i ) )
        args . append ( i )
cls . merge_option_and_config_str ( 'package_file' , config , options )
cls . merge_option_and_config_str ( 'entry_point_process' , config , options )
cls . merge_option_and_config_str ( 'target_engine' , config , options )
cls . merge_option_and_config_str ( 'target_engine_version' , config , options )
cls . merge_option_and_config_str ( 'editor' , config , options )


signature:
def merge_option_and_config_str ( cls , option_name , config , options ) :
    """Utility method to merge an option and config , with the option taking "
 precedence"""

implementation:
opt = getattr ( options , option_name , None )
if opt :
    config . set ( CONFIG_SECTION_NAME , option_name , opt )
elif config . has_option ( CONFIG_SECTION_NAME , option_name ) :
    setattr ( options , option_name , config . get ( CONFIG_SECTION_NAME , option_name ) )


signature:
def create_meta_data ( cls , options , args , parser ) :
    """Override in subclass if required ."""

implementation:
meta_data = [ ]
meta_data . append ( ( 'spiff_version' , cls . get_version ( ) ) )
if options . target_engine :
    meta_data . append ( ( 'target_engine' , options . target_engine ) )
if options . target_engine :
    meta_data . append ( ( 'target_engine_version' , options . target_engine_version ) )
return meta_data


signature:
def parse_node ( self , node ) :
    """Parses the specified child task node , and returns the task spec . This
 can be called by a TaskParser instance , that is owned by this
 ProcessParser ."""

implementation:
if node . get ( 'id' ) in self . parsed_nodes :
    return self . parsed_nodes [ node . get ( 'id' ) ]
( node_parser , spec_class ) = self . parser . _get_parser_class ( node . tag )
if not node_parser or not spec_class :
    raise ValidationException ( "There is no support implemented for this task type." , node = node , filename = self . filename )
np = node_parser ( self , spec_class , node )
task_spec = np . parse_node ( )
return task_spec


signature:
def get_spec ( self ) :
    """Parse this process ( if it has not already been parsed ) , and return the
 workflow spec ."""

implementation:
if self . is_parsed :
    return self . spec
if self . parsing_started :
    raise NotImplementedError ( 'Recursive call Activities are not supported.' )
self . _parse ( )
return self . get_spec ( )


signature:
def _on_trigger ( self , task_spec ) :
    """May be called after execute ( ) was already completed to create an
 additional outbound task ."""

implementation:
# Find a Task for this TaskSpec .
my_task = self . _find_my_task ( task_spec )
if my_task . _has_state ( Task . COMPLETED ) :
    state = Task . READY
else :
    state = Task . FUTURE
for output in self . outputs :
    new_task = my_task . _add_child ( output , state )
    new_task . triggered = True
    output . _predict ( new_task )


signature:
def connect ( self , task_spec ) :
    """Connect the * following * task to this one . In other words , the
 given task is added as an output task .
 task - - the task to connect to ."""

implementation:
self . thread_starter . outputs . append ( task_spec )
task_spec . _connect_notify ( self . thread_starter )


signature:
def _get_activated_tasks ( self , my_task , destination ) :
    """Returns the list of tasks that were activated in the previous
 call of execute ( ) . Only returns tasks that point towards the
 destination task , i . e . those which have destination as a
 descendant .
 my _ task - - the task of this TaskSpec
 destination - - the child task"""

implementation:
task = destination . _find_ancestor ( self . thread_starter )
return self . thread_starter . _get_activated_tasks ( task , destination )


signature:
def _on_trigger ( self , my_task ) :
    """May be called after execute ( ) was already completed to create an
 additional outbound task ."""

implementation:
for output in self . outputs :
    new_task = my_task . add_child ( output , Task . READY )
    new_task . triggered = True


signature:
def deserialize_assign ( self , workflow , start_node ) :
    """Reads the " pre - assign " or " post - assign " tag from the given node .
 start _ node - - the xml node ( xml . dom . minidom . Node )"""

implementation:
name = start_node . getAttribute ( 'name' )
attrib = start_node . getAttribute ( 'field' )
value = start_node . getAttribute ( 'value' )
kwargs = { }
if name == '' :
    _exc ( 'name attribute required' )
if attrib != '' and value != '' :
    _exc ( 'Both, field and right-value attributes found' )
elif attrib == '' and value == '' :
    _exc ( 'field or value attribute required' )
elif value != '' :
    kwargs [ 'right' ] = value
else :
    kwargs [ 'right_attribute' ] = attrib
return operators . Assign ( name , ** kwargs )


signature:
def deserialize_data ( self , workflow , start_node ) :
    """Reads a " data " or " define " tag from the given node .
 start _ node - - the xml node ( xml . dom . minidom . Node )"""

implementation:
name = start_node . getAttribute ( 'name' )
value = start_node . getAttribute ( 'value' )
return name , value


signature:
def deserialize_assign_list ( self , workflow , start_node ) :
    """Reads a list of assignments from the given node .
 workflow - - the workflow
 start _ node - - the xml structure ( xml . dom . minidom . Node )"""

implementation:
# Collect all information .
assignments = [ ]
for node in start_node . childNodes :
    if node . nodeType != minidom . Node . ELEMENT_NODE :
        continue
    if node . nodeName . lower ( ) == 'assign' :
        assignments . append ( self . deserialize_assign ( workflow , node ) )
    else :
        _exc ( 'Unknown node: %s' % node . nodeName )
return assignments


signature:
def deserialize_logical ( self , node ) :
    """Reads the logical tag from the given node , returns a Condition object .
 node - - the xml node ( xml . dom . minidom . Node )"""

implementation:
term1_attrib = node . getAttribute ( 'left-field' )
term1_value = node . getAttribute ( 'left-value' )
op = node . nodeName . lower ( )
term2_attrib = node . getAttribute ( 'right-field' )
term2_value = node . getAttribute ( 'right-value' )
if op not in _op_map :
    _exc ( 'Invalid operator' )
if term1_attrib != '' and term1_value != '' :
    _exc ( 'Both, left-field and left-value attributes found' )
elif term1_attrib == '' and term1_value == '' :
    _exc ( 'left-field or left-value attribute required' )
elif term1_value != '' :
    left = term1_value
else :
    left = operators . Attrib ( term1_attrib )
if term2_attrib != '' and term2_value != '' :
    _exc ( 'Both, right-field and right-value attributes found' )
elif term2_attrib == '' and term2_value == '' :
    _exc ( 'right-field or right-value attribute required' )
elif term2_value != '' :
    right = term2_value
else :
    right = operators . Attrib ( term2_attrib )
return _op_map [ op ] ( left , right )


signature:
def deserialize_condition ( self , workflow , start_node ) :
    """Reads the conditional statement from the given node .
 workflow - - the workflow with which the concurrence is associated
 start _ node - - the xml structure ( xml . dom . minidom . Node )"""

implementation:
# Collect all information .
condition = None
spec_name = None
for node in start_node . childNodes :
    if node . nodeType != minidom . Node . ELEMENT_NODE :
        continue
    if node . nodeName . lower ( ) == 'successor' :
        if spec_name is not None :
            _exc ( 'Duplicate task name %s' % spec_name )
        if node . firstChild is None :
            _exc ( 'Successor tag without a task name' )
        spec_name = node . firstChild . nodeValue
    elif node . nodeName . lower ( ) in _op_map :
        if condition is not None :
            _exc ( 'Multiple conditions are not yet supported' )
        condition = self . deserialize_logical ( node )
    else :
        _exc ( 'Unknown node: %s' % node . nodeName )
if condition is None :
    _exc ( 'Missing condition in conditional statement' )
if spec_name is None :
    _exc ( 'A %s has no task specified' % start_node . nodeName )
return condition , spec_name


signature:
def deserialize_task_spec ( self , workflow , start_node , read_specs ) :
    """Reads the task from the given node and returns a tuple
 ( start , end ) that contains the stream of objects that model
 the behavior .
 workflow - - the workflow with which the task is associated
 start _ node - - the xml structure ( xml . dom . minidom . Node )"""

implementation:
# Extract attributes from the node .
nodetype = start_node . nodeName . lower ( )
name = start_node . getAttribute ( 'name' ) . lower ( )
context = start_node . getAttribute ( 'context' ) . lower ( )
mutex = start_node . getAttribute ( 'mutex' ) . lower ( )
cancel = start_node . getAttribute ( 'cancel' ) . lower ( )
success = start_node . getAttribute ( 'success' ) . lower ( )
times = start_node . getAttribute ( 'times' ) . lower ( )
times_field = start_node . getAttribute ( 'times-field' ) . lower ( )
threshold = start_node . getAttribute ( 'threshold' ) . lower ( )
threshold_field = start_node . getAttribute ( 'threshold-field' ) . lower ( )
file = start_node . getAttribute ( 'file' ) . lower ( )
file_field = start_node . getAttribute ( 'file-field' ) . lower ( )
kwargs = { 'lock' : [ ] , 'data' : { } , 'defines' : { } , 'pre_assign' : [ ] , 'post_assign' : [ ] }
if nodetype not in _spec_map :
    _exc ( 'Invalid task type "%s"' % nodetype )
if nodetype == 'start-task' :
    name = 'start'
if name == '' :
    _exc ( 'Invalid task name "%s"' % name )
if name in read_specs :
    _exc ( 'Duplicate task name "%s"' % name )
if cancel != '' and cancel != '0' :
    kwargs [ 'cancel' ] = True
if success != '' and success != '0' :
    kwargs [ 'success' ] = True
if times != '' :
    kwargs [ 'times' ] = int ( times )
if times_field != '' :
    kwargs [ 'times' ] = operators . Attrib ( times_field )
if threshold != '' :
    kwargs [ 'threshold' ] = int ( threshold )
if threshold_field != '' :
    kwargs [ 'threshold' ] = operators . Attrib ( threshold_field )
if file != '' :
    kwargs [ 'file' ] = file
if file_field != '' :
    kwargs [ 'file' ] = operators . Attrib ( file_field )
if nodetype == 'choose' :
    kwargs [ 'choice' ] = [ ]
if nodetype == 'trigger' :
    context = [ context ]
if mutex != '' :
    context = mutex
# Walk through the children of the node .
successors = [ ]
for node in start_node . childNodes :
    if node . nodeType != minidom . Node . ELEMENT_NODE :
        continue
    if node . nodeName == 'description' :
        kwargs [ 'description' ] = node . firstChild . nodeValue
    elif node . nodeName == 'successor' or node . nodeName == 'default-successor' :
        if node . firstChild is None :
            _exc ( 'Empty %s tag' % node . nodeName )
        successors . append ( ( None , node . firstChild . nodeValue ) )
    elif node . nodeName == 'conditional-successor' :
        successors . append ( self . deserialize_condition ( workflow , node ) )
    elif node . nodeName == 'define' :
        key , value = self . deserialize_data ( workflow , node )
        kwargs [ 'defines' ] [ key ] = value
    # " property " tag exists for backward compatibility .
    elif node . nodeName == 'data' or node . nodeName == 'property' :
        key , value = self . deserialize_data ( workflow , node )
        kwargs [ 'data' ] [ key ] = value
    elif node . nodeName == 'pre-assign' :
        kwargs [ 'pre_assign' ] . append ( self . deserialize_assign ( workflow , node ) )
    elif node . nodeName == 'post-assign' :
        kwargs [ 'post_assign' ] . append ( self . deserialize_assign ( workflow , node ) )
    elif node . nodeName == 'in' :
        kwargs [ 'in_assign' ] = self . deserialize_assign_list ( workflow , node )
    elif node . nodeName == 'out' :
        kwargs [ 'out_assign' ] = self . deserialize_assign_list ( workflow , node )
    elif node . nodeName == 'cancel' :
        if node . firstChild is None :
            _exc ( 'Empty %s tag' % node . nodeName )
        if context == '' :
            context = [ ]
        elif not isinstance ( context , list ) :
            context = [ context ]
        context . append ( node . firstChild . nodeValue )
    elif node . nodeName == 'lock' :
        if node . firstChild is None :
            _exc ( 'Empty %s tag' % node . nodeName )
        kwargs [ 'lock' ] . append ( node . firstChild . nodeValue )
    elif node . nodeName == 'pick' :
        if node . firstChild is None :
            _exc ( 'Empty %s tag' % node . nodeName )
        kwargs [ 'choice' ] . append ( node . firstChild . nodeValue )
    else :
        _exc ( 'Unknown node: %s' % node . nodeName )
# Create a new instance of the task spec .
module = _spec_map [ nodetype ]
if nodetype == 'start-task' :
    spec = module ( workflow , ** kwargs )
elif nodetype == 'multi-instance' or nodetype == 'thread-split' :
    if times == '' and times_field == '' :
        _exc ( 'Missing "times" or "times-field" in "%s"' % name )
    elif times != '' and times_field != '' :
        _exc ( 'Both, "times" and "times-field" in "%s"' % name )
    spec = module ( workflow , name , ** kwargs )
elif context == '' :
    spec = module ( workflow , name , ** kwargs )
else :
    spec = module ( workflow , name , context , ** kwargs )
read_specs [ name ] = spec , successors


signature:
def deserialize_workflow_spec ( self , s_state , filename = None ) :
    """Reads the workflow from the given XML structure and returns a
 WorkflowSpec instance ."""

implementation:
dom = minidom . parseString ( s_state )
node = dom . getElementsByTagName ( 'process-definition' ) [ 0 ]
name = node . getAttribute ( 'name' )
if name == '' :
    _exc ( '%s without a name attribute' % node . nodeName )
# Read all task specs and create a list of successors .
workflow_spec = specs . WorkflowSpec ( name , filename )
del workflow_spec . task_specs [ 'Start' ]
end = specs . Simple ( workflow_spec , 'End' ) , [ ]
read_specs = dict ( end = end )
for child_node in node . childNodes :
    if child_node . nodeType != minidom . Node . ELEMENT_NODE :
        continue
    if child_node . nodeName == 'name' :
        workflow_spec . name = child_node . firstChild . nodeValue
    elif child_node . nodeName == 'description' :
        workflow_spec . description = child_node . firstChild . nodeValue
    elif child_node . nodeName . lower ( ) in _spec_map :
        self . deserialize_task_spec ( workflow_spec , child_node , read_specs )
    else :
        _exc ( 'Unknown node: %s' % child_node . nodeName )
# Remove the default start - task from the workflow .
workflow_spec . start = read_specs [ 'start' ] [ 0 ]
# Connect all task specs .
for name in read_specs :
    spec , successors = read_specs [ name ]
    for condition , successor_name in successors :
        if successor_name not in read_specs :
            _exc ( 'Unknown successor: "%s"' % successor_name )
        successor , foo = read_specs [ successor_name ]
        if condition is None :
            spec . connect ( successor )
        else :
            spec . connect_if ( condition , successor )
return workflow_spec


signature:
def has_fired ( self , my_task ) :
    """The Timer is considered to have fired if the evaluated dateTime
 expression is before datetime . datetime . now ( )"""

implementation:
dt = my_task . workflow . script_engine . evaluate ( my_task , self . dateTime )
if dt is None :
    return False
if dt . tzinfo :
    tz = dt . tzinfo
    now = tz . fromutc ( datetime . datetime . utcnow ( ) . replace ( tzinfo = tz ) )
else :
    now = datetime . datetime . now ( )
return now > dt


signature:
def _add_notify ( self , task_spec ) :
    """Called by a task spec when it was added into the workflow ."""

implementation:
if task_spec . name in self . task_specs :
    raise KeyError ( 'Duplicate task spec name: ' + task_spec . name )
self . task_specs [ task_spec . name ] = task_spec
task_spec . id = len ( self . task_specs )


signature:
def validate ( self ) :
    """Checks integrity of workflow and reports any problems with it .
 Detects :
 - loops ( tasks that wait on each other in a loop )
 : returns : empty list if valid , a list of errors if not"""

implementation:
results = [ ]
from . . specs import Join
def recursive_find_loop ( task , history ) :
    current = history [ : ]
    current . append ( task )
    if isinstance ( task , Join ) :
        if task in history :
            msg = "Found loop with '%s': %s then '%s' again" % ( task . name , '->' . join ( [ p . name for p in history ] ) , task . name )
            raise Exception ( msg )
        for predecessor in task . inputs :
            recursive_find_loop ( predecessor , current )
    for parent in task . inputs :
        recursive_find_loop ( parent , current )
for task_id , task in list ( self . task_specs . items ( ) ) : # Check for cyclic waits
    try :
        recursive_find_loop ( task , [ ] )
    except Exception as exc :
        results . append ( exc . __str__ ( ) )
    # Check for disconnected tasks
    if not task . inputs and task . name not in [ 'Start' , 'Root' ] :
        if task . outputs :
            results . append ( "Task '%s' is disconnected (no inputs)" % task . name )
        else :
            LOG . debug ( "Task '%s' is not being used" % task . name )
return results


signature:
def accept_message ( self , message ) :
    """Indicate to the workflow that a message has been received . The message
 will be processed by any waiting Intermediate or Boundary Message
 Events , that are waiting for the message ."""

implementation:
assert not self . read_only
self . refresh_waiting_tasks ( )
self . do_engine_steps ( )
for my_task in Task . Iterator ( self . task_tree , Task . WAITING ) :
    my_task . task_spec . accept_message ( my_task , message )


signature:
def do_engine_steps ( self ) :
    """Execute any READY tasks that are engine specific ( for example , gateways
 or script tasks ) . This is done in a loop , so it will keep completing
 those tasks until there are only READY User tasks , or WAITING tasks
 left ."""

implementation:
assert not self . read_only
engine_steps = list ( [ t for t in self . get_tasks ( Task . READY ) if self . _is_engine_task ( t . task_spec ) ] )
while engine_steps :
    for task in engine_steps :
        task . complete ( )
    engine_steps = list ( [ t for t in self . get_tasks ( Task . READY ) if self . _is_engine_task ( t . task_spec ) ] )


signature:
def refresh_waiting_tasks ( self ) :
    """Refresh the state of all WAITING tasks . This will , for example , update
 Catching Timer Events whose waiting time has passed ."""

implementation:
assert not self . read_only
for my_task in self . get_tasks ( Task . WAITING ) :
    my_task . task_spec . _update ( my_task )


signature:
def get_ready_user_tasks ( self ) :
    """Returns a list of User Tasks that are READY for user action"""

implementation: return [ t for t in self . get_tasks ( Task . READY ) if not self . _is_engine_task ( t . task_spec ) ]


signature:
def _on_trigger ( self , my_task ) :
    """Enqueue a trigger , such that this tasks triggers multiple times later
 when _ on _ complete ( ) is called ."""

implementation:
self . queued += 1
# All tasks that have already completed need to be put back to
# READY .
for thetask in my_task . workflow . task_tree :
    if thetask . thread_id != my_task . thread_id :
        continue
    if ( thetask . task_spec == self and thetask . _has_state ( Task . COMPLETED ) ) :
        thetask . _set_state ( Task . FUTURE , True )
        thetask . _ready ( )


signature:
def _on_complete_hook ( self , my_task ) :
    """A hook into _ on _ complete ( ) that does the task specific work .
 : type my _ task : Task
 : param my _ task : A task in which this method is executed .
 : rtype : bool
 : returns : True on success , False otherwise ."""

implementation:
times = int ( valueof ( my_task , self . times , 1 ) ) + self . queued
for i in range ( times ) :
    for task_name in self . context :
        task = my_task . workflow . get_task_spec_from_name ( task_name )
        task . _on_trigger ( my_task )
self . queued = 0
TaskSpec . _on_complete_hook ( self , my_task )


signature:
def deserialize ( cls , serializer , wf_spec , s_state , ** kwargs ) :
    """Deserializes the trigger using the provided serializer ."""

implementation: return serializer . deserialize_trigger ( wf_spec , s_state , ** kwargs )


signature:
def evaluate ( self , task , expression ) :
    """Evaluate the given expression , within the context of the given task and
 return the result ."""

implementation:
if isinstance ( expression , Operator ) :
    return expression . _matches ( task )
else :
    return self . _eval ( task , expression , ** task . data )


signature:
def execute ( self , task , script , ** kwargs ) :
    """Execute the script , within the context of the specified task"""

implementation:
locals ( ) . update ( kwargs )
exec ( script )


signature:
def merge_dictionary ( dst , src ) :
    """Recursive merge two dicts ( vs . update which overwrites the hashes at the
 root level )
 Note : This updates dst .
 Copied from checkmate . utils"""

implementation:
stack = [ ( dst , src ) ]
while stack :
    current_dst , current_src = stack . pop ( )
    for key in current_src :
        source = current_src [ key ]
        if key not in current_dst :
            current_dst [ key ] = source
        else :
            dest = current_dst [ key ]
            if isinstance ( source , dict ) and isinstance ( dest , dict ) :
                stack . append ( ( dest , source ) )
            elif isinstance ( source , list ) and isinstance ( dest , list ) : # Make them the same size
                r = dest [ : ]
                s = source [ : ]
                if len ( dest ) > len ( source ) :
                    s . append ( [ None for i in range ( len ( dest ) - len ( source ) ) ] )
                elif len ( dest ) < len ( source ) :
                    r . append ( [ None for i in range ( len ( source ) - len ( dest ) ) ] )
                # Merge lists
                for index , value in enumerate ( r ) :
                    if ( not value ) and s [ index ] :
                        r [ index ] = s [ index ]
                    elif isinstance ( value , dict ) and isinstance ( s [ index ] , dict ) :
                        stack . append ( ( dest [ index ] , source [ index ] ) )
                    else :
                        dest [ index ] = s [ index ]
                current_dst [ key ] = r
            else :
                current_dst [ key ] = source
return dst


signature:
def _start ( self , my_task , force = False ) :
    """Checks whether the preconditions for going to READY state are met .
 Returns True if the threshold was reached , False otherwise .
 Also returns the list of tasks that yet need to be completed ."""

implementation:
# If the threshold was already reached , there is nothing else to do .
if my_task . _has_state ( Task . COMPLETED ) :
    return True , None
if my_task . _has_state ( Task . READY ) :
    return True , None
# Check whether we may fire .
if self . split_task is None :
    return self . _check_threshold_unstructured ( my_task , force )
return self . _check_threshold_structured ( my_task , force )


signature:
def _on_trigger ( self , my_task ) :
    """May be called to fire the Join before the incoming branches are
 completed ."""

implementation:
for task in my_task . workflow . task_tree . _find_any ( self ) :
    if task . thread_id != my_task . thread_id :
        continue
    self . _do_join ( task )


signature:
def connect ( self , task_spec ) :
    """Connects the task spec that is executed if no other condition
 matches .
 : type task _ spec : TaskSpec
 : param task _ spec : The following task spec ."""

implementation:
assert self . default_task_spec is None
self . outputs . append ( task_spec )
self . default_task_spec = task_spec . name
task_spec . _connect_notify ( self )


signature:
def container_id ( self , name ) :
    '''Try to find the container ID with the specified name'''

implementation:
container = self . _containers . get ( name , None )
if not container is None :
    return container . get ( 'id' , None )
return None


signature:
def initialize ( self , containers ) :
    '''Initialize a new state file with the given contents .
 This function fails in case the state file already exists .'''

implementation:
self . _containers = deepcopy ( containers )
self . __write ( containers , initialize = True )


signature:
def update ( self , containers ) :
    '''Update the current state file with the specified contents'''

implementation:
self . _containers = deepcopy ( containers )
self . __write ( containers , initialize = False )


signature:
def load ( self ) :
    '''Try to load a blockade state file in the current directory'''

implementation:
try :
    with open ( self . _state_file ) as f :
        state = yaml . safe_load ( f )
        self . _containers = state [ 'containers' ]
except ( IOError , OSError ) as err :
    if err . errno == errno . ENOENT :
        raise NotInitializedError ( "No blockade exists in this context" )
    raise InconsistentStateError ( "Failed to load Blockade state: " + str ( err ) )
except Exception as err :
    raise InconsistentStateError ( "Failed to load Blockade state: " + str ( err ) )


signature:
def _get_blockade_id_from_cwd ( self , cwd = None ) :
    '''Generate a new blockade ID based on the CWD'''

implementation:
if not cwd :
    cwd = os . getcwd ( )
# this follows a similar pattern as docker - compose uses
parent_dir = os . path . abspath ( cwd )
basename = os . path . basename ( parent_dir ) . lower ( )
blockade_id = re . sub ( r"[^a-z0-9]" , "" , basename )
if not blockade_id : # if we can ' t get a valid name from CWD , use " default "
    blockade_id = "default"
return blockade_id


signature:
def _assure_dir ( self ) :
    '''Make sure the state directory exists'''

implementation:
try :
    os . makedirs ( self . _state_dir )
except OSError as err :
    if err . errno != errno . EEXIST :
        raise


signature:
def _state_delete ( self ) :
    '''Try to delete the state . yml file and the folder . blockade'''

implementation:
try :
    os . remove ( self . _state_file )
except OSError as err :
    if err . errno not in ( errno . EPERM , errno . ENOENT ) :
        raise
try :
    os . rmdir ( self . _state_dir )
except OSError as err :
    if err . errno not in ( errno . ENOTEMPTY , errno . ENOENT ) :
        raise


signature:
def __base_state ( self , containers ) :
    '''Convert blockade ID and container information into
 a state dictionary object .'''

implementation: return dict ( blockade_id = self . _blockade_id , containers = containers , version = self . _state_version )


signature:
def __write ( self , containers , initialize = True ) :
    '''Write the given state information into a file'''

implementation:
path = self . _state_file
self . _assure_dir ( )
try :
    flags = os . O_WRONLY | os . O_CREAT
    if initialize :
        flags |= os . O_EXCL
    with os . fdopen ( os . open ( path , flags ) , "w" ) as f :
        yaml . safe_dump ( self . __base_state ( containers ) , f )
except OSError as err :
    if err . errno == errno . EEXIST :
        raise AlreadyInitializedError ( "Path %s exists. " "You may need to destroy a previous blockade." % path )
    raise
except Exception : # clean up our created file
    self . _state_delete ( )
    raise


signature:
def expand_partitions ( containers , partitions ) :
    '''Validate the partitions of containers . If there are any containers
 not in any partition , place them in an new partition .'''

implementation:
# filter out holy containers that don ' t belong
# to any partition at all
all_names = frozenset ( c . name for c in containers if not c . holy )
holy_names = frozenset ( c . name for c in containers if c . holy )
neutral_names = frozenset ( c . name for c in containers if c . neutral )
partitions = [ frozenset ( p ) for p in partitions ]
unknown = set ( )
holy = set ( )
union = set ( )
for partition in partitions :
    unknown . update ( partition - all_names - holy_names )
    holy . update ( partition - all_names )
    union . update ( partition )
if unknown :
    raise BlockadeError ( 'Partitions contain unknown containers: %s' % list ( unknown ) )
if holy :
    raise BlockadeError ( 'Partitions contain holy containers: %s' % list ( holy ) )
# put any leftover containers in an implicit partition
leftover = all_names . difference ( union )
if leftover :
    partitions . append ( leftover )
# we create an ' implicit ' partition for the neutral containers
# in case they are not part of the leftover anyways
if not neutral_names . issubset ( leftover ) :
    partitions . append ( neutral_names )
return partitions


signature:
def get_source_chains ( self , blockade_id ) :
    """Get a map of blockade chains IDs - > list of IPs targeted at them
 For figuring out which container is in which partition"""

implementation:
result = { }
if not blockade_id :
    raise ValueError ( "invalid blockade_id" )
lines = self . get_chain_rules ( "FORWARD" )
for line in lines :
    parts = line . split ( )
    if len ( parts ) < 4 :
        continue
    try :
        partition_index = parse_partition_index ( blockade_id , parts [ 0 ] )
    except ValueError :
        continue
    # not a rule targetting a blockade chain
    source = parts [ 3 ]
    if source :
        result [ source ] = partition_index
return result


signature:
def insert_rule ( self , chain , src = None , dest = None , target = None ) :
    """Insert a new rule in the chain"""

implementation:
if not chain :
    raise ValueError ( "Invalid chain" )
if not target :
    raise ValueError ( "Invalid target" )
if not ( src or dest ) :
    raise ValueError ( "Need src, dest, or both" )
args = [ "-I" , chain ]
if src :
    args += [ "-s" , src ]
if dest :
    args += [ "-d" , dest ]
args += [ "-j" , target ]
self . call ( * args )


signature:
def _sm_start ( self , * args , ** kwargs ) :
    """Start the timer waiting for pain"""

implementation:
millisec = random . randint ( self . _start_min_delay , self . _start_max_delay )
self . _timer = threading . Timer ( millisec / 1000.0 , self . event_timeout )
self . _timer . start ( )


signature:
def _sm_to_pain ( self , * args , ** kwargs ) :
    """Start the blockade event"""

implementation:
_logger . info ( "Starting chaos for blockade %s" % self . _blockade_name )
self . _do_blockade_event ( )
# start the timer to end the pain
millisec = random . randint ( self . _run_min_time , self . _run_max_time )
self . _timer = threading . Timer ( millisec / 1000.0 , self . event_timeout )
self . _timer . start ( )


signature:
def _sm_stop_from_no_pain ( self , * args , ** kwargs ) :
    """Stop chaos when there is no current blockade operation"""

implementation:
# Just stop the timer . It is possible that it was too late and the
# timer is about to run
_logger . info ( "Stopping chaos for blockade %s" % self . _blockade_name )
self . _timer . cancel ( )


signature:
def _sm_relieve_pain ( self , * args , ** kwargs ) :
    """End the blockade event and return to a steady state"""

implementation:
_logger . info ( "Ending the degradation for blockade %s" % self . _blockade_name )
self . _do_reset_all ( )
# set a timer for the next pain event
millisec = random . randint ( self . _start_min_delay , self . _start_max_delay )
self . _timer = threading . Timer ( millisec / 1000.0 , self . event_timeout )
self . _timer . start ( )


signature:
def _sm_stop_from_pain ( self , * args , ** kwargs ) :
    """Stop chaos while there is a blockade event in progress"""

implementation:
_logger . info ( "Stopping chaos for blockade %s" % self . _blockade_name )
self . _do_reset_all ( )


signature:
def _sm_cleanup ( self , * args , ** kwargs ) :
    """Delete all state associated with the chaos session"""

implementation:
if self . _done_notification_func is not None :
    self . _done_notification_func ( )
self . _timer . cancel ( )


signature:
def dependency_sorted ( containers ) :
    """Sort a dictionary or list of containers into dependency order
 Returns a sequence"""

implementation:
if not isinstance ( containers , collections . Mapping ) :
    containers = dict ( ( c . name , c ) for c in containers )
container_links = dict ( ( name , set ( c . links . keys ( ) ) ) for name , c in containers . items ( ) )
sorted_names = _resolve ( container_links )
return [ containers [ name ] for name in sorted_names ]


signature:
def from_dict ( name , values ) :
    '''Convert a dictionary of configuration values
 into a sequence of BlockadeContainerConfig instances'''

implementation:
# determine the number of instances of this container
count = 1
count_value = values . get ( 'count' , 1 )
if isinstance ( count_value , int ) :
    count = max ( count_value , 1 )
def with_index ( name , idx ) :
    if name and idx :
        return '%s_%d' % ( name , idx )
    return name
def get_instance ( n , idx = None ) :
    return BlockadeContainerConfig ( with_index ( n , idx ) , values [ 'image' ] , command = values . get ( 'command' ) , links = values . get ( 'links' ) , volumes = values . get ( 'volumes' ) , publish_ports = values . get ( 'ports' ) , expose_ports = values . get ( 'expose' ) , environment = values . get ( 'environment' ) , hostname = values . get ( 'hostname' ) , dns = values . get ( 'dns' ) , start_delay = values . get ( 'start_delay' , 0 ) , neutral = values . get ( 'neutral' , False ) , holy = values . get ( 'holy' , False ) , container_name = with_index ( values . get ( 'container_name' ) , idx ) , cap_add = values . get ( 'cap_add' ) )
if count == 1 :
    yield get_instance ( name )
else :
    for idx in range ( 1 , count + 1 ) : # TODO : configurable name / index format
        yield get_instance ( name , idx )


signature:
def from_dict ( values ) :
    '''Instantiate a BlockadeConfig instance based on
 a given dictionary of configuration values'''

implementation:
try :
    containers = values [ 'containers' ]
    parsed_containers = { }
    for name , container_dict in containers . items ( ) :
        try : # one config entry might result in many container
        # instances ( indicated by the ' count ' config value )
            for cnt in BlockadeContainerConfig . from_dict ( name , container_dict ) : # check for duplicate ' container _ name ' definitions
                if cnt . container_name :
                    cname = cnt . container_name
                    existing = [ c for c in parsed_containers . values ( ) if c . container_name == cname ]
                    if existing :
                        raise BlockadeConfigError ( "Duplicate 'container_name' definition: %s" % ( cname ) )
                parsed_containers [ cnt . name ] = cnt
        except Exception as err :
            raise BlockadeConfigError ( "Container '%s' config problem: %s" % ( name , err ) )
    network = values . get ( 'network' )
    if network :
        defaults = _DEFAULT_NETWORK_CONFIG . copy ( )
        defaults . update ( network )
        network = defaults
    else :
        network = _DEFAULT_NETWORK_CONFIG . copy ( )
    return BlockadeConfig ( parsed_containers , network = network )
except KeyError as err :
    raise BlockadeConfigError ( "Config missing value: " + str ( err ) )
except Exception as err : # TODO log this to some debug stream ?
    raise BlockadeConfigError ( "Failed to load config: " + str ( err ) )


signature:
def cmd_up ( opts ) :
    """Start the containers and link them together"""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
containers = b . create ( verbose = opts . verbose , force = opts . force )
print_containers ( containers , opts . json )


signature:
def cmd_destroy ( opts ) :
    """Destroy all containers and restore networks"""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
b . destroy ( )


signature:
def cmd_status ( opts ) :
    """Print status of containers and networks"""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
containers = b . status ( )
print_containers ( containers , opts . json )


signature:
def cmd_kill ( opts ) :
    """Kill some or all containers"""

implementation:
kill_signal = opts . signal if hasattr ( opts , 'signal' ) else "SIGKILL"
__with_containers ( opts , Blockade . kill , signal = kill_signal )


signature:
def cmd_partition ( opts ) :
    """Partition the network between containers
 Replaces any existing partitions outright . Any containers NOT specified
 in arguments will be globbed into a single implicit partition . For
 example if you have three containers : c1 , c2 , and c3 and you run :
 blockade partition c1
 The result will be a partition with just c1 and another partition with
 c2 and c3.
 Alternatively , - - random may be specified , and zero or more random
 partitions will be generated by blockade ."""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
if opts . random :
    if opts . partitions :
        raise BlockadeError ( "Either specify individual partitions " "or --random, but not both" )
    b . random_partition ( )
else :
    partitions = [ ]
    for partition in opts . partitions :
        names = [ ]
        for name in partition . split ( "," ) :
            name = name . strip ( )
            if name :
                names . append ( name )
        partitions . append ( names )
    if not partitions :
        raise BlockadeError ( "Either specify individual partitions " "or random" )
    b . partition ( partitions )


signature:
def cmd_join ( opts ) :
    """Restore full networking between containers"""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
b . join ( )


signature:
def cmd_logs ( opts ) :
    """Fetch the logs of a container"""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
puts ( b . logs ( opts . container ) . decode ( encoding = 'UTF-8' ) )


signature:
def cmd_daemon ( opts ) :
    """Start the Blockade REST API"""

implementation:
if opts . data_dir is None :
    raise BlockadeError ( "You must supply a data directory for the daemon" )
rest . start ( data_dir = opts . data_dir , port = opts . port , debug = opts . debug , host_exec = get_host_exec ( ) )


signature:
def cmd_add ( opts ) :
    """Add one or more existing Docker containers to a Blockade group"""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
b . add_container ( opts . containers )


signature:
def cmd_events ( opts ) :
    """Get the event log for a given blockade"""

implementation:
config = load_config ( opts . config )
b = get_blockade ( config , opts )
if opts . json :
    outf = None
    _write = puts
    if opts . output is not None :
        outf = open ( opts . output , "w" )
        _write = outf . write
    try :
        delim = ""
        logs = b . get_audit ( ) . read_logs ( as_json = False )
        _write ( '{"events": [' )
        _write ( os . linesep )
        for l in logs :
            _write ( delim + l )
            delim = "," + os . linesep
        _write ( os . linesep )
        _write ( ']}' )
    finally :
        if opts . output is not None :
            outf . close ( )
else :
    puts ( colored . blue ( columns ( [ "EVENT" , 10 ] , [ "TARGET" , 16 ] , [ "STATUS" , 8 ] , [ "TIME" , 16 ] , [ "MESSAGE" , 25 ] ) ) )
    logs = b . get_audit ( ) . read_logs ( as_json = True )
    for l in logs :
        puts ( columns ( [ l [ 'event' ] , 10 ] , [ str ( [ str ( t ) for t in l [ 'targets' ] ] ) , 16 ] , [ l [ 'status' ] , 8 ] , [ str ( l [ 'timestamp' ] ) , 16 ] , [ l [ 'message' ] , 25 ] ) )


signature:
def set_cors_headers ( resp , options ) :
    """Performs the actual evaluation of Flas - CORS options and actually
 modifies the response object .
 This function is used both in the decorator and the after _ request
 callback"""

implementation:
# If CORS has already been evaluated via the decorator , skip
if hasattr ( resp , FLASK_CORS_EVALUATED ) :
    LOG . debug ( 'CORS have been already evaluated, skipping' )
    return resp
# Some libraries , like OAuthlib , set resp . headers to non Multidict
# objects ( Werkzeug Headers work as well ) . This is a problem because
# headers allow repeated values .
if ( not isinstance ( resp . headers , Headers ) and not isinstance ( resp . headers , MultiDict ) ) :
    resp . headers = MultiDict ( resp . headers )
headers_to_set = get_cors_headers ( options , request . headers , request . method )
LOG . debug ( 'Settings CORS headers: %s' , str ( headers_to_set ) )
for k , v in headers_to_set . items ( ) :
    resp . headers . add ( k , v )
return resp


signature:
def try_match ( request_origin , maybe_regex ) :
    """Safely attempts to match a pattern or string to a request origin ."""

implementation:
if isinstance ( maybe_regex , RegexObject ) :
    return re . match ( maybe_regex , request_origin )
elif probably_regex ( maybe_regex ) :
    return re . match ( maybe_regex , request_origin , flags = re . IGNORECASE )
else :
    try :
        return request_origin . lower ( ) == maybe_regex . lower ( )
    except AttributeError :
        return request_origin == maybe_regex


signature:
def get_cors_options ( appInstance , * dicts ) :
    """Compute CORS options for an application by combining the DEFAULT _ OPTIONS ,
 the app ' s configuration - specified options and any dictionaries passed . The
 last specified option wins ."""

implementation:
options = DEFAULT_OPTIONS . copy ( )
options . update ( get_app_kwarg_dict ( appInstance ) )
if dicts :
    for d in dicts :
        options . update ( d )
return serialize_options ( options )


signature:
def get_app_kwarg_dict ( appInstance = None ) :
    """Returns the dictionary of CORS specific app configurations ."""

implementation:
app = ( appInstance or current_app )
# In order to support blueprints which do not have a config attribute
app_config = getattr ( app , 'config' , { } )
return { k . lower ( ) . replace ( 'cors_' , '' ) : app_config . get ( k ) for k in CONFIG_OPTIONS if app_config . get ( k ) is not None }


signature:
def ensure_iterable ( inst ) :
    """Wraps scalars or string types as a list , or returns the iterable instance ."""

implementation:
if isinstance ( inst , string_types ) :
    return [ inst ]
elif not isinstance ( inst , collections . Iterable ) :
    return [ inst ]
else :
    return inst


signature:
def serialize_options ( opts ) :
    """A helper method to serialize and processes the options dictionary ."""

implementation:
options = ( opts or { } ) . copy ( )
for key in opts . keys ( ) :
    if key not in DEFAULT_OPTIONS :
        LOG . warning ( "Unknown option passed to Flask-CORS: %s" , key )
# Ensure origins is a list of allowed origins with at least one entry .
options [ 'origins' ] = sanitize_regex_param ( options . get ( 'origins' ) )
options [ 'allow_headers' ] = sanitize_regex_param ( options . get ( 'allow_headers' ) )
# This is expressly forbidden by the spec . Raise a value error so people
# don ' t get burned in production .
if r'.*' in options [ 'origins' ] and options [ 'supports_credentials' ] and options [ 'send_wildcard' ] :
    raise ValueError ( "Cannot use supports_credentials in conjunction with" "an origin string of '*'. See: " "http://www.w3.org/TR/cors/#resource-requests" )
serialize_option ( options , 'expose_headers' )
serialize_option ( options , 'methods' , upper = True )
if isinstance ( options . get ( 'max_age' ) , timedelta ) :
    options [ 'max_age' ] = str ( int ( options [ 'max_age' ] . total_seconds ( ) ) )
return options


signature:
def cross_origin ( * args , ** kwargs ) :
    """This function is the decorator which is used to wrap a Flask route with .
 In the simplest case , simply use the default parameters to allow all
 origins in what is the most permissive configuration . If this method
 modifies state or performs authentication which may be brute - forced , you
 should add some degree of protection , such as Cross Site Forgery
 Request protection .
 : param origins :
 The origin , or list of origins to allow requests from .
 The origin ( s ) may be regular expressions , case - sensitive strings ,
 or else an asterisk
 Default : ' * '
 : type origins : list , string or regex
 : param methods :
 The method or list of methods which the allowed origins are allowed to
 access for non - simple requests .
 Default : [ GET , HEAD , POST , OPTIONS , PUT , PATCH , DELETE ]
 : type methods : list or string
 : param expose _ headers :
 The header or list which are safe to expose to the API of a CORS API
 specification .
 Default : None
 : type expose _ headers : list or string
 : param allow _ headers :
 The header or list of header field names which can be used when this
 resource is accessed by allowed origins . The header ( s ) may be regular
 expressions , case - sensitive strings , or else an asterisk .
 Default : ' * ' , allow all headers
 : type allow _ headers : list , string or regex
 : param supports _ credentials :
 Allows users to make authenticated requests . If true , injects the
 ` Access - Control - Allow - Credentials ` header in responses . This allows
 cookies and credentials to be submitted across domains .
 : note : This option cannot be used in conjuction with a ' * ' origin
 Default : False
 : type supports _ credentials : bool
 : param max _ age :
 The maximum time for which this CORS request maybe cached . This value
 is set as the ` Access - Control - Max - Age ` header .
 Default : None
 : type max _ age : timedelta , integer , string or None
 : param send _ wildcard : If True , and the origins parameter is ` * ` , a wildcard
 ` Access - Control - Allow - Origin ` header is sent , rather than the
 request ' s ` Origin ` header .
 Default : False
 : type send _ wildcard : bool
 : param vary _ header :
 If True , the header Vary : Origin will be returned as per the W3
 implementation guidelines .
 Setting this header when the ` Access - Control - Allow - Origin ` is
 dynamically generated ( e . g . when there is more than one allowed
 origin , and an Origin than ' * ' is returned ) informs CDNs and other
 caches that the CORS headers are dynamic , and cannot be cached .
 If False , the Vary header will never be injected or altered .
 Default : True
 : type vary _ header : bool
 : param automatic _ options :
 Only applies to the ` cross _ origin ` decorator . If True , Flask - CORS will
 override Flask ' s default OPTIONS handling to return CORS headers for
 OPTIONS requests .
 Default : True
 : type automatic _ options : bool"""

implementation:
_options = kwargs
def decorator ( f ) :
    LOG . debug ( "Enabling %s for cross_origin using options:%s" , f , _options )
    # If True , intercept OPTIONS requests by modifying the view function ,
    # replicating Flask ' s default behavior , and wrapping the response with
    # CORS headers .
    # If f . provide _ automatic _ options is unset or True , Flask ' s route
    # decorator ( which is actually wraps the function object we return )
    # intercepts OPTIONS handling , and requests will not have CORS headers
    if _options . get ( 'automatic_options' , True ) :
        f . required_methods = getattr ( f , 'required_methods' , set ( ) )
        f . required_methods . add ( 'OPTIONS' )
        f . provide_automatic_options = False
    def wrapped_function ( * args , ** kwargs ) : # Handle setting of Flask - Cors parameters
        options = get_cors_options ( current_app , _options )
        if options . get ( 'automatic_options' ) and request . method == 'OPTIONS' :
            resp = current_app . make_default_options_response ( )
        else :
            resp = make_response ( f ( * args , ** kwargs ) )
        set_cors_headers ( resp , options )
        setattr ( resp , FLASK_CORS_EVALUATED , True )
        return resp
    return update_wrapper ( wrapped_function , f )
return decorator


signature:
def calendar ( type = 'holiday' , direction = 'next' , last = 1 , startDate = None , token = '' , version = '' ) :
    '''This call allows you to fetch a number of trade dates or holidays from a given date . For example , if you want the next trading day , you would call / ref - data / us / dates / trade / next / 1.
 https : / / iexcloud . io / docs / api / # u - s - exchanges
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 type ( string ) ; " holiday " or " trade "
 direction ( string ) ; " next " or " last "
 last ( int ) ; number to move in direction
 startDate ( date ) ; start date for next or last , YYYYMMDD
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
if startDate :
    startDate = _strOrDate ( startDate )
    return _getJson ( 'ref-data/us/dates/{type}/{direction}/{last}/{date}' . format ( type = type , direction = direction , last = last , date = startDate ) , token , version )
return _getJson ( 'ref-data/us/dates/' + type + '/' + direction + '/' + str ( last ) , token , version )


signature:
def calendarDF ( type = 'holiday' , direction = 'next' , last = 1 , startDate = None , token = '' , version = '' ) :
    '''This call allows you to fetch a number of trade dates or holidays from a given date . For example , if you want the next trading day , you would call / ref - data / us / dates / trade / next / 1.
 https : / / iexcloud . io / docs / api / # u - s - exchanges
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 type ( string ) ; " holiday " or " trade "
 direction ( string ) ; " next " or " last "
 last ( int ) ; number to move in direction
 startDate ( date ) ; start date for next or last , YYYYMMDD
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
dat = pd . DataFrame ( calendar ( type , direction , last , startDate , token , version ) )
_toDatetime ( dat )
return dat


signature:
def holidays ( direction = 'next' , last = 1 , startDate = None , token = '' , version = '' ) :
    '''This call allows you to fetch a number of trade dates or holidays from a given date . For example , if you want the next trading day , you would call / ref - data / us / dates / trade / next / 1.
 https : / / iexcloud . io / docs / api / # u - s - exchanges
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 direction ( string ) ; " next " or " last "
 last ( int ) ; number to move in direction
 startDate ( date ) ; start date for next or last , YYYYMMDD
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation: return calendar ( 'holiday' , direction , last , startDate , token , version )


signature:
def holidaysDF ( direction = 'next' , last = 1 , startDate = None , token = '' , version = '' ) :
    '''This call allows you to fetch a number of trade dates or holidays from a given date . For example , if you want the next trading day , you would call / ref - data / us / dates / trade / next / 1.
 https : / / iexcloud . io / docs / api / # u - s - exchanges
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 direction ( string ) ; " next " or " last "
 last ( int ) ; number to move in direction
 startDate ( date ) ; start date for next or last , YYYYMMDD
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation: return calendarDF ( 'holiday' , direction , last , startDate , token , version )


signature:
def internationalSymbols ( region = '' , exchange = '' , token = '' , version = '' ) :
    '''This call returns an array of international symbols that IEX Cloud supports for API calls .
 https : / / iexcloud . io / docs / api / # international - symbols
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 region ( string ) ; region , 2 letter case insensitive string of country codes using ISO 3166-1 alpha - 2
 exchange ( string ) : Case insensitive string of Exchange using IEX Supported Exchanges list
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
if region :
    return _getJson ( 'ref-data/region/{region}/symbols' . format ( region = region ) , token , version )
elif exchange :
    return _getJson ( 'ref-data/exchange/{exchange}/exchange' . format ( exchange = exchange ) , token , version )
return _getJson ( 'ref-data/region/us/symbols' , token , version )


signature:
def symbolsDF ( token = '' , version = '' ) :
    '''This call returns an array of symbols that IEX Cloud supports for API calls .
 https : / / iexcloud . io / docs / api / # symbols
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dataframe : result'''

implementation:
df = pd . DataFrame ( symbols ( token , version ) )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def iexSymbolsDF ( token = '' , version = '' ) :
    '''This call returns an array of symbols the Investors Exchange supports for trading .
 This list is updated daily as of 7:45 a . m . ET . Symbols may be added or removed by the Investors Exchange after the list was produced .
 https : / / iexcloud . io / docs / api / # iex - symbols
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
df = pd . DataFrame ( iexSymbols ( token , version ) )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def mutualFundSymbolsDF ( token = '' , version = '' ) :
    '''This call returns an array of mutual fund symbols that IEX Cloud supports for API calls .
 https : / / iexcloud . io / docs / api / # mutual - fund - symbols
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
df = pd . DataFrame ( mutualFundSymbols ( token , version ) )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def otcSymbolsDF ( token = '' , version = '' ) :
    '''This call returns an array of OTC symbols that IEX Cloud supports for API calls .
 https : / / iexcloud . io / docs / api / # otc - symbols
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
df = pd . DataFrame ( otcSymbols ( token , version ) )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def internationalSymbolsDF ( region = '' , exchange = '' , token = '' , version = '' ) :
    '''This call returns an array of international symbols that IEX Cloud supports for API calls .
 https : / / iexcloud . io / docs / api / # international - symbols
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 region ( string ) ; region , 2 letter case insensitive string of country codes using ISO 3166-1 alpha - 2
 exchange ( string ) : Case insensitive string of Exchange using IEX Supported Exchanges list
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
df = pd . DataFrame ( internationalSymbols ( region , exchange , token , version ) )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def internationalSymbolsList ( region = '' , exchange = '' , token = '' , version = '' ) :
    '''This call returns an array of international symbols that IEX Cloud supports for API calls .
 https : / / iexcloud . io / docs / api / # international - symbols
 8am , 9am , 12pm , 1pm UTC daily
 Args :
 region ( string ) ; region , 2 letter case insensitive string of country codes using ISO 3166-1 alpha - 2
 exchange ( string ) : Case insensitive string of Exchange using IEX Supported Exchanges list
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 list : result'''

implementation: return internationalSymbolsDF ( region , exchange , token , version ) . index . tolist ( )


signature:
def marketsDF ( token = '' , version = '' ) :
    '''https : / / iextrading . com / developer / docs / # intraday'''

implementation:
df = pd . DataFrame ( markets ( token , version ) )
_toDatetime ( df )
return df


signature:
def _getJson ( url , token = '' , version = '' ) :
    '''for backwards compat , accepting token and version but ignoring'''

implementation:
if token :
    return _getJsonIEXCloud ( url , token , version )
return _getJsonOrig ( url )


signature:
def _getJsonOrig ( url ) :
    '''internal'''

implementation:
url = _URL_PREFIX + url
resp = requests . get ( urlparse ( url ) . geturl ( ) , proxies = _PYEX_PROXIES )
if resp . status_code == 200 :
    return resp . json ( )
raise PyEXception ( 'Response %d - ' % resp . status_code , resp . text )


signature:
def _getJsonIEXCloud ( url , token = '' , version = 'beta' ) :
    '''for iex cloud'''

implementation:
url = _URL_PREFIX2 . format ( version = version ) + url
resp = requests . get ( urlparse ( url ) . geturl ( ) , proxies = _PYEX_PROXIES , params = { 'token' : token } )
if resp . status_code == 200 :
    return resp . json ( )
raise PyEXception ( 'Response %d - ' % resp . status_code , resp . text )


signature:
def _strOrDate ( st ) :
    '''internal'''

implementation:
if isinstance ( st , string_types ) :
    return st
elif isinstance ( st , datetime ) :
    return st . strftime ( '%Y%m%d' )
raise PyEXception ( 'Not a date: %s' , str ( st ) )


signature:
def _raiseIfNotStr ( s ) :
    '''internal'''

implementation:
if s is not None and not isinstance ( s , string_types ) :
    raise PyEXception ( 'Cannot use type %s' % str ( type ( s ) ) )


signature:
def _tryJson ( data , raw = True ) :
    '''internal'''

implementation:
if raw :
    return data
try :
    return json . loads ( data )
except ValueError :
    return data


signature:
def _stream ( url , sendinit = None , on_data = print ) :
    '''internal'''

implementation:
cl = WSClient ( url , sendinit = sendinit , on_data = on_data )
return cl


signature:
def _streamSSE ( url , on_data = print , accrue = False ) :
    '''internal'''

implementation:
messages = SSEClient ( url )
if accrue :
    ret = [ ]
for msg in messages :
    data = msg . data
    on_data ( json . loads ( data ) )
    if accrue :
        ret . append ( msg )
return ret


signature:
def _reindex ( df , col ) :
    '''internal'''

implementation:
if col in df . columns :
    df . set_index ( col , inplace = True )


signature:
def _toDatetime ( df , cols = None , tcols = None ) :
    '''internal'''

implementation:
cols = cols or _STANDARD_DATE_FIELDS
tcols = tcols = _STANDARD_TIME_FIELDS
for col in cols :
    if col in df :
        df [ col ] = pd . to_datetime ( df [ col ] , infer_datetime_format = True , errors = 'coerce' )
for tcol in tcols :
    if tcol in df :
        df [ tcol ] = pd . to_datetime ( df [ tcol ] , unit = 'ms' , errors = 'coerce' )


signature:
def balanceSheet ( symbol , token = '' , version = '' ) :
    '''Pulls balance sheet data . Available quarterly ( 4 quarters ) and annually ( 4 years )
 https : / / iexcloud . io / docs / api / # balance - sheet
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/balance-sheet' , token , version )


signature:
def balanceSheetDF ( symbol , token = '' , version = '' ) :
    '''Pulls balance sheet data . Available quarterly ( 4 quarters ) and annually ( 4 years )
 https : / / iexcloud . io / docs / api / # balance - sheet
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
val = balanceSheet ( symbol , token , version )
df = pd . io . json . json_normalize ( val , 'balancesheet' , 'symbol' )
_toDatetime ( df )
_reindex ( df , 'reportDate' )
return df


signature:
def batch ( symbols , fields = None , range_ = '1m' , last = 10 , token = '' , version = '' ) :
    '''Batch several data requests into one invocation
 https : / / iexcloud . io / docs / api / # batch - requests
 Args :
 symbols ( list ) ; List of tickers to request
 fields ( list ) ; List of fields to request
 range _ ( string ) ; Date range for chart
 last ( int ) ;
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : results in json'''

implementation:
fields = fields or _BATCH_TYPES [ : 10 ]
# limit 10
if not isinstance ( symbols , [ ] . __class__ ) :
    if not isinstance ( symbols , str ) :
        raise PyEXception ( 'batch expects string or list of strings for symbols argument' )
if isinstance ( fields , str ) :
    fields = [ fields ]
if range_ not in _TIMEFRAME_CHART :
    raise PyEXception ( 'Range must be in %s' % str ( _TIMEFRAME_CHART ) )
if isinstance ( symbols , str ) :
    route = 'stock/{}/batch?types={}&range={}&last={}' . format ( symbols , ',' . join ( fields ) , range_ , last )
    return _getJson ( route , token , version )
if len ( symbols ) > 100 :
    raise PyEXception ( 'IEX will only handle up to 100 symbols at a time!' )
route = 'stock/market/batch?symbols={}&types={}&range={}&last={}' . format ( ',' . join ( symbols ) , ',' . join ( fields ) , range_ , last )
return _getJson ( route , token , version )


signature:
def batchDF ( symbols , fields = None , range_ = '1m' , last = 10 , token = '' , version = '' ) :
    '''Batch several data requests into one invocation
 https : / / iexcloud . io / docs / api / # batch - requests
 Args :
 symbols ( list ) ; List of tickers to request
 fields ( list ) ; List of fields to request
 range _ ( string ) ; Date range for chart
 last ( int ) ;
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : results in json'''

implementation:
x = batch ( symbols , fields , range_ , last , token , version )
ret = { }
if isinstance ( symbols , str ) :
    for field in x . keys ( ) :
        ret [ field ] = _MAPPING [ field ] ( x [ field ] )
else :
    for symbol in x . keys ( ) :
        for field in x [ symbol ] . keys ( ) :
            if field not in ret :
                ret [ field ] = pd . DataFrame ( )
            dat = x [ symbol ] [ field ]
            dat = _MAPPING [ field ] ( dat )
            dat [ 'symbol' ] = symbol
            ret [ field ] = pd . concat ( [ ret [ field ] , dat ] , sort = True )
return ret


signature:
def bulkBatch ( symbols , fields = None , range_ = '1m' , last = 10 , token = '' , version = '' ) :
    '''Optimized batch to fetch as much as possible at once
 https : / / iexcloud . io / docs / api / # batch - requests
 Args :
 symbols ( list ) ; List of tickers to request
 fields ( list ) ; List of fields to request
 range _ ( string ) ; Date range for chart
 last ( int ) ;
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : results in json'''

implementation:
fields = fields or _BATCH_TYPES
args = [ ]
empty_data = [ ]
list_orig = empty_data . __class__
if not isinstance ( symbols , list_orig ) :
    raise PyEXception ( 'Symbols must be of type list' )
for i in range ( 0 , len ( symbols ) , 99 ) :
    args . append ( ( symbols [ i : i + 99 ] , fields , range_ , last , token , version ) )
pool = ThreadPool ( 20 )
rets = pool . starmap ( batch , args )
pool . close ( )
ret = { }
for i , d in enumerate ( rets ) :
    symbols_subset = args [ i ] [ 0 ]
    if len ( d ) != len ( symbols_subset ) :
        empty_data . extend ( list_orig ( set ( symbols_subset ) - set ( d . keys ( ) ) ) )
    ret . update ( d )
for k in empty_data :
    if k not in ret :
        if isinstance ( fields , str ) :
            ret [ k ] = { }
        else :
            ret [ k ] = { x : { } for x in fields }
return ret


signature:
def bulkBatchDF ( symbols , fields = None , range_ = '1m' , last = 10 , token = '' , version = '' ) :
    '''Optimized batch to fetch as much as possible at once
 https : / / iexcloud . io / docs / api / # batch - requests
 Args :
 symbols ( list ) ; List of tickers to request
 fields ( list ) ; List of fields to request
 range _ ( string ) ; Date range for chart
 last ( int ) ;
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : results in json'''

implementation:
dat = bulkBatch ( symbols , fields , range_ , last , token , version )
ret = { }
for symbol in dat :
    for field in dat [ symbol ] :
        if field not in ret :
            ret [ field ] = pd . DataFrame ( )
        d = dat [ symbol ] [ field ]
        d = _MAPPING [ field ] ( d )
        d [ 'symbol' ] = symbol
        ret [ field ] = pd . concat ( [ ret [ field ] , d ] , sort = True )
return ret


signature:
def _bookToDF ( b ) :
    '''internal'''

implementation:
quote = b . get ( 'quote' , [ ] )
asks = b . get ( 'asks' , [ ] )
bids = b . get ( 'bids' , [ ] )
trades = b . get ( 'trades' , [ ] )
df1 = pd . io . json . json_normalize ( quote )
df1 [ 'type' ] = 'quote'
df2 = pd . io . json . json_normalize ( asks )
df2 [ 'symbol' ] = quote [ 'symbol' ]
df2 [ 'type' ] = 'ask'
df3 = pd . io . json . json_normalize ( bids )
df3 [ 'symbol' ] = quote [ 'symbol' ]
df3 [ 'type' ] = 'bid'
df4 = pd . io . json . json_normalize ( trades )
df4 [ 'symbol' ] = quote [ 'symbol' ]
df3 [ 'type' ] = 'trade'
df = pd . concat ( [ df1 , df2 , df3 , df4 ] , sort = True )
_toDatetime ( df )
return df


signature:
def bookDF ( symbol , token = '' , version = '' ) :
    '''Book data
 https : / / iextrading . com / developer / docs / # book
 realtime during Investors Exchange market hours
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
x = book ( symbol , token , version )
df = _bookToDF ( x )
return df


signature:
def cashFlow ( symbol , token = '' , version = '' ) :
    '''Pulls cash flow data . Available quarterly ( 4 quarters ) or annually ( 4 years ) .
 https : / / iexcloud . io / docs / api / # cash - flow
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/cash-flow' , token , version )


signature:
def cashFlowDF ( symbol , token = '' , version = '' ) :
    '''Pulls cash flow data . Available quarterly ( 4 quarters ) or annually ( 4 years ) .
 https : / / iexcloud . io / docs / api / # cash - flow
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
val = cashFlow ( symbol , token , version )
df = pd . io . json . json_normalize ( val , 'cashflow' , 'symbol' )
_toDatetime ( df )
_reindex ( df , 'reportDate' )
df . replace ( to_replace = [ None ] , value = np . nan , inplace = True )
return df


signature:
def chart ( symbol , timeframe = '1m' , date = None , token = '' , version = '' ) :
    '''Historical price / volume data , daily and intraday
 https : / / iexcloud . io / docs / api / # historical - prices
 Data Schedule
 1d : - 9:30-4pm ET Mon - Fri on regular market trading days
 -9:30-1pm ET on early close trading days
 All others :
 - Prior trading day available after 4am ET Tue - Sat
 Args :
 symbol ( string ) ; Ticker to request
 timeframe ( string ) ; Timeframe to request e . g . 1m
 date ( datetime ) : date , if requesting intraday
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
if timeframe is not None and timeframe != '1d' :
    if timeframe not in _TIMEFRAME_CHART :
        raise PyEXception ( 'Range must be in %s' % str ( _TIMEFRAME_CHART ) )
    return _getJson ( 'stock/' + symbol + '/chart' + '/' + timeframe , token , version )
if date :
    date = _strOrDate ( date )
    return _getJson ( 'stock/' + symbol + '/chart' + '/date/' + date , token , version )
return _getJson ( 'stock/' + symbol + '/chart' , token , version )


signature:
def _chartToDF ( c ) :
    '''internal'''

implementation:
df = pd . DataFrame ( c )
_toDatetime ( df )
_reindex ( df , 'date' )
return df


signature:
def chartDF ( symbol , timeframe = '1m' , date = None , token = '' , version = '' ) :
    '''Historical price / volume data , daily and intraday
 https : / / iexcloud . io / docs / api / # historical - prices
 Data Schedule
 1d : - 9:30-4pm ET Mon - Fri on regular market trading days
 -9:30-1pm ET on early close trading days
 All others :
 - Prior trading day available after 4am ET Tue - Sat
 Args :
 symbol ( string ) ; Ticker to request
 timeframe ( string ) ; Timeframe to request e . g . 1m
 date ( datetime ) : date , if requesting intraday
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
c = chart ( symbol , timeframe , date , token , version )
df = pd . DataFrame ( c )
_toDatetime ( df )
if timeframe is not None and timeframe != '1d' :
    _reindex ( df , 'date' )
else :
    if not df . empty :
        df . set_index ( [ 'date' , 'minute' ] , inplace = True )
    else :
        return pd . DataFrame ( )
return df


signature:
def bulkMinuteBars ( symbol , dates , token = '' , version = '' ) :
    '''fetch many dates worth of minute - bars for a given symbol'''

implementation:
_raiseIfNotStr ( symbol )
dates = [ _strOrDate ( date ) for date in dates ]
list_orig = dates . __class__
args = [ ]
for date in dates :
    args . append ( ( symbol , '1d' , date , token , version ) )
pool = ThreadPool ( 20 )
rets = pool . starmap ( chart , args )
pool . close ( )
return list_orig ( itertools . chain ( * rets ) )


signature:
def bulkMinuteBarsDF ( symbol , dates , token = '' , version = '' ) :
    '''fetch many dates worth of minute - bars for a given symbol'''

implementation:
data = bulkMinuteBars ( symbol , dates , token , version )
df = pd . DataFrame ( data )
if df . empty :
    return df
_toDatetime ( df )
df . set_index ( [ 'date' , 'minute' ] , inplace = True )
return df


signature:
def collections ( tag , collectionName , token = '' , version = '' ) :
    '''Returns an array of quote objects for a given collection type . Currently supported collection types are sector , tag , and list
 https : / / iexcloud . io / docs / api / # collections
 Args :
 tag ( string ) ; Sector , Tag , or List
 collectionName ( string ) ; Associated name for tag
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
if tag not in _COLLECTION_TAGS :
    raise PyEXception ( 'Tag must be in %s' % str ( _COLLECTION_TAGS ) )
return _getJson ( 'stock/market/collection/' + tag + '?collectionName=' + collectionName , token , version )


signature:
def collectionsDF ( tag , query , token = '' , version = '' ) :
    '''Returns an array of quote objects for a given collection type . Currently supported collection types are sector , tag , and list
 https : / / iexcloud . io / docs / api / # collections
 Args :
 tag ( string ) ; Sector , Tag , or List
 collectionName ( string ) ; Associated name for tag
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
df = pd . DataFrame ( collections ( tag , query , token , version ) )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def company ( symbol , token = '' , version = '' ) :
    '''Company reference data
 https : / / iexcloud . io / docs / api / # company
 Updates at 4am and 5am UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/company' , token , version )


signature:
def _companyToDF ( c , token = '' , version = '' ) :
    '''internal'''

implementation:
df = pd . io . json . json_normalize ( c )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def companyDF ( symbol , token = '' , version = '' ) :
    '''Company reference data
 https : / / iexcloud . io / docs / api / # company
 Updates at 4am and 5am UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
c = company ( symbol , token , version )
df = _companyToDF ( c )
return df


signature:
def delayedQuote ( symbol , token = '' , version = '' ) :
    '''This returns the 15 minute delayed market quote .
 https : / / iexcloud . io / docs / api / # delayed - quote
 15min delayed
 4:30am - 8pm ET M - F when market is open
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/delayed-quote' , token , version )


signature:
def delayedQuoteDF ( symbol , token = '' , version = '' ) :
    '''This returns the 15 minute delayed market quote .
 https : / / iexcloud . io / docs / api / # delayed - quote
 15min delayed
 4:30am - 8pm ET M - F when market is open
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
df = pd . io . json . json_normalize ( delayedQuote ( symbol , token , version ) )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def dividends ( symbol , timeframe = 'ytd' , token = '' , version = '' ) :
    '''Dividend history
 https : / / iexcloud . io / docs / api / # dividends
 Updated at 9am UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
if timeframe not in _TIMEFRAME_DIVSPLIT :
    raise PyEXception ( 'Range must be in %s' % str ( _TIMEFRAME_DIVSPLIT ) )
return _getJson ( 'stock/' + symbol + '/dividends/' + timeframe , token , version )


signature:
def _dividendsToDF ( d ) :
    '''internal'''

implementation:
df = pd . DataFrame ( d )
_toDatetime ( df )
_reindex ( df , 'exDate' )
return df


signature:
def dividendsDF ( symbol , timeframe = 'ytd' , token = '' , version = '' ) :
    '''Dividend history
 https : / / iexcloud . io / docs / api / # dividends
 Updated at 9am UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
d = dividends ( symbol , timeframe , token , version )
df = _dividendsToDF ( d )
return df


signature:
def earnings ( symbol , token = '' , version = '' ) :
    '''Earnings data for a given company including the actual EPS , consensus , and fiscal period . Earnings are available quarterly ( last 4 quarters ) and annually ( last 4 years ) .
 https : / / iexcloud . io / docs / api / # earnings
 Updates at 9am , 11am , 12pm UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/earnings' , token , version )


signature:
def _earningsToDF ( e ) :
    '''internal'''

implementation:
if e :
    df = pd . io . json . json_normalize ( e , 'earnings' , 'symbol' )
    _toDatetime ( df )
    _reindex ( df , 'EPSReportDate' )
else :
    df = pd . DataFrame ( )
return df


signature:
def earningsDF ( symbol , token = '' , version = '' ) :
    '''Earnings data for a given company including the actual EPS , consensus , and fiscal period . Earnings are available quarterly ( last 4 quarters ) and annually ( last 4 years ) .
 https : / / iexcloud . io / docs / api / # earnings
 Updates at 9am , 11am , 12pm UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
e = earnings ( symbol , token , version )
df = _earningsToDF ( e )
return df


signature:
def earningsTodayDF ( token = '' , version = '' ) :
    '''Returns earnings that will be reported today as two arrays : before the open bto and after market close amc .
 Each array contains an object with all keys from earnings , a quote object , and a headline key .
 https : / / iexcloud . io / docs / api / # earnings - today
 Updates at 9am , 11am , 12pm UTC daily
 Args :
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
x = earningsToday ( token , version )
z = [ ]
for k in x :
    ds = x [ k ]
    for d in ds :
        d [ 'when' ] = k
        z . extend ( ds )
df = pd . io . json . json_normalize ( z )
if not df . empty :
    df . drop_duplicates ( inplace = True )
_toDatetime ( df )
_reindex ( df , 'symbol' )
return df


signature:
def spread ( symbol , token = '' , version = '' ) :
    '''This returns an array of effective spread , eligible volume , and price improvement of a stock , by market .
 Unlike volume - by - venue , this will only return a venue if effective spread is not ‘ N / A ’ . Values are sorted in descending order by effectiveSpread .
 Lower effectiveSpread and higher priceImprovement values are generally considered optimal .
 Effective spread is designed to measure marketable orders executed in relation to the market center ’ s
 quoted spread and takes into account hidden and midpoint liquidity available at each market center .
 Effective Spread is calculated by using eligible trade prices recorded to the consolidated tape and
 comparing those trade prices to the National Best Bid and Offer ( “ NBBO ” ) at the time of the execution .
 View the data disclaimer at the bottom of the stocks app for more information about how these values are calculated .
 https : / / iexcloud . io / docs / api / # earnings - today
 8am ET M - F
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/effective-spread' , token , version )


signature:
def spreadDF ( symbol , token = '' , version = '' ) :
    '''This returns an array of effective spread , eligible volume , and price improvement of a stock , by market .
 Unlike volume - by - venue , this will only return a venue if effective spread is not ‘ N / A ’ . Values are sorted in descending order by effectiveSpread .
 Lower effectiveSpread and higher priceImprovement values are generally considered optimal .
 Effective spread is designed to measure marketable orders executed in relation to the market center ’ s
 quoted spread and takes into account hidden and midpoint liquidity available at each market center .
 Effective Spread is calculated by using eligible trade prices recorded to the consolidated tape and
 comparing those trade prices to the National Best Bid and Offer ( “ NBBO ” ) at the time of the execution .
 View the data disclaimer at the bottom of the stocks app for more information about how these values are calculated .
 https : / / iexcloud . io / docs / api / # earnings - today
 8am ET M - F
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
df = pd . DataFrame ( spread ( symbol , token , version ) )
_toDatetime ( df )
_reindex ( df , 'venue' )
return df


signature:
def estimates ( symbol , token = '' , version = '' ) :
    '''Provides the latest consensus estimate for the next fiscal period
 https : / / iexcloud . io / docs / api / # estimates
 Updates at 9am , 11am , 12pm UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/estimates' , token , version )


signature:
def estimatesDF ( symbol , token = '' , version = '' ) :
    '''Provides the latest consensus estimate for the next fiscal period
 https : / / iexcloud . io / docs / api / # estimates
 Updates at 9am , 11am , 12pm UTC every day
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
f = estimates ( symbol , token , version )
df = _estimatesToDF ( f )
return df


signature:
def financials ( symbol , token = '' , version = '' ) :
    '''Pulls income statement , balance sheet , and cash flow data from the four most recent reported quarters .
 https : / / iexcloud . io / docs / api / # financials
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/financials' , token , version )


signature:
def _financialsToDF ( f ) :
    '''internal'''

implementation:
if f :
    df = pd . io . json . json_normalize ( f , 'financials' , 'symbol' )
    _toDatetime ( df )
    _reindex ( df , 'reportDate' )
else :
    df = pd . DataFrame ( )
return df


signature:
def financialsDF ( symbol , token = '' , version = '' ) :
    '''Pulls income statement , balance sheet , and cash flow data from the four most recent reported quarters .
 https : / / iexcloud . io / docs / api / # financials
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
f = financials ( symbol , token , version )
df = _financialsToDF ( f )
return df


signature:
def incomeStatement ( symbol , token = '' , version = '' ) :
    '''Pulls income statement data . Available quarterly ( 4 quarters ) or annually ( 4 years ) .
 https : / / iexcloud . io / docs / api / # income - statement
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/income' , token , version )


signature:
def incomeStatementDF ( symbol , token = '' , version = '' ) :
    '''Pulls income statement data . Available quarterly ( 4 quarters ) or annually ( 4 years ) .
 https : / / iexcloud . io / docs / api / # income - statement
 Updates at 8am , 9am UTC daily
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
val = incomeStatement ( symbol , token , version )
df = pd . io . json . json_normalize ( val , 'income' , 'symbol' )
_toDatetime ( df )
_reindex ( df , 'reportDate' )
return df


signature:
def ipoTodayDF ( token = '' , version = '' ) :
    '''This returns a list of upcoming or today IPOs scheduled for the current and next month . The response is split into two structures :
 rawData and viewData . rawData represents all available data for an IPO . viewData represents data structured for display to a user .
 https : / / iexcloud . io / docs / api / # ipo - calendar
 10am , 10:30am UTC daily
 Args :
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
val = ipoToday ( token , version )
if val :
    df = pd . io . json . json_normalize ( val , 'rawData' )
    _toDatetime ( df )
    _reindex ( df , 'symbol' )
else :
    df = pd . DataFrame ( )
return df


signature:
def ipoUpcomingDF ( token = '' , version = '' ) :
    '''This returns a list of upcoming or today IPOs scheduled for the current and next month . The response is split into two structures :
 rawData and viewData . rawData represents all available data for an IPO . viewData represents data structured for display to a user .
 https : / / iexcloud . io / docs / api / # ipo - calendar
 10am , 10:30am UTC daily
 Args :
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 DataFrame : result'''

implementation:
val = ipoUpcoming ( token , version )
if val :
    df = pd . io . json . json_normalize ( val , 'rawData' )
    _toDatetime ( df )
    _reindex ( df , 'symbol' )
else :
    df = pd . DataFrame ( )
return df


signature:
def keyStats ( symbol , token = '' , version = '' ) :
    '''Key Stats about company
 https : / / iexcloud . io / docs / api / # key - stats
 8am , 9am ET
 Args :
 symbol ( string ) ; Ticker to request
 token ( string ) ; Access token
 version ( string ) ; API version
 Returns :
 dict : result'''

implementation:
_raiseIfNotStr ( symbol )
return _getJson ( 'stock/' + symbol + '/stats' , token , version )


signature:
def _statsToDF ( s ) :
    '''internal'''

implementation:
if s :
    df = pd . io . json . json_normalize ( s )
    _toDatetime ( df )
    _reindex ( df , 'symbol' )
else :
    df = pd . DataFrame ( )
return df