def _PerformBulkload(arg_dict,
                     oauth2_parameters=None,
                     check_file=CheckFile,
                     check_output_file=CheckOutputFile):
  app_id = arg_dict['application']
  url = arg_dict['url']
  filename = arg_dict['filename']
  batch_size = arg_dict['batch_size']
  kind = arg_dict['kind']
  num_threads = arg_dict['num_threads']
  bandwidth_limit = arg_dict['bandwidth_limit']
  rps_limit = arg_dict['rps_limit']
  http_limit = arg_dict['http_limit']
  db_filename = arg_dict['db_filename']
  config_file = arg_dict['config_file']
  auth_domain = arg_dict['auth_domain']
  has_header = arg_dict['has_header']
  download = arg_dict['download']
  result_db_filename = arg_dict['result_db_filename']
  loader_opts = arg_dict['loader_opts']
  exporter_opts = arg_dict['exporter_opts']
  mapper_opts = arg_dict['mapper_opts']
  perform_map = arg_dict['map']
  dump = arg_dict['dump']
  restore = arg_dict['restore']
  create_config = arg_dict['create_config']
  namespace = arg_dict['namespace']
  dry_run = arg_dict['dry_run']
  throttle_class = (arg_dict['throttle_class']
                    or remote_api_throttle.ThrottledHttpRpcServerOAuth2)

  if namespace:
    namespace_manager.set_namespace(namespace)
  os.environ['AUTH_DOMAIN'] = auth_domain

  kind = arg_dict['kind'] = ParseKind(kind)

  if not dump and not restore and not create_config:

    check_file(config_file)

  if download or dump or create_config:

    check_output_file(filename)
  elif not perform_map:
    check_file(filename)


  throttle_layout = ThrottleLayout(bandwidth_limit, http_limit, rps_limit)
  logger.info('Throttling transfers:')
  logger.info('Bandwidth: %s bytes/second', bandwidth_limit)
  logger.info('HTTP connections: %s/second', http_limit)
  logger.info('Entities inserted/fetched/modified: %s/second', rps_limit)
  logger.info('Batch Size: %s', batch_size)

  throttle = remote_api_throttle.Throttle(layout=throttle_layout)


  throttle.Register(threading.currentThread())
  threading.currentThread().exit_flag = False


  server = None
  if not app_id:
    if dry_run:

      raise ConfigurationError('Must sepcify application ID in dry run mode.')
    app_id, server = _GetRemoteAppId(url, throttle, oauth2_parameters,
                                     throttle_class=throttle_class)

    arg_dict['application'] = app_id

  if dump:
    Exporter.RegisterExporter(DumpExporter(kind, result_db_filename))
  elif restore:
    Loader.RegisterLoader(RestoreLoader(kind, app_id))
  elif create_config:

    kind = '__Stat_PropertyType_PropertyName_Kind__'
    arg_dict['kind'] = kind


    root_dir = os.path.dirname(os.path.abspath(__file__))
    if os.path.basename(root_dir) == 'tools':
      root_dir = os.path.dirname(os.path.dirname(os.path.dirname(root_dir)))

    LoadYamlConfig(os.path.join(
        root_dir, os.path.normpath(
            'google/appengine/ext/bulkload/bulkloader_wizard.yaml')))
  elif (config_file and
        (config_file.endswith('.yaml') or config_file.endswith('.yml'))):
    LoadYamlConfig(config_file)
  else:
    LoadConfig(config_file)

  os.environ['APPLICATION_ID'] = app_id

  signature = _MakeSignature(app_id=app_id,
                             url=url,
                             kind=kind,
                             db_filename=db_filename,
                             download=download,
                             perform_map=perform_map,
                             has_header=has_header,
                             result_db_filename=result_db_filename,
                             dump=dump,
                             restore=restore)



  max_queue_size = max(DEFAULT_QUEUE_SIZE, 3 * num_threads + 5)

  upload = not (download or dump or restore or perform_map or create_config)

  if db_filename == 'skip':
    progress_db = StubProgressDatabase()
  elif upload or restore:
    progress_db = ProgressDatabase(db_filename, signature)
  else:
    progress_db = ExportProgressDatabase(db_filename, signature)

  return_code = 1

  if upload or restore:
    loader = Loader.RegisteredLoader(kind)
    try:
      loader.initialize(filename, loader_opts)


      workitem_generator_factory = GetCSVGeneratorFactory(
          kind, filename, batch_size, has_header)

      app = BulkUploaderApp(arg_dict,
                            workitem_generator_factory,
                            throttle,
                            progress_db,
                            ProgressTrackerThread,
                            max_queue_size,
                            RequestManager,
                            DataSourceThread,
                            Queue.Queue,
                            throttle_class=throttle_class,
                            server=server,
                            oauth2_parameters=oauth2_parameters)
      try:
        return_code = app.Run()
      except AuthenticationError:
        logger.error(AUTH_FAILED_MESSAGE)
    finally:
      loader.finalize()
  elif download or dump or create_config:
    exporter = Exporter.RegisteredExporter(kind)
    result_db = ResultDatabase(result_db_filename, signature, exporter=exporter)
    try:
      exporter.initialize(filename, exporter_opts)

      def KeyRangeGeneratorFactory(request_manager, progress_queue,
                                   progress_gen, kinds):
        logger.info('Downloading kinds: %s', kinds)
        return KeyRangeItemGenerator(request_manager, kinds, progress_queue,
                                     progress_gen, DownloadItem)

      def ExportProgressThreadFactory(progress_queue, progress_db):
        return ExportProgressThread(exporter,
                                    progress_queue,
                                    progress_db,
                                    result_db)

      app = BulkDownloaderApp(arg_dict,
                              KeyRangeGeneratorFactory,
                              throttle,
                              progress_db,
                              ExportProgressThreadFactory,
                              0,
                              RequestManager,
                              DataSourceThread,
                              Queue.Queue,
                              throttle_class=throttle_class,
                              server=server)
      try:
        return_code = app.Run()
      except AuthenticationError:
        logger.error(AUTH_FAILED_MESSAGE)
      except KindStatError:
        logger.error('Unable to download kind stats for all-kinds download.')
        logger.error('Kind stats are generated periodically by the appserver')
        logger.error('Kind stats are not available on dev_appserver.')
    finally:
      exporter.finalize()
  elif perform_map:
    mapper = Mapper.RegisteredMapper(kind)
    try:
      mapper.initialize(mapper_opts)

      def KeyRangeGeneratorFactory(request_manager, progress_queue,
                                   progress_gen, kinds):
        return KeyRangeItemGenerator(request_manager, kinds, progress_queue,
                                     progress_gen, MapperItem)

      def MapperProgressThreadFactory(progress_queue, progress_db):
        return MapperProgressThread(mapper,
                                    progress_queue,
                                    progress_db)

      app = BulkMapperApp(arg_dict,
                          KeyRangeGeneratorFactory,
                          throttle,
                          progress_db,
                          MapperProgressThreadFactory,
                          0,
                          RequestManager,
                          DataSourceThread,
                          Queue.Queue,
                          throttle_class=throttle_class,
                          server=server)
      try:
        return_code = app.Run()
      except AuthenticationError:
        logger.error(AUTH_FAILED_MESSAGE)
    finally:
      mapper.finalize()
  return return_code
Runs the bulkloader, given the command line options.
dataset/ETHPy150Open GoogleCloudPlatform/python-compat-runtime/appengine-compat/exported_appengine_sdk/google/appengine/tools/bulkloader.py/_PerformBulkload