files: #!/usr/bin/env python
  #
  # Copyright 2007 Google Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  
  
  
  
  
  """Handler for data backup operation.
  
  Generic datastore admin console transfers control to ConfirmBackupHandler
  after selection of entities. The ConfirmBackupHandler confirms with user
  his choice, enters a backup name and transfers control to
  DoBackupHandler. DoBackupHandler starts backup mappers and displays confirmation
  page.
  
  This module also contains actual mapper code for backing data over.
  """
  
  
  
  
  
  
  
  
  from __future__ import with_statement
  
  
  
  import copy
  import cStringIO
  import datetime
  import itertools
  import logging
  import os
  import random
  import re
  import time
  import urllib
  import xml.dom.minidom
  
  from google.appengine._internal import cloudstorage
  
  # AppScale: Use bundled mapreduce library.
  from google.appengine.ext.mapreduce import context
  from google.appengine.ext.mapreduce import datastore_range_iterators as db_iters
  from google.appengine.ext.mapreduce import input_readers
  from google.appengine.ext.mapreduce import json_util
  from google.appengine.ext.mapreduce import operation as op
  from google.appengine.ext.mapreduce import output_writers
  from google.appengine.ext.mapreduce import parameters
  from google.appengine.ext.mapreduce import records
  
  from google.appengine.api import apiproxy_stub_map
  from google.appengine.api import app_identity
  from google.appengine.api import blobstore as blobstore_api
  from google.appengine.api import capabilities
  from google.appengine.api import datastore
  from google.appengine.api import datastore_errors
  from google.appengine.api import datastore_types
  from google.appengine.api import taskqueue
  from google.appengine.api import urlfetch
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.ext import blobstore
  from google.appengine.ext import db
  from google.appengine.ext import deferred
  import webapp2 as webapp
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext.datastore_admin import backup_pb2
  from google.appengine.ext.datastore_admin import config
  from google.appengine.ext.datastore_admin import utils
  
  
  XSRF_ACTION = 'backup'
  BUCKET_PATTERN = (r'^([a-zA-Z0-9]+([\-_]+[a-zA-Z0-9]+)*)'
                    r'(\.([a-zA-Z0-9]+([\-_]+[a-zA-Z0-9]+)*))*$')
  MAX_BUCKET_LEN = 222
  MIN_BUCKET_LEN = 3
  MAX_BUCKET_SEGMENT_LEN = 63
  NUM_KINDS_DEFERRED_THRESHOLD = 10
  MAX_BLOBS_PER_DELETE = 500
  TEST_WRITE_FILENAME_PREFIX = 'datastore_backup_write_test'
  MAX_KEYS_LIST_SIZE = 100
  MAX_TEST_FILENAME_TRIES = 10
  
  MEANING_TO_PRIMITIVE_TYPE = {
      entity_pb.Property.GD_WHEN: backup_pb2.EntitySchema.DATE_TIME,
      entity_pb.Property.GD_RATING: backup_pb2.EntitySchema.RATING,
      entity_pb.Property.ATOM_LINK: backup_pb2.EntitySchema.LINK,
      entity_pb.Property.ATOM_CATEGORY: backup_pb2.EntitySchema.CATEGORY,
      entity_pb.Property.GD_PHONENUMBER: backup_pb2.EntitySchema.PHONE_NUMBER,
      entity_pb.Property.GD_POSTALADDRESS: backup_pb2.EntitySchema.POSTAL_ADDRESS,
      entity_pb.Property.GD_EMAIL: backup_pb2.EntitySchema.EMAIL,
      entity_pb.Property.GD_IM: backup_pb2.EntitySchema.IM_HANDLE,
      entity_pb.Property.BLOBKEY: backup_pb2.EntitySchema.BLOB_KEY,
      entity_pb.Property.TEXT: backup_pb2.EntitySchema.TEXT,
      entity_pb.Property.BLOB: backup_pb2.EntitySchema.BLOB,
      entity_pb.Property.BYTESTRING: backup_pb2.EntitySchema.SHORT_BLOB
  }
  
  
  
  
  FILES_API_GS_FILESYSTEM = 'gs'
  FILES_API_BLOBSTORE_FILESYSTEM = 'blobstore'
  FILES_API_BLOBSTORE_PREFIX = '/blobstore/'
  
  BLOBSTORE_BACKUP_DISABLED_FEATURE_NAME = 'DisableBlobstoreBackups'
  
  BLOBSTORE_BACKUP_DISABLED_ERROR_MSG = (
      'Backups to blobstore are disabled, see '
      'https://cloud.google.com/appengine/docs/deprecations/blobstore_backups')
  
  BACKUP_RESTORE_HANDLER = __name__ + '.RestoreEntity.map'
  
  
  def _get_gcs_path_prefix_from_params_dict(params):
    """Returs the gcs_path_prefix from request or mapreduce dict.
  
    Args:
      params: A dict of parameters.
  
    Returns:
      A string containing GCS prefix for the backup. It can be either just the
      bucket or a bucket with a path (as present in the dictionary). Return None
      if 'gs_bucket_name' is not present in the dictionary.
    """
  
  
  
  
  
    return params.get('gs_bucket_name')
  
  
  def _get_basic_mapper_params(handler):
    namespace = handler.request.get('namespace', None)
    if namespace == '*':
      namespace = None
    return {'namespace': namespace}
  
  
  class GCSUtil(object):
    """Wrappers for cloudstorage methods to deal with /gs/ prefix.
  
    Files API requires GCS files to have a /gs/ prefix. cloudstorage treats /gs/
    as a regular bucket name. This class transparently strips the /gs prefix
    before passing the filename to cloudstorage. Also, for listbucket it
    adds the /gs/ back.
  
    backup_handler should not use cloudstorage directly - all access should
    be proxied by GCSUtil.
    """
  
    @classmethod
    def _strip_gs_prefix_if_present(cls, name):
      if name.startswith('/gs/'):
        return name[3:]
      return name
  
    @classmethod
    def add_gs_prefix_if_missing(cls, name):
      if not name.startswith('/gs/'):
        return '/gs%s' % name
      return name
  
    @classmethod
    def open(cls, filename, *args, **kwargs):
      return cloudstorage.open(
          cls._strip_gs_prefix_if_present(filename), *args, **kwargs)
  
    @classmethod
    def listbucket(cls, prefix, *args, **kwargs):
      output = []
      for f in cloudstorage.listbucket(cls._strip_gs_prefix_if_present(prefix),
                                       *args, **kwargs):
        f.filename = cls.add_gs_prefix_if_missing(f.filename)
        output.append(f)
      return output
  
    @classmethod
    def delete(cls, filename, *args, **kwargs):
      return cloudstorage.delete(
          cls._strip_gs_prefix_if_present(filename), *args, **kwargs)
  
  
  def _MaybeStripBlobstorePrefix(blob_key):
    """Strip /blobstore/ so that this reader can accept files API paths."""
    if blob_key.startswith(FILES_API_BLOBSTORE_PREFIX):
      blob_key = blob_key[len(FILES_API_BLOBSTORE_PREFIX):]
    return blob_key
  
  
  class BlobstoreReaderParamsException(Exception):
    pass
  
  
  class BlobstoreRecordsReader(input_readers.InputReader):
    """Reads records from blobstore without files API dependency.
  
    This reader accesses blobstore using (read-only) blobstore API. It only allows
    reading LevelDB records.
  
    On to_json() we record how many bytes we've read and on from_json() we
    seek to the previous location.
  
    Files API prepends blobstore keys with /blobstore/. We need to accept same
    params as files API so we silently strip /blobstore/ from any path.
    """
  
  
    BLOB_KEYS_PARAM = 'files'
  
    def __init__(self, blob_key, start_position):
      self._blob_key = _MaybeStripBlobstorePrefix(blob_key)
      self._records_reader = records.RecordsReader(
          blobstore.BlobReader(self._blob_key))
      self._records_reader.seek(start_position)
  
    @classmethod
    def _get_params(cls, mapper_spec):
  
      return input_readers._get_params(mapper_spec)
  
    @classmethod
    @db.non_transactional(allow_existing=True)
    def validate(cls, mapper_spec):
      params = cls._get_params(mapper_spec)
      if cls.BLOB_KEYS_PARAM not in params or not params[cls.BLOB_KEYS_PARAM]:
        raise BlobstoreReaderParamsException
  
    def __iter__(self):
      for record in self._records_reader:
        yield record
  
    def to_json(self):
      return self._blob_key, self._records_reader.tell()
  
    @classmethod
    def from_json(cls, json):
      return cls(*json)
  
    @classmethod
    def split_input(cls, mapper_spec):
  
      blob_keys = cls._get_params(mapper_spec)[cls.BLOB_KEYS_PARAM]
      return [cls(blob_key, 0) for blob_key in blob_keys]
  
  
  class ConfirmBackupHandler(webapp.RequestHandler):
    """Handler to deal with requests from the admin console to backup data."""
  
    SUFFIX = 'confirm_backup'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      kinds = handler.request.get_all('kind')
      sizes_known, size_total, remainder = utils.ParseKindsAndSizes(kinds)
      notreadonly_warning = capabilities.CapabilitySet(
          'datastore_v3', capabilities=['write']).is_enabled()
      blob_warning = bool(blobstore.BlobInfo.all().count(1))
      template_params = {
          'run_as_a_service': handler.request.get('run_as_a_service'),
          'hide_blobstore': True,
          'form_target': DoBackupHandler.SUFFIX,
          'kind_list': kinds,
          'remainder': remainder,
          'sizes_known': sizes_known,
          'size_total': size_total,
          'queues': None,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'namespaces': get_namespaces(handler.request.get('namespace', None)),
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'notreadonly_warning': notreadonly_warning,
          'blob_warning': blob_warning,
          'backup_name': 'datastore_backup_%s' % time.strftime('%Y_%m_%d')
      }
      utils.RenderToResponse(handler, 'confirm_backup.html', template_params)
  
  
  def get_namespaces(selected_namespace):
    namespaces = [('--All--', '*', selected_namespace is None)]
  
  
  
    for ns in datastore.Query('__namespace__', keys_only=True).Run(limit=250):
      ns_name = ns.name() or ''
      namespaces.append((ns_name or '--Default--',
                         ns_name,
                         ns_name == selected_namespace))
    return namespaces
  
  
  class ConfirmDeleteBackupHandler(webapp.RequestHandler):
    """Handler to confirm admin console requests to delete a backup copy."""
  
    SUFFIX = 'confirm_delete_backup'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      requested_backup_ids = handler.request.get_all('backup_id')
      backups = []
      gs_warning = False
      if requested_backup_ids:
        for backup in db.get(requested_backup_ids):
          if backup:
            backups.append(backup)
            gs_warning |= backup.filesystem == FILES_API_GS_FILESYSTEM
      template_params = {
          'form_target': DoBackupDeleteHandler.SUFFIX,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'backups': backups,
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'gs_warning': gs_warning,
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_delete_backup.html',
                             template_params)
  
  
  class ConfirmAbortBackupHandler(webapp.RequestHandler):
    """Handler to confirm admin console requests to abort a backup copy."""
  
    SUFFIX = 'confirm_abort_backup'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      requested_backup_ids = handler.request.get_all('backup_id')
      backups = []
      if requested_backup_ids:
        for backup in db.get(requested_backup_ids):
          if backup:
            backups.append(backup)
      template_params = {
          'form_target': DoBackupAbortHandler.SUFFIX,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'backups': backups,
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_abort_backup.html',
                             template_params)
  
  
  class ConfirmRestoreFromBackupHandler(webapp.RequestHandler):
    """Handler to confirm admin console requests to restore from backup."""
  
    SUFFIX = 'confirm_restore_from_backup'
  
    @classmethod
    def Render(cls, handler, default_backup_id=None,
               default_delete_backup_after_restore=False):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
        default_backup_id: default value for handler.request
        default_delete_backup_after_restore: default value for handler.request
      """
      backup_id = handler.request.get('backup_id', default_backup_id)
      backup = db.get(backup_id) if backup_id else None
      notreadonly_warning = capabilities.CapabilitySet(
          'datastore_v3', capabilities=['write']).is_enabled()
      original_app_warning = backup.original_app
      if os.getenv('APPLICATION_ID') == original_app_warning:
        original_app_warning = None
      template_params = {
          'form_target': DoBackupRestoreHandler.SUFFIX,
          'queues': None,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'backup': backup,
          'delete_backup_after_restore': handler.request.get(
              'delete_backup_after_restore', default_delete_backup_after_restore),
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'notreadonly_warning': notreadonly_warning,
          'original_app_warning': original_app_warning,
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_restore_from_backup.html',
                             template_params)
  
  
  class ConfirmBackupImportHandler(webapp.RequestHandler):
    """Handler to import backup information."""
  
    SUFFIX = 'backup_information'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      gs_handle = handler.request.get('gs_handle')
      error = None if gs_handle else 'Google Cloud Storage path is missing'
      other_backup_info_files = []
      selected_backup_info_file = None
      backup_info_specified = False
      if not error:
        try:
          gs_handle = gs_handle.rstrip()
          bucket_name, prefix = parse_gs_handle(gs_handle)
          validate_gcs_bucket_name(bucket_name)
          if not is_accessible_bucket_name(bucket_name):
            raise BackupValidationError(
                'Bucket "%s" is not accessible' % bucket_name)
          if prefix.endswith('.backup_info'):
            prefix = prefix[0:prefix.rfind('/')]
            backup_info_specified = True
          elif prefix and not prefix.endswith('/'):
            prefix += '/'
          for backup_info_file in list_bucket_files(bucket_name, prefix):
            backup_info_path = '/gs/%s/%s' % (bucket_name, backup_info_file)
            if backup_info_specified and backup_info_path == gs_handle:
              selected_backup_info_file = backup_info_path
            elif (backup_info_file.endswith('.backup_info')
                  and backup_info_file.count('.') == 1):
              other_backup_info_files.append(backup_info_path)
        except Exception, ex:
          error = 'Failed to read bucket: %s' % ex.message
          logging.exception(ex.message)
      template_params = {
          'error': error,
          'form_target': DoBackupImportHandler.SUFFIX,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'selected_backup_info_file': selected_backup_info_file,
          'other_backup_info_files': other_backup_info_files,
          'backup_info_specified': backup_info_specified,
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_backup_import.html',
                             template_params)
  
  
  class BackupInformationHandler(webapp.RequestHandler):
    """Handler to display backup information."""
  
    SUFFIX = 'backup_information'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      backup_ids = handler.request.get_all('backup_id')
      template_params = {
          'backups': db.get(backup_ids),
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'backup_information.html', template_params)
  
  
  class BaseLinkHandler(webapp.RequestHandler):
    """Base handler that allows only access from cron or taskqueue."""
  
    def get(self):
      """Handler for get requests."""
      self.post()
  
    def post(self):
      """Handler for post requests with access control."""
  
  
  
  
      if ('X-AppEngine-TaskName' not in self.request.headers and
          'X-AppEngine-Cron' not in self.request.headers):
        logging.error('%s must be started via task queue or cron.',
                      self.request.url)
        self.response.set_status(403)
        return
  
      try:
        self._ProcessPostRequest()
      except Exception, e:
        logging.error('Could not perform %s: %s', self.request.url, e.message)
        self.response.set_status(400, e.message)
  
    def _ProcessPostRequest(self):
      """Process the HTTP/POST request and return the result as parameters."""
      raise NotImplementedError
  
  
  class BaseDoHandler(webapp.RequestHandler):
    """Base class for all Do*Handlers."""
  
    MAPREDUCE_DETAIL = config.MAPREDUCE_PATH + '/detail?mapreduce_id='
  
    def get(self):
      """Handler for get requests to datastore_admin backup operations.
  
      Status of executed jobs is displayed.
      """
      jobs = self.request.get_all('job')
      remote_job = self.request.get('remote_job')
      tasks = self.request.get_all('task')
      error = self.request.get('error', '')
      xsrf_error = self.request.get('xsrf_error', '')
      template_params = {
          'job_list': jobs,
          'remote_job': remote_job,
          'task_list': tasks,
          'mapreduce_detail': self.MAPREDUCE_DETAIL,
          'error': error,
          'xsrf_error': xsrf_error,
          'datastore_admin_home': utils.GenerateHomeUrl(self.request),
      }
      utils.RenderToResponse(self, self._get_html_page, template_params)
  
    @property
    def _get_html_page(self):
      """Return the name of the HTML page for HTTP/GET requests."""
      raise NotImplementedError
  
    @property
    def _get_post_html_page(self):
      """Return the name of the HTML page for HTTP/POST requests."""
      raise NotImplementedError
  
    def _ProcessPostRequest(self):
      """Process the HTTP/POST request and return the result as parametrs."""
      raise NotImplementedError
  
    def SendRedirect(self, path=None, params=()):
      """Send a redirect response."""
  
      run_as_a_service = self.request.get('run_as_a_service')
      if run_as_a_service:
        params = list(params)
        params.append(('run_as_a_service', True))
      dest = config.BASE_PATH
      if path:
        dest = '%s/%s' % (dest, path)
      if params:
        dest = '%s?%s' % (dest, urllib.urlencode(params))
      self.redirect(dest)
  
    def post(self):
      """Handler for post requests to datastore_admin/backup.do.
  
      Redirects to the get handler after processing the request.
      """
      token = self.request.get('xsrf_token')
  
      if not utils.ValidateXsrfToken(token, XSRF_ACTION):
        parameters = [('xsrf_error', '1')]
      else:
        try:
          parameters = self._ProcessPostRequest()
  
  
        except Exception, e:
          error = self._HandleException(e)
          parameters = [('error', error)]
  
      self.SendRedirect(self._get_post_html_page, parameters)
  
    def _HandleException(self, e):
      """Make exception handling overridable by tests.
  
      Args:
        e: The exception to handle.
  
      Returns:
        The exception error string.
      """
      logging.exception(e.message)
      return '%s: %s' % (type(e), e.message)
  
  
  class BackupValidationError(utils.Error):
    """Raised upon backup request validation."""
  
  
  def _perform_backup(run_as_a_service, kinds, selected_namespace,
                      filesystem, gcs_path_prefix, backup,
                      queue, mapper_params, max_jobs):
    """Triggers backup mapper jobs.
  
    Args:
      run_as_a_service: True if backup should be done via admin-jobs
      kinds: a sequence of kind names
      selected_namespace: The selected namespace or None for all
      filesystem: FILES_API_GS_FILESYSTEM for GCS (the only possible option).
      gcs_path_prefix: a path (prefix) in which to store the backup when using the
          GS file system. Can be a bucket or a bucket/dir1/dir2 path; ignored if
          filesystem is not GS.
      backup: the backup name
      queue: the task queue for the backup task
      mapper_params: the mapper parameters
      max_jobs: if backup needs more jobs than this, defer them
  
    Returns:
      The job or task ids.
  
    Raises:
      BackupValidationError: On validation error.
      Exception: On other error.
    """
    BACKUP_COMPLETE_HANDLER = __name__ + '.BackupCompleteHandler'
    BACKUP_HANDLER = __name__ + '.BackupEntity.map'
    INPUT_READER = __name__ + '.DatastoreEntityProtoInputReader'
  
    if run_as_a_service:
  
      raise BackupValidationError(
          'Please use the Managed Import/Export service, or remove '
          'run_as_a_service=T. '
          'https://cloud.google.com/datastore/docs/export-import-entities')
  
    queue = queue or os.environ.get('HTTP_X_APPENGINE_QUEUENAME', 'default')
    if queue[0] == '_':
  
      queue = 'default'
  
    backup_info = None
    job_operation = None
  
    job_name = 'datastore_backup_%s_%%(kind)s' % re.sub(r'[^\w]', '_', backup)
    try:
      job_operation = utils.StartOperation('Backup: %s' % backup)
      backup_info = BackupInformation(parent=job_operation)
      backup_info.filesystem = filesystem
      backup_info.name = backup
      backup_info.kinds = kinds
      if selected_namespace is not None:
        backup_info.namespaces = [selected_namespace]
      backup_info.put(force_writes=True)
      mapreduce_params = {
          'done_callback_handler': BACKUP_COMPLETE_HANDLER,
          'backup_info_pk': str(backup_info.key()),
          'force_ops_writes': True,
      }
      mapper_params = dict(mapper_params)
      mapper_params['filesystem'] = filesystem
      if filesystem != FILES_API_GS_FILESYSTEM:
        raise BackupValidationError('Unknown filesystem "%s".' % filesystem)
      gcs_output_writer = (output_writers.__name__ +
                           '.GoogleCloudStorageConsistentRecordOutputWriter')
  
      if not gcs_path_prefix:
        raise BackupValidationError('GCS path missing.')
      bucket_name, path_prefix = validate_and_split_gcs_path(
          gcs_path_prefix, mapper_params['account_id'])
      mapper_params['gs_bucket_name'] = (
          '%s/%s' % (bucket_name, path_prefix)).rstrip('/')
      naming_format = '$name/$id/output-$num'
      if path_prefix:
        naming_format = '%s/%s' % (path_prefix, naming_format)
  
  
  
  
  
  
  
  
  
  
  
  
  
      mapper_params['output_writer'] = copy.copy(mapper_params)
      mapper_params['output_writer'].update({
          'bucket_name': bucket_name,
          'naming_format': naming_format,
      })
  
      if len(kinds) <= max_jobs:
        return [('job', job) for job in _run_map_jobs(
            job_operation.key(), backup_info.key(), kinds, job_name,
            BACKUP_HANDLER, INPUT_READER, gcs_output_writer,
            mapper_params, mapreduce_params, queue)]
      else:
        retry_options = taskqueue.TaskRetryOptions(task_retry_limit=1)
        deferred_task = deferred.defer(_run_map_jobs_deferred,
                                       backup, job_operation.key(),
                                       backup_info.key(), kinds, job_name,
                                       BACKUP_HANDLER, INPUT_READER,
                                       gcs_output_writer, mapper_params,
                                       mapreduce_params, queue, _queue=queue,
                                       _url=config.DEFERRED_PATH,
                                       _retry_options=retry_options)
        return [('task', deferred_task.name)]
    except Exception:
      logging.exception('Failed to start a datastore backup job[s] for "%s".',
                        backup)
      if backup_info:
        delete_backup_info(backup_info)
      if job_operation:
        job_operation.status = utils.DatastoreAdminOperation.STATUS_FAILED
        job_operation.put(force_writes=True)
      raise
  
  
  class BackupLinkHandler(BaseLinkHandler):
    """Handler to deal with requests to the backup link to backup data."""
  
    SUFFIX = 'backup.create'
  
    def _ProcessPostRequest(self):
      """Handler for post requests to datastore_admin/backup.create."""
      if self.request.get('filesystem') != FILES_API_GS_FILESYSTEM:
        self.errorResponse(BLOBSTORE_BACKUP_DISABLED_ERROR_MSG)
        return
  
      backup_prefix = self.request.get('name')
      if not backup_prefix:
        if self.request.headers.get('X-AppEngine-Cron'):
          backup_prefix = 'cron-'
        else:
          backup_prefix = 'link-'
      backup_prefix_with_date = backup_prefix + time.strftime('%Y_%m_%d')
      backup_name = backup_prefix_with_date
      backup_suffix_counter = 1
      while BackupInformation.name_exists(backup_name):
        backup_suffix_counter += 1
        backup_name = backup_prefix_with_date + '-' + str(backup_suffix_counter)
      kinds = self.request.get_all('kind')
      is_all_kinds = self.request.get('all_kinds', False)
      if kinds and is_all_kinds:
        self.errorResponse('"kind" and "all_kinds" must not be both specified.')
        return
      if is_all_kinds:
        kinds, more_kinds = utils.GetKinds()
        if more_kinds:
          self.errorResponse(
              'There are too many kinds (more than %d). Backing up all kinds is'
              ' disabled.' % len(kinds))
          return
      if not kinds:
        self.errorResponse('Backup must include at least one kind.')
        return
      for kind in kinds:
        if not utils.IsKindNameVisible(kind):
          self.errorResponse('Invalid kind %s.' % kind)
          return
      namespace = self.request.get('namespace', None)
      if namespace == '*':
        namespace = None
      mapper_params = {'namespace': namespace}
      _perform_backup(self.request.get('run_as_a_service', False),
                      kinds,
                      namespace,
                      self.request.get('filesystem'),
                      _get_gcs_path_prefix_from_params_dict(self.request),
                      backup_name,
                      self.request.get('queue'),
                      mapper_params,
                      1000000)
  
  
  class DatastoreEntityProtoInputReader(input_readers.RawDatastoreInputReader):
    """An input reader which yields datastore entity proto for a kind."""
  
    _KEY_RANGE_ITER_CLS = db_iters.KeyRangeEntityProtoIterator
  
  
  class DoBackupHandler(BaseDoHandler):
    """Handler to deal with requests from the admin console to backup data."""
  
    SUFFIX = 'backup.do'
    _get_html_page = 'do_backup.html'
    _get_post_html_page = SUFFIX
  
    def _ProcessPostRequest(self):
      """Triggers backup mapper jobs and returns their ids."""
      try:
        if self.request.get('filesystem') != FILES_API_GS_FILESYSTEM:
          raise BackupValidationError(BLOBSTORE_BACKUP_DISABLED_ERROR_MSG)
  
        backup = self.request.get('backup_name').strip()
        if not backup:
          raise BackupValidationError('Unspecified backup name.')
        if BackupInformation.name_exists(backup):
          raise BackupValidationError('Backup "%s" already exists.' % backup)
        mapper_params = _get_basic_mapper_params(self)
  
        # AppScale: Use custom service account if specified.
        account_id = self.request.get('service_account_name', None)
        mapper_params['account_id'] = account_id
        mapper_params['tmp_account_id'] = account_id
  
        backup_result = _perform_backup(
            self.request.get('run_as_a_service', False),
            self.request.get_all('kind'),
            mapper_params.get('namespace'),
            self.request.get('filesystem'),
            _get_gcs_path_prefix_from_params_dict(self.request),
            backup,
            self.request.get('queue'),
            mapper_params,
            10)
        return backup_result
      except Exception, e:
        logging.exception(e.message)
        return [('error', e.message)]
  
  
  def _run_map_jobs_deferred(backup_name, job_operation_key, backup_info_key,
                             kinds, job_name, backup_handler, input_reader,
                             output_writer, mapper_params, mapreduce_params,
                             queue):
    backup_info = BackupInformation.get(backup_info_key)
    if backup_info:
      try:
        _run_map_jobs(job_operation_key, backup_info_key, kinds, job_name,
                      backup_handler, input_reader, output_writer, mapper_params,
                      mapreduce_params, queue)
      except BaseException:
        logging.exception('Failed to start a datastore backup job[s] for "%s".',
                          backup_name)
        delete_backup_info(backup_info)
    else:
      logging.info('Missing backup info, can not start backup jobs for "%s"',
                   backup_name)
  
  
  def _run_map_jobs(job_operation_key, backup_info_key, kinds, job_name,
                    backup_handler, input_reader, output_writer, mapper_params,
                    mapreduce_params, queue):
    """Creates backup/restore MR jobs for the given operation.
  
    Args:
      job_operation_key: a key of utils.DatastoreAdminOperation entity.
      backup_info_key: a key of BackupInformation entity.
      kinds: a list of kinds to run the M/R for.
      job_name: the M/R job name prefix.
      backup_handler: M/R job completion handler.
      input_reader: M/R input reader.
      output_writer: M/R output writer.
      mapper_params: custom parameters to pass to mapper.
      mapreduce_params: dictionary parameters relevant to the whole job.
      queue: the name of the queue that will be used by the M/R.
  
    Returns:
      Ids of all started mapper jobs as list of strings.
    """
    backup_info = BackupInformation.get(backup_info_key)
    if not backup_info:
      return []
    jobs = utils.RunMapForKinds(
        job_operation_key,
        kinds,
        job_name,
        backup_handler,
        input_reader,
        output_writer,
        mapper_params,
        mapreduce_params,
        queue_name=queue)
    backup_info.active_jobs = jobs
    backup_info.put(force_writes=True)
    return jobs
  
  
  def get_backup_files(backup_info, selected_kinds=None):
    """Returns the backup filenames for selected kinds or all if None/Empty."""
    if backup_info.blob_files:
  
      return backup_info.blob_files
    else:
      kinds_backup_files = backup_info.get_kind_backup_files(selected_kinds)
      return list(itertools.chain(*(
          kind_backup_files.files for kind_backup_files in kinds_backup_files)))
  
  
  def get_blob_key(fname):
    return blobstore.BlobKey(_MaybeStripBlobstorePrefix(fname))
  
  
  def delete_backup_files(filesystem, backup_files):
    if backup_files:
  
  
  
      if filesystem == FILES_API_BLOBSTORE_FILESYSTEM:
        blob_keys = []
        for fname in backup_files:
          blob_key = get_blob_key(fname)
          if blob_key:
            blob_keys.append(blob_key)
            if len(blob_keys) == MAX_BLOBS_PER_DELETE:
              blobstore_api.delete(blob_keys)
              blob_keys = []
        if blob_keys:
          blobstore_api.delete(blob_keys)
  
  
  def delete_backup_info(backup_info, delete_files=True):
    """Deletes a backup including its associated files and other metadata."""
    if backup_info.blob_files:
      delete_backup_files(backup_info.filesystem, backup_info.blob_files)
      backup_info.delete(force_writes=True)
    else:
      kinds_backup_files = tuple(backup_info.get_kind_backup_files())
      if delete_files:
        delete_backup_files(backup_info.filesystem, itertools.chain(*(
            kind_backup_files.files for kind_backup_files in kinds_backup_files)))
      db.delete(kinds_backup_files + (backup_info,), force_writes=True)
  
  
  class DoBackupDeleteHandler(BaseDoHandler):
    """Handler to deal with datastore admin requests to delete backup data."""
  
    SUFFIX = 'backup_delete.do'
  
    def get(self):
      self.post()
  
    def post(self):
      """Handler for post requests to datastore_admin/backup_delete.do.
  
      Deletes are executed and user is redirected to the base-path handler.
      """
      backup_ids = self.request.get_all('backup_id')
      token = self.request
  
  [... the rest is omitted ...]
files: AppServer/google/appengine/ext/datastore_admin/backup_handler.py
files: #!/usr/bin/env python
  #
  # Copyright 2007 Google Inc.
  #
  # Licensed under the Apache License, Version 2.0 (the "License");
  # you may not use this file except in compliance with the License.
  # You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  #
  
  
  
  
  
  """Handler for data backup operation.
  
  Generic datastore admin console transfers control to ConfirmBackupHandler
  after selection of entities. The ConfirmBackupHandler confirms with user
  his choice, enters a backup name and transfers control to
  DoBackupHandler. DoBackupHandler starts backup mappers and displays confirmation
  page.
  
  This module also contains actual mapper code for backing data over.
  """
  
  
  
  
  
  
  
  
  from __future__ import with_statement
  
  
  
  import copy
  import cStringIO
  import datetime
  import itertools
  import logging
  import os
  import random
  import re
  import time
  import urllib
  import xml.dom.minidom
  
  from google.appengine._internal import cloudstorage
  
  # AppScale: Use bundled mapreduce library.
  from google.appengine.ext.mapreduce import context
  from google.appengine.ext.mapreduce import datastore_range_iterators as db_iters
  from google.appengine.ext.mapreduce import input_readers
  from google.appengine.ext.mapreduce import json_util
  from google.appengine.ext.mapreduce import operation as op
  from google.appengine.ext.mapreduce import output_writers
  from google.appengine.ext.mapreduce import parameters
  from google.appengine.ext.mapreduce import records
  
  from google.appengine.api import apiproxy_stub_map
  from google.appengine.api import app_identity
  from google.appengine.api import blobstore as blobstore_api
  from google.appengine.api import capabilities
  from google.appengine.api import datastore
  from google.appengine.api import datastore_errors
  from google.appengine.api import datastore_types
  from google.appengine.api import taskqueue
  from google.appengine.api import urlfetch
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.ext import blobstore
  from google.appengine.ext import db
  from google.appengine.ext import deferred
  import webapp2 as webapp
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext.datastore_admin import backup_pb2
  from google.appengine.ext.datastore_admin import config
  from google.appengine.ext.datastore_admin import utils
  
  
  XSRF_ACTION = 'backup'
  BUCKET_PATTERN = (r'^([a-zA-Z0-9]+([\-_]+[a-zA-Z0-9]+)*)'
                    r'(\.([a-zA-Z0-9]+([\-_]+[a-zA-Z0-9]+)*))*$')
  MAX_BUCKET_LEN = 222
  MIN_BUCKET_LEN = 3
  MAX_BUCKET_SEGMENT_LEN = 63
  NUM_KINDS_DEFERRED_THRESHOLD = 10
  MAX_BLOBS_PER_DELETE = 500
  TEST_WRITE_FILENAME_PREFIX = 'datastore_backup_write_test'
  MAX_KEYS_LIST_SIZE = 100
  MAX_TEST_FILENAME_TRIES = 10
  
  MEANING_TO_PRIMITIVE_TYPE = {
      entity_pb.Property.GD_WHEN: backup_pb2.EntitySchema.DATE_TIME,
      entity_pb.Property.GD_RATING: backup_pb2.EntitySchema.RATING,
      entity_pb.Property.ATOM_LINK: backup_pb2.EntitySchema.LINK,
      entity_pb.Property.ATOM_CATEGORY: backup_pb2.EntitySchema.CATEGORY,
      entity_pb.Property.GD_PHONENUMBER: backup_pb2.EntitySchema.PHONE_NUMBER,
      entity_pb.Property.GD_POSTALADDRESS: backup_pb2.EntitySchema.POSTAL_ADDRESS,
      entity_pb.Property.GD_EMAIL: backup_pb2.EntitySchema.EMAIL,
      entity_pb.Property.GD_IM: backup_pb2.EntitySchema.IM_HANDLE,
      entity_pb.Property.BLOBKEY: backup_pb2.EntitySchema.BLOB_KEY,
      entity_pb.Property.TEXT: backup_pb2.EntitySchema.TEXT,
      entity_pb.Property.BLOB: backup_pb2.EntitySchema.BLOB,
      entity_pb.Property.BYTESTRING: backup_pb2.EntitySchema.SHORT_BLOB
  }
  
  
  
  
  FILES_API_GS_FILESYSTEM = 'gs'
  FILES_API_BLOBSTORE_FILESYSTEM = 'blobstore'
  FILES_API_BLOBSTORE_PREFIX = '/blobstore/'
  
  BLOBSTORE_BACKUP_DISABLED_FEATURE_NAME = 'DisableBlobstoreBackups'
  
  BLOBSTORE_BACKUP_DISABLED_ERROR_MSG = (
      'Backups to blobstore are disabled, see '
      'https://cloud.google.com/appengine/docs/deprecations/blobstore_backups')
  
  BACKUP_RESTORE_HANDLER = __name__ + '.RestoreEntity.map'
  
  
  def _get_gcs_path_prefix_from_params_dict(params):
    """Returs the gcs_path_prefix from request or mapreduce dict.
  
    Args:
      params: A dict of parameters.
  
    Returns:
      A string containing GCS prefix for the backup. It can be either just the
      bucket or a bucket with a path (as present in the dictionary). Return None
      if 'gs_bucket_name' is not present in the dictionary.
    """
  
  
  
  
  
    return params.get('gs_bucket_name')
  
  
  def _get_basic_mapper_params(handler):
    namespace = handler.request.get('namespace', None)
    if namespace == '*':
      namespace = None
    return {'namespace': namespace}
  
  
  class GCSUtil(object):
    """Wrappers for cloudstorage methods to deal with /gs/ prefix.
  
    Files API requires GCS files to have a /gs/ prefix. cloudstorage treats /gs/
    as a regular bucket name. This class transparently strips the /gs prefix
    before passing the filename to cloudstorage. Also, for listbucket it
    adds the /gs/ back.
  
    backup_handler should not use cloudstorage directly - all access should
    be proxied by GCSUtil.
    """
  
    @classmethod
    def _strip_gs_prefix_if_present(cls, name):
      if name.startswith('/gs/'):
        return name[3:]
      return name
  
    @classmethod
    def add_gs_prefix_if_missing(cls, name):
      if not name.startswith('/gs/'):
        return '/gs%s' % name
      return name
  
    @classmethod
    def open(cls, filename, *args, **kwargs):
      return cloudstorage.open(
          cls._strip_gs_prefix_if_present(filename), *args, **kwargs)
  
    @classmethod
    def listbucket(cls, prefix, *args, **kwargs):
      output = []
      for f in cloudstorage.listbucket(cls._strip_gs_prefix_if_present(prefix),
                                       *args, **kwargs):
        f.filename = cls.add_gs_prefix_if_missing(f.filename)
        output.append(f)
      return output
  
    @classmethod
    def delete(cls, filename, *args, **kwargs):
      return cloudstorage.delete(
          cls._strip_gs_prefix_if_present(filename), *args, **kwargs)
  
  
  def _MaybeStripBlobstorePrefix(blob_key):
    """Strip /blobstore/ so that this reader can accept files API paths."""
    if blob_key.startswith(FILES_API_BLOBSTORE_PREFIX):
      blob_key = blob_key[len(FILES_API_BLOBSTORE_PREFIX):]
    return blob_key
  
  
  class BlobstoreReaderParamsException(Exception):
    pass
  
  
  class BlobstoreRecordsReader(input_readers.InputReader):
    """Reads records from blobstore without files API dependency.
  
    This reader accesses blobstore using (read-only) blobstore API. It only allows
    reading LevelDB records.
  
    On to_json() we record how many bytes we've read and on from_json() we
    seek to the previous location.
  
    Files API prepends blobstore keys with /blobstore/. We need to accept same
    params as files API so we silently strip /blobstore/ from any path.
    """
  
  
    BLOB_KEYS_PARAM = 'files'
  
    def __init__(self, blob_key, start_position):
      self._blob_key = _MaybeStripBlobstorePrefix(blob_key)
      self._records_reader = records.RecordsReader(
          blobstore.BlobReader(self._blob_key))
      self._records_reader.seek(start_position)
  
    @classmethod
    def _get_params(cls, mapper_spec):
  
      return input_readers._get_params(mapper_spec)
  
    @classmethod
    @db.non_transactional(allow_existing=True)
    def validate(cls, mapper_spec):
      params = cls._get_params(mapper_spec)
      if cls.BLOB_KEYS_PARAM not in params or not params[cls.BLOB_KEYS_PARAM]:
        raise BlobstoreReaderParamsException
  
    def __iter__(self):
      for record in self._records_reader:
        yield record
  
    def to_json(self):
      return self._blob_key, self._records_reader.tell()
  
    @classmethod
    def from_json(cls, json):
      return cls(*json)
  
    @classmethod
    def split_input(cls, mapper_spec):
  
      blob_keys = cls._get_params(mapper_spec)[cls.BLOB_KEYS_PARAM]
      return [cls(blob_key, 0) for blob_key in blob_keys]
  
  
  class ConfirmBackupHandler(webapp.RequestHandler):
    """Handler to deal with requests from the admin console to backup data."""
  
    SUFFIX = 'confirm_backup'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      kinds = handler.request.get_all('kind')
      sizes_known, size_total, remainder = utils.ParseKindsAndSizes(kinds)
      notreadonly_warning = capabilities.CapabilitySet(
          'datastore_v3', capabilities=['write']).is_enabled()
      blob_warning = bool(blobstore.BlobInfo.all().count(1))
      template_params = {
          'run_as_a_service': handler.request.get('run_as_a_service'),
          'hide_blobstore': True,
          'form_target': DoBackupHandler.SUFFIX,
          'kind_list': kinds,
          'remainder': remainder,
          'sizes_known': sizes_known,
          'size_total': size_total,
          'queues': None,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'namespaces': get_namespaces(handler.request.get('namespace', None)),
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'notreadonly_warning': notreadonly_warning,
          'blob_warning': blob_warning,
          'backup_name': 'datastore_backup_%s' % time.strftime('%Y_%m_%d')
      }
      utils.RenderToResponse(handler, 'confirm_backup.html', template_params)
  
  
  def get_namespaces(selected_namespace):
    namespaces = [('--All--', '*', selected_namespace is None)]
  
  
  
    for ns in datastore.Query('__namespace__', keys_only=True).Run(limit=250):
      ns_name = ns.name() or ''
      namespaces.append((ns_name or '--Default--',
                         ns_name,
                         ns_name == selected_namespace))
    return namespaces
  
  
  class ConfirmDeleteBackupHandler(webapp.RequestHandler):
    """Handler to confirm admin console requests to delete a backup copy."""
  
    SUFFIX = 'confirm_delete_backup'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      requested_backup_ids = handler.request.get_all('backup_id')
      backups = []
      gs_warning = False
      if requested_backup_ids:
        for backup in db.get(requested_backup_ids):
          if backup:
            backups.append(backup)
            gs_warning |= backup.filesystem == FILES_API_GS_FILESYSTEM
      template_params = {
          'form_target': DoBackupDeleteHandler.SUFFIX,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'backups': backups,
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'gs_warning': gs_warning,
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_delete_backup.html',
                             template_params)
  
  
  class ConfirmAbortBackupHandler(webapp.RequestHandler):
    """Handler to confirm admin console requests to abort a backup copy."""
  
    SUFFIX = 'confirm_abort_backup'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      requested_backup_ids = handler.request.get_all('backup_id')
      backups = []
      if requested_backup_ids:
        for backup in db.get(requested_backup_ids):
          if backup:
            backups.append(backup)
      template_params = {
          'form_target': DoBackupAbortHandler.SUFFIX,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'backups': backups,
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_abort_backup.html',
                             template_params)
  
  
  class ConfirmRestoreFromBackupHandler(webapp.RequestHandler):
    """Handler to confirm admin console requests to restore from backup."""
  
    SUFFIX = 'confirm_restore_from_backup'
  
    @classmethod
    def Render(cls, handler, default_backup_id=None,
               default_delete_backup_after_restore=False):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
        default_backup_id: default value for handler.request
        default_delete_backup_after_restore: default value for handler.request
      """
      backup_id = handler.request.get('backup_id', default_backup_id)
      backup = db.get(backup_id) if backup_id else None
      notreadonly_warning = capabilities.CapabilitySet(
          'datastore_v3', capabilities=['write']).is_enabled()
      original_app_warning = backup.original_app
      if os.getenv('APPLICATION_ID') == original_app_warning:
        original_app_warning = None
      template_params = {
          'form_target': DoBackupRestoreHandler.SUFFIX,
          'queues': None,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'backup': backup,
          'delete_backup_after_restore': handler.request.get(
              'delete_backup_after_restore', default_delete_backup_after_restore),
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'notreadonly_warning': notreadonly_warning,
          'original_app_warning': original_app_warning,
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_restore_from_backup.html',
                             template_params)
  
  
  class ConfirmBackupImportHandler(webapp.RequestHandler):
    """Handler to import backup information."""
  
    SUFFIX = 'backup_information'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      gs_handle = handler.request.get('gs_handle')
      error = None if gs_handle else 'Google Cloud Storage path is missing'
      other_backup_info_files = []
      selected_backup_info_file = None
      backup_info_specified = False
      if not error:
        try:
          gs_handle = gs_handle.rstrip()
          bucket_name, prefix = parse_gs_handle(gs_handle)
          validate_gcs_bucket_name(bucket_name)
          if not is_accessible_bucket_name(bucket_name):
            raise BackupValidationError(
                'Bucket "%s" is not accessible' % bucket_name)
          if prefix.endswith('.backup_info'):
            prefix = prefix[0:prefix.rfind('/')]
            backup_info_specified = True
          elif prefix and not prefix.endswith('/'):
            prefix += '/'
          for backup_info_file in list_bucket_files(bucket_name, prefix):
            backup_info_path = '/gs/%s/%s' % (bucket_name, backup_info_file)
            if backup_info_specified and backup_info_path == gs_handle:
              selected_backup_info_file = backup_info_path
            elif (backup_info_file.endswith('.backup_info')
                  and backup_info_file.count('.') == 1):
              other_backup_info_files.append(backup_info_path)
        except Exception, ex:
          error = 'Failed to read bucket: %s' % ex.message
          logging.exception(ex.message)
      template_params = {
          'error': error,
          'form_target': DoBackupImportHandler.SUFFIX,
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'selected_backup_info_file': selected_backup_info_file,
          'other_backup_info_files': other_backup_info_files,
          'backup_info_specified': backup_info_specified,
          'xsrf_token': utils.CreateXsrfToken(XSRF_ACTION),
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'confirm_backup_import.html',
                             template_params)
  
  
  class BackupInformationHandler(webapp.RequestHandler):
    """Handler to display backup information."""
  
    SUFFIX = 'backup_information'
  
    @classmethod
    def Render(cls, handler):
      """Rendering method that can be called by main.py.
  
      Args:
        handler: the webapp.RequestHandler invoking the method
      """
      backup_ids = handler.request.get_all('backup_id')
      template_params = {
          'backups': db.get(backup_ids),
          'datastore_admin_home': utils.GenerateHomeUrl(handler.request),
          'run_as_a_service': handler.request.get('run_as_a_service'),
      }
      utils.RenderToResponse(handler, 'backup_information.html', template_params)
  
  
  class BaseLinkHandler(webapp.RequestHandler):
    """Base handler that allows only access from cron or taskqueue."""
  
    def get(self):
      """Handler for get requests."""
      self.post()
  
    def post(self):
      """Handler for post requests with access control."""
  
  
  
  
      if ('X-AppEngine-TaskName' not in self.request.headers and
          'X-AppEngine-Cron' not in self.request.headers):
        logging.error('%s must be started via task queue or cron.',
                      self.request.url)
        self.response.set_status(403)
        return
  
      try:
        self._ProcessPostRequest()
      except Exception, e:
        logging.error('Could not perform %s: %s', self.request.url, e.message)
        self.response.set_status(400, e.message)
  
    def _ProcessPostRequest(self):
      """Process the HTTP/POST request and return the result as parameters."""
      raise NotImplementedError
  
  
  class BaseDoHandler(webapp.RequestHandler):
    """Base class for all Do*Handlers."""
  
    MAPREDUCE_DETAIL = config.MAPREDUCE_PATH + '/detail?mapreduce_id='
  
    def get(self):
      """Handler for get requests to datastore_admin backup operations.
  
      Status of executed jobs is displayed.
      """
      jobs = self.request.get_all('job')
      remote_job = self.request.get('remote_job')
      tasks = self.request.get_all('task')
      error = self.request.get('error', '')
      xsrf_error = self.request.get('xsrf_error', '')
      template_params = {
          'job_list': jobs,
          'remote_job': remote_job,
          'task_list': tasks,
          'mapreduce_detail': self.MAPREDUCE_DETAIL,
          'error': error,
          'xsrf_error': xsrf_error,
          'datastore_admin_home': utils.GenerateHomeUrl(self.request),
      }
      utils.RenderToResponse(self, self._get_html_page, template_params)
  
    @property
    def _get_html_page(self):
      """Return the name of the HTML page for HTTP/GET requests."""
      raise NotImplementedError
  
    @property
    def _get_post_html_page(self):
      """Return the name of the HTML page for HTTP/POST requests."""
      raise NotImplementedError
  
    def _ProcessPostRequest(self):
      """Process the HTTP/POST request and return the result as parametrs."""
      raise NotImplementedError
  
    def SendRedirect(self, path=None, params=()):
      """Send a redirect response."""
  
      run_as_a_service = self.request.get('run_as_a_service')
      if run_as_a_service:
        params = list(params)
        params.append(('run_as_a_service', True))
      dest = config.BASE_PATH
      if path:
        dest = '%s/%s' % (dest, path)
      if params:
        dest = '%s?%s' % (dest, urllib.urlencode(params))
      self.redirect(dest)
  
    def post(self):
      """Handler for post requests to datastore_admin/backup.do.
  
      Redirects to the get handler after processing the request.
      """
      token = self.request.get('xsrf_token')
  
      if not utils.ValidateXsrfToken(token, XSRF_ACTION):
        parameters = [('xsrf_error', '1')]
      else:
        try:
          parameters = self._ProcessPostRequest()
  
  
        except Exception, e:
          error = self._HandleException(e)
          parameters = [('error', error)]
  
      self.SendRedirect(self._get_post_html_page, parameters)
  
    def _HandleException(self, e):
      """Make exception handling overridable by tests.
  
      Args:
        e: The exception to handle.
  
      Returns:
        The exception error string.
      """
      logging.exception(e.message)
      return '%s: %s' % (type(e), e.message)
  
  
  class BackupValidationError(utils.Error):
    """Raised upon backup request validation."""
  
  
  def _perform_backup(run_as_a_service, kinds, selected_namespace,
                      filesystem, gcs_path_prefix, backup,
                      queue, mapper_params, max_jobs):
    """Triggers backup mapper jobs.
  
    Args:
      run_as_a_service: True if backup should be done via admin-jobs
      kinds: a sequence of kind names
      selected_namespace: The selected namespace or None for all
      filesystem: FILES_API_GS_FILESYSTEM for GCS (the only possible option).
      gcs_path_prefix: a path (prefix) in which to store the backup when using the
          GS file system. Can be a bucket or a bucket/dir1/dir2 path; ignored if
          filesystem is not GS.
      backup: the backup name
      queue: the task queue for the backup task
      mapper_params: the mapper parameters
      max_jobs: if backup needs more jobs than this, defer them
  
    Returns:
      The job or task ids.
  
    Raises:
      BackupValidationError: On validation error.
      Exception: On other error.
    """
    BACKUP_COMPLETE_HANDLER = __name__ + '.BackupCompleteHandler'
    BACKUP_HANDLER = __name__ + '.BackupEntity.map'
    INPUT_READER = __name__ + '.DatastoreEntityProtoInputReader'
  
    if run_as_a_service:
  
      raise BackupValidationError(
          'Please use the Managed Import/Export service, or remove '
          'run_as_a_service=T. '
          'https://cloud.google.com/datastore/docs/export-import-entities')
  
    queue = queue or os.environ.get('HTTP_X_APPENGINE_QUEUENAME', 'default')
    if queue[0] == '_':
  
      queue = 'default'
  
    backup_info = None
    job_operation = None
  
    job_name = 'datastore_backup_%s_%%(kind)s' % re.sub(r'[^\w]', '_', backup)
    try:
      job_operation = utils.StartOperation('Backup: %s' % backup)
      backup_info = BackupInformation(parent=job_operation)
      backup_info.filesystem = filesystem
      backup_info.name = backup
      backup_info.kinds = kinds
      if selected_namespace is not None:
        backup_info.namespaces = [selected_namespace]
      backup_info.put(force_writes=True)
      mapreduce_params = {
          'done_callback_handler': BACKUP_COMPLETE_HANDLER,
          'backup_info_pk': str(backup_info.key()),
          'force_ops_writes': True,
      }
      mapper_params = dict(mapper_params)
      mapper_params['filesystem'] = filesystem
      if filesystem != FILES_API_GS_FILESYSTEM:
        raise BackupValidationError('Unknown filesystem "%s".' % filesystem)
      gcs_output_writer = (output_writers.__name__ +
                           '.GoogleCloudStorageConsistentRecordOutputWriter')
  
      if not gcs_path_prefix:
        raise BackupValidationError('GCS path missing.')
      bucket_name, path_prefix = validate_and_split_gcs_path(gcs_path_prefix)
      mapper_params['gs_bucket_name'] = (
          '%s/%s' % (bucket_name, path_prefix)).rstrip('/')
      naming_format = '$name/$id/output-$num'
      if path_prefix:
        naming_format = '%s/%s' % (path_prefix, naming_format)
  
  
  
  
  
  
  
  
  
  
  
  
  
      mapper_params['output_writer'] = copy.copy(mapper_params)
      mapper_params['output_writer'].update({
          'bucket_name': bucket_name,
          'naming_format': naming_format,
      })
  
      if len(kinds) <= max_jobs:
        return [('job', job) for job in _run_map_jobs(
            job_operation.key(), backup_info.key(), kinds, job_name,
            BACKUP_HANDLER, INPUT_READER, gcs_output_writer,
            mapper_params, mapreduce_params, queue)]
      else:
        retry_options = taskqueue.TaskRetryOptions(task_retry_limit=1)
        deferred_task = deferred.defer(_run_map_jobs_deferred,
                                       backup, job_operation.key(),
                                       backup_info.key(), kinds, job_name,
                                       BACKUP_HANDLER, INPUT_READER,
                                       gcs_output_writer, mapper_params,
                                       mapreduce_params, queue, _queue=queue,
                                       _url=config.DEFERRED_PATH,
                                       _retry_options=retry_options)
        return [('task', deferred_task.name)]
    except Exception:
      logging.exception('Failed to start a datastore backup job[s] for "%s".',
                        backup)
      if backup_info:
        delete_backup_info(backup_info)
      if job_operation:
        job_operation.status = utils.DatastoreAdminOperation.STATUS_FAILED
        job_operation.put(force_writes=True)
      raise
  
  
  class BackupLinkHandler(BaseLinkHandler):
    """Handler to deal with requests to the backup link to backup data."""
  
    SUFFIX = 'backup.create'
  
    def _ProcessPostRequest(self):
      """Handler for post requests to datastore_admin/backup.create."""
      if self.request.get('filesystem') != FILES_API_GS_FILESYSTEM:
        self.errorResponse(BLOBSTORE_BACKUP_DISABLED_ERROR_MSG)
        return
  
      backup_prefix = self.request.get('name')
      if not backup_prefix:
        if self.request.headers.get('X-AppEngine-Cron'):
          backup_prefix = 'cron-'
        else:
          backup_prefix = 'link-'
      backup_prefix_with_date = backup_prefix + time.strftime('%Y_%m_%d')
      backup_name = backup_prefix_with_date
      backup_suffix_counter = 1
      while BackupInformation.name_exists(backup_name):
        backup_suffix_counter += 1
        backup_name = backup_prefix_with_date + '-' + str(backup_suffix_counter)
      kinds = self.request.get_all('kind')
      is_all_kinds = self.request.get('all_kinds', False)
      if kinds and is_all_kinds:
        self.errorResponse('"kind" and "all_kinds" must not be both specified.')
        return
      if is_all_kinds:
        kinds, more_kinds = utils.GetKinds()
        if more_kinds:
          self.errorResponse(
              'There are too many kinds (more than %d). Backing up all kinds is'
              ' disabled.' % len(kinds))
          return
      if not kinds:
        self.errorResponse('Backup must include at least one kind.')
        return
      for kind in kinds:
        if not utils.IsKindNameVisible(kind):
          self.errorResponse('Invalid kind %s.' % kind)
          return
      namespace = self.request.get('namespace', None)
      if namespace == '*':
        namespace = None
      mapper_params = {'namespace': namespace}
      _perform_backup(self.request.get('run_as_a_service', False),
                      kinds,
                      namespace,
                      self.request.get('filesystem'),
                      _get_gcs_path_prefix_from_params_dict(self.request),
                      backup_name,
                      self.request.get('queue'),
                      mapper_params,
                      1000000)
  
  
  class DatastoreEntityProtoInputReader(input_readers.RawDatastoreInputReader):
    """An input reader which yields datastore entity proto for a kind."""
  
    _KEY_RANGE_ITER_CLS = db_iters.KeyRangeEntityProtoIterator
  
  
  class DoBackupHandler(BaseDoHandler):
    """Handler to deal with requests from the admin console to backup data."""
  
    SUFFIX = 'backup.do'
    _get_html_page = 'do_backup.html'
    _get_post_html_page = SUFFIX
  
    def _ProcessPostRequest(self):
      """Triggers backup mapper jobs and returns their ids."""
      try:
        if self.request.get('filesystem') != FILES_API_GS_FILESYSTEM:
          raise BackupValidationError(BLOBSTORE_BACKUP_DISABLED_ERROR_MSG)
  
        backup = self.request.get('backup_name').strip()
        if not backup:
          raise BackupValidationError('Unspecified backup name.')
        if BackupInformation.name_exists(backup):
          raise BackupValidationError('Backup "%s" already exists.' % backup)
        mapper_params = _get_basic_mapper_params(self)
        backup_result = _perform_backup(
            self.request.get('run_as_a_service', False),
            self.request.get_all('kind'),
            mapper_params.get('namespace'),
            self.request.get('filesystem'),
            _get_gcs_path_prefix_from_params_dict(self.request),
            backup,
            self.request.get('queue'),
            mapper_params,
            10)
        return backup_result
      except Exception, e:
        logging.exception(e.message)
        return [('error', e.message)]
  
  
  def _run_map_jobs_deferred(backup_name, job_operation_key, backup_info_key,
                             kinds, job_name, backup_handler, input_reader,
                             output_writer, mapper_params, mapreduce_params,
                             queue):
    backup_info = BackupInformation.get(backup_info_key)
    if backup_info:
      try:
        _run_map_jobs(job_operation_key, backup_info_key, kinds, job_name,
                      backup_handler, input_reader, output_writer, mapper_params,
                      mapreduce_params, queue)
      except BaseException:
        logging.exception('Failed to start a datastore backup job[s] for "%s".',
                          backup_name)
        delete_backup_info(backup_info)
    else:
      logging.info('Missing backup info, can not start backup jobs for "%s"',
                   backup_name)
  
  
  def _run_map_jobs(job_operation_key, backup_info_key, kinds, job_name,
                    backup_handler, input_reader, output_writer, mapper_params,
                    mapreduce_params, queue):
    """Creates backup/restore MR jobs for the given operation.
  
    Args:
      job_operation_key: a key of utils.DatastoreAdminOperation entity.
      backup_info_key: a key of BackupInformation entity.
      kinds: a list of kinds to run the M/R for.
      job_name: the M/R job name prefix.
      backup_handler: M/R job completion handler.
      input_reader: M/R input reader.
      output_writer: M/R output writer.
      mapper_params: custom parameters to pass to mapper.
      mapreduce_params: dictionary parameters relevant to the whole job.
      queue: the name of the queue that will be used by the M/R.
  
    Returns:
      Ids of all started mapper jobs as list of strings.
    """
    backup_info = BackupInformation.get(backup_info_key)
    if not backup_info:
      return []
    jobs = utils.RunMapForKinds(
        job_operation_key,
        kinds,
        job_name,
        backup_handler,
        input_reader,
        output_writer,
        mapper_params,
        mapreduce_params,
        queue_name=queue)
    backup_info.active_jobs = jobs
    backup_info.put(force_writes=True)
    return jobs
  
  
  def get_backup_files(backup_info, selected_kinds=None):
    """Returns the backup filenames for selected kinds or all if None/Empty."""
    if backup_info.blob_files:
  
      return backup_info.blob_files
    else:
      kinds_backup_files = backup_info.get_kind_backup_files(selected_kinds)
      return list(itertools.chain(*(
          kind_backup_files.files for kind_backup_files in kinds_backup_files)))
  
  
  def get_blob_key(fname):
    return blobstore.BlobKey(_MaybeStripBlobstorePrefix(fname))
  
  
  def delete_backup_files(filesystem, backup_files):
    if backup_files:
  
  
  
      if filesystem == FILES_API_BLOBSTORE_FILESYSTEM:
        blob_keys = []
        for fname in backup_files:
          blob_key = get_blob_key(fname)
          if blob_key:
            blob_keys.append(blob_key)
            if len(blob_keys) == MAX_BLOBS_PER_DELETE:
              blobstore_api.delete(blob_keys)
              blob_keys = []
        if blob_keys:
          blobstore_api.delete(blob_keys)
  
  
  def delete_backup_info(backup_info, delete_files=True):
    """Deletes a backup including its associated files and other metadata."""
    if backup_info.blob_files:
      delete_backup_files(backup_info.filesystem, backup_info.blob_files)
      backup_info.delete(force_writes=True)
    else:
      kinds_backup_files = tuple(backup_info.get_kind_backup_files())
      if delete_files:
        delete_backup_files(backup_info.filesystem, itertools.chain(*(
            kind_backup_files.files for kind_backup_files in kinds_backup_files)))
      db.delete(kinds_backup_files + (backup_info,), force_writes=True)
  
  
  class DoBackupDeleteHandler(BaseDoHandler):
    """Handler to deal with datastore admin requests to delete backup data."""
  
    SUFFIX = 'backup_delete.do'
  
    def get(self):
      self.post()
  
    def post(self):
      """Handler for post requests to datastore_admin/backup_delete.do.
  
      Deletes are executed and user is redirected to the base-path handler.
      """
      backup_ids = self.request.get_all('backup_id')
      token = self.request.get('xsrf_token')
      params = ()
      if backup_ids and utils.ValidateXsrfToken(token, XSRF_ACTION):
        try:
          for backup_info in db.get(backup_ids):
            if backup_info:
              delete_backup_info(backup
  
  [... the rest is omitted ...]
files: AppServer/google/appengine/ext/datastore_admin/backup_handler.py
files: {% extends "base.html" %}
  {% block title %}Confirm Backup of {{kind_str|escape}}{% endblock %}
  {% block body %}
    <h2>Datastore Admin: Backup to Google Cloud Storage</h2>
    {% if kind_list %}
      {% comment %}
      size_total represents the total size of the figures for which we have
      datastore statistics available.
      sizes_known indicates whether or not we have datastore statistics for all
      kinds.
      These two variables are used to render different messages for the three
      possible scenarios: all sizes known, some sizes known, no sizes known.
      {% endcomment %}
  
      <form class="ae-form" action="{{base_path}}/{{form_target}}" method="post"
          style="width:39.39em;">
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        {% for kind in kind_list %}
        <input type="hidden" name="kind" value="{{kind|escape}}">
        {% endfor %}
  
        <div class="ae-input-row">
          <div class="ae-label-row">
            <label for="backup_name">Backup name: </label>
          </div>
          <div class="ae-input-row">
            <input type="text" id="backup_name" name="backup_name"
                style="width:100%"
                value="{{backup_name}}">
          </div>
        </div>
        <input type="hidden" name="xsrf_token" value="{{xsrf_token|escape}}">
  
        {% if notreadonly_warning %}
        <div>
          <h4 style="padding:0;margin-top:1em;margin-bottom:0.46em;">WARNING</h4>
          <p>The application is not in read-only mode and updates
          during the backup process may not be captured.
          To place the application in read-only mode,
          please follow the instructions at
          <a href="https://cloud.google.com/datastore/docs/console/managing-datastore#enabling_datastore_writes" target="_blank">
          Enabling or Disabling Datastore Writes</a>
          </p>
        </div>
        {% endif %}
  
        {% if blob_warning %}
        <p>The application has blob data.  BlobInfo records and data will NOT be
        backed up.</p>
        {% endif %}
  
        <p>Selected namespace:
          <select name="namespace">
            {% for namespace in namespaces %}
              <option value="{{ namespace.1 }}" {% if namespace.2 %}selected="selected"{% endif %}>{{ namespace.0 }}</option>
            {% endfor %}
          </select>
        </p>
  
        {% if not run_as_a_service %}
        <p>Operation will be performed using the selected queue:
          {% if queues %}
          <select name="queue">
            {% for queue in queues %}
              <option value="{{ queue }}">{{ queue }}</option>
            {% endfor %}
          </select>
          {% else %}
          <input type="text" name="queue" value="default"/>
          {% endif %}
        </p>
        {% endif %}
  
        <table>
          <tr>
            <td>
              Backup storage destination:
              <select name="filesystem" id="ae-datastore-admin-filesystem">
                {% if not run_as_a_service and not hide_blobstore %}
                <option value="blobstore">Blobstore</option>
                {% endif %}
                <option value="gs" selected="selected">Google Cloud Storage</option>
              </select>
            </td>
          </tr>
          <tr id="gs_bucket_tr">
            <td>
              Google Cloud Storage bucket name
              (<a target="_blank"
                href="https://cloud.google.com/datastore/docs/console/datastore-backing-up-restoring">More info?</a>):
              <input type="text" id="gs_bucket_name" name="gs_bucket_name" value="" />
            </td>
          </tr>
          <tr id="gs_service_account_tr">
            <td>
              Service account name
              <img class="ae-help-icon" src="{{ base_path }}/static/img/help.gif" height="14" width="14" alt="help"
                   title="Use the client_email field from any custom service accounts you've defined.">
              <input name="service_account_name">
            </td>
          </tr>
        </table>
  
        <table style="margin-top: 1em;"><tr>
          <td style="padding-right: 0.5em;">
            <input class="goog-button" type="submit" name="backup"
                value="Backup Entities"></td>
          <td style="color: #666667;">
            or <a href="{{ datastore_admin_home }}">Cancel</a></td>
        </tr></table>
      </form>
    {% else %}
      <div class="ae-errorbox">No entity kinds to backup.</div>
    {% endif %}
  {% endblock %}
  
  
  [... the rest is omitted ...]
files: AppServer/google/appengine/ext/datastore_admin/templates/confirm_backup.html
files: {% extends "base.html" %}
  {% block title %}Confirm Backup of {{kind_str|escape}}{% endblock %}
  {% block body %}
    <h2>Datastore Admin: Backup to Google Cloud Storage</h2>
    {% if kind_list %}
      {% comment %}
      size_total represents the total size of the figures for which we have
      datastore statistics available.
      sizes_known indicates whether or not we have datastore statistics for all
      kinds.
      These two variables are used to render different messages for the three
      possible scenarios: all sizes known, some sizes known, no sizes known.
      {% endcomment %}
  
      <form class="ae-form" action="{{base_path}}/{{form_target}}" method="post"
          style="width:39.39em;">
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        {% for kind in kind_list %}
        <input type="hidden" name="kind" value="{{kind|escape}}">
        {% endfor %}
  
        <div class="ae-input-row">
          <div class="ae-label-row">
            <label for="backup_name">Backup name: </label>
          </div>
          <div class="ae-input-row">
            <input type="text" id="backup_name" name="backup_name"
                style="width:100%"
                value="{{backup_name}}">
          </div>
        </div>
        <input type="hidden" name="xsrf_token" value="{{xsrf_token|escape}}">
  
        {% if notreadonly_warning %}
        <div>
          <h4 style="padding:0;margin-top:1em;margin-bottom:0.46em;">WARNING</h4>
          <p>The application is not in read-only mode and updates
          during the backup process may not be captured.
          To place the application in read-only mode,
          please follow the instructions at
          <a href="https://cloud.google.com/datastore/docs/console/managing-datastore#enabling_datastore_writes" target="_blank">
          Enabling or Disabling Datastore Writes</a>
          </p>
        </div>
        {% endif %}
  
        {% if blob_warning %}
        <p>The application has blob data.  BlobInfo records and data will NOT be
        backed up.</p>
        {% endif %}
  
        <p>Selected namespace:
          <select name="namespace">
            {% for namespace in namespaces %}
              <option value="{{ namespace.1 }}" {% if namespace.2 %}selected="selected"{% endif %}>{{ namespace.0 }}</option>
            {% endfor %}
          </select>
        </p>
  
        {% if not run_as_a_service %}
        <p>Operation will be performed using the selected queue:
          {% if queues %}
          <select name="queue">
            {% for queue in queues %}
              <option value="{{ queue }}">{{ queue }}</option>
            {% endfor %}
          </select>
          {% else %}
          <input type="text" name="queue" value="default"/>
          {% endif %}
        </p>
        {% endif %}
  
        <table>
          <tr>
            <td>
              Backup storage destination:
              <select name="filesystem" id="ae-datastore-admin-filesystem">
                {% if not run_as_a_service and not hide_blobstore %}
                <option value="blobstore">Blobstore</option>
                {% endif %}
                <option value="gs" selected="selected">Google Cloud Storage</option>
              </select>
            </td>
          </tr>
          <tr id="gs_bucket_tr">
            <td>
              Google Cloud Storage bucket name
              (<a target="_blank"
                href="https://cloud.google.com/datastore/docs/console/datastore-backing-up-restoring">More info?</a>):
              <input type="text" id="gs_bucket_name" name="gs_bucket_name" value="" />
            </td>
          </tr>
        </table>
  
        <table style="margin-top: 1em;"><tr>
          <td style="padding-right: 0.5em;">
            <input class="goog-button" type="submit" name="backup"
                value="Backup Entities"></td>
          <td style="color: #666667;">
            or <a href="{{ datastore_admin_home }}">Cancel</a></td>
        </tr></table>
      </form>
    {% else %}
      <div class="ae-errorbox">No entity kinds to backup.</div>
    {% endif %}
  {% endblock %}
  
  
  [... the rest is omitted ...]
files: AppServer/google/appengine/ext/datastore_admin/templates/confirm_backup.html