original_docstring: Import index snapshot into current collection index.
docstring: Import index snapshot into current collection index.
docstring_tokens: Import index snapshot into current collection index .
code: def import_index(delete=False, stream=None):
      """Import index snapshot into current collection index."""
  
      es_index = collections.current().es_index
      if delete:
          delete_index(es_index)
  
      with snapshot_repo() as (repo, repo_path):
          log.info('Unpack tar archive')
  
          tar = tarfile.open(mode='r|*', fileobj=stream or sys.stdin.buffer)
          tar.extractall(repo_path)
          tar.close()
  
          snapshots_resp = requests.get(f'{repo}/*')
          check_response(snapshots_resp)
          for s in snapshots_resp.json()['snapshots']:
              if s['state'] == 'SUCCESS':
                  [snapshot_index] = s['indices']
                  if snapshot_index != es_index:
                      continue
                  snapshot = f'{repo}/{s["snapshot"]}'
                  break
          else:
              raise RuntimeError(f"No snapshots for index {es_index}")
  
          log.info("Starting restore for index %r as %r", snapshot_index, es_index)
          restore = f'{snapshot}/_restore'
          restore_resp = post_json(restore, {
              'indices': es_index,
              'include_global_state': False,
              'include_aliases': False,
          })
          check_response(restore_resp)
          assert restore_resp.json()['accepted']
  
          status = f'{ES_URL}/{es_index}/_recovery'
          while True:
              status_resp = requests.get(status)
              check_response(status_resp)
              if not status_resp.json():
                  log.debug("Waiting for restore to start")
                  time.sleep(1)
                  continue
  
              for shard in status_resp.json()[es_index]['shards']:
                  stage = shard['stage']
                  if stage != 'DONE':
                      log.debug("Shard %r stage=%r", shard['id'], stage)
                      time.sleep(1)
                      break
  
              else:
                  log.info('Snapshot restored successfully')
                  break
code_tokens: def import_index ( delete = False , stream = None ) : es_index = collections . current ( ) . es_index if delete : delete_index ( es_index ) with snapshot_repo ( ) as ( repo , repo_path ) : log . info ( 'Unpack tar archive' ) tar = tarfile . open ( mode = 'r|*' , fileobj = stream or sys . stdin . buffer ) tar . extractall ( repo_path ) tar . close ( ) snapshots_resp = requests . get ( f'{repo}/*' ) check_response ( snapshots_resp ) for s in snapshots_resp . json ( ) [ 'snapshots' ] : if s [ 'state' ] == 'SUCCESS' : [ snapshot_index ] = s [ 'indices' ] if snapshot_index != es_index : continue snapshot = f'{repo}/{s["snapshot"]}' break else : raise RuntimeError ( f"No snapshots for index {es_index}" ) log . info ( "Starting restore for index %r as %r" , snapshot_index , es_index ) restore = f'{snapshot}/_restore' restore_resp = post_json ( restore , { 'indices' : es_index , 'include_global_state' : False , 'include_aliases' : False , } ) check_response ( restore_resp ) assert restore_resp . json ( ) [ 'accepted' ] status = f'{ES_URL}/{es_index}/_recovery' while True : status_resp = requests . get ( status ) check_response ( status_resp ) if not status_resp . json ( ) : log . debug ( "Waiting for restore to start" ) time . sleep ( 1 ) continue for shard in status_resp . json ( ) [ es_index ] [ 'shards' ] : stage = shard [ 'stage' ] if stage != 'DONE' : log . debug ( "Shard %r stage=%r" , shard [ 'id' ] , stage ) time . sleep ( 1 ) break else : log . info ( 'Snapshot restored successfully' ) break
short_docstring: Import index snapshot into current collection index.
short_docstring_tokens: Import index snapshot into current collection index .
comment: """Import index snapshot into current collection index."""