original_docstring:  Start dataflow in the pipeline and wait for it to complete 
docstring: Start dataflow in the pipeline and wait for it to complete
docstring_tokens: Start dataflow in the pipeline and wait for it to complete
code: def run(self):
          """ Start dataflow in the pipeline and wait for it to complete """
  
          if len(self.nodes) < 2:
              raise Exception('There needs to be at least two nodes in your pipeline. For example,\n'
                              'pipeline:\n'
                              '    binance:\n'
                              '        ...\n'
                              '    backtest:\n'
                              '        ...')
  
          if not isinstance(self.nodes[0], Producer):
              raise Exception('The first node in your pipeline must be a Producer. For example,\n'
                              'pipeline:\n'
                              '    binance:  <- Producer\n'
                              '        ...')
  
          if not isinstance(self.nodes[-1], Consumer):
              raise Exception('The last node in your pipeline must be a Consumer. For example,\n'
                              'pipeline:\n'
                              '   ...\n'
                              '   backtest:  <- Consumer\n'
                              '       ...')
          for node in self.nodes[1:-1]:
              if not isinstance(node, Producer) or not isinstance(node, Consumer):
                  raise Exception('The "' + str(node) + '" node is not a Producer and Consumer.\n'
                                                       'For the pipeline to be valid, intermediate nodes must fill both of '
                                                       'these roles.')
  
          loop = asyncio.get_event_loop()
          logger.debug(str(self) + ' starting all edge coroutines')
          loop.run_until_complete(asyncio.gather(*[edge.open() for edge in self._chain()]))
          logger.debug(str(self) + ' all coroutines exited')
code_tokens: def run ( self ) : if len ( self . nodes ) < 2 : raise Exception ( 'There needs to be at least two nodes in your pipeline. For example,\n' 'pipeline:\n' '    binance:\n' '        ...\n' '    backtest:\n' '        ...' ) if not isinstance ( self . nodes [ 0 ] , Producer ) : raise Exception ( 'The first node in your pipeline must be a Producer. For example,\n' 'pipeline:\n' '    binance:  <- Producer\n' '        ...' ) if not isinstance ( self . nodes [ - 1 ] , Consumer ) : raise Exception ( 'The last node in your pipeline must be a Consumer. For example,\n' 'pipeline:\n' '   ...\n' '   backtest:  <- Consumer\n' '       ...' ) for node in self . nodes [ 1 : - 1 ] : if not isinstance ( node , Producer ) or not isinstance ( node , Consumer ) : raise Exception ( 'The "' + str ( node ) + '" node is not a Producer and Consumer.\n' 'For the pipeline to be valid, intermediate nodes must fill both of ' 'these roles.' ) loop = asyncio . get_event_loop ( ) logger . debug ( str ( self ) + ' starting all edge coroutines' ) loop . run_until_complete ( asyncio . gather ( * [ edge . open ( ) for edge in self . _chain ( ) ] ) ) logger . debug ( str ( self ) + ' all coroutines exited' )
short_docstring: Start dataflow in the pipeline and wait for it to complete
short_docstring_tokens: Start dataflow in the pipeline and wait for it to complete
comment: """ Start dataflow in the pipeline and wait for it to complete """