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'),
          'service_accounts': utils.get_service_account_names()
      }
      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'),
          'service_accounts': utils.get_service_account_names(),
      }
      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')
  
      # AppScale: Use custom service account if specified.
      account_id = handler.request.get('service_account_name', '')
      if not account_id:
        raise ValueError('Invalid service account name')
  
      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, account_id):
            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,
                                                    account_id=account_id):
            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'),
          'service_account_name': account_id,
      }
      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', '')
        if not account_id:
          raise ValueError('Invalid service account name')
  
        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.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_info)
        except Exception, e:
          logging.exception('Failed to delete datastore backup.')
          params = [('error', e.message)]
  
      self.SendRedirect(params=params)
  
  
  class DoBackupAbortHandler(BaseDoHandler):
    """Handler to deal with datastore admin requests to abort pending backups."""
  
    SUFFIX = 'backup_abort.do'
  
    def get(self):
      self.post()
  
    def post(self):
      """Handler for post requests to datastore_admin/backup_abort.do.
  
      Abort is 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:
              operation = backup_info.parent()
              utils.AbortAdminOperation(operation.key())
              delete_backup_info(backup_info)
        except Exception, e:
          logging.exception('Failed to abort pending datastore backup.')
          params = [('error', e.message)]
  
      self.SendRedirect(params=params)
  
  
  def _restore(backup_id, kinds, run_as_a_service, queue, mapper_params):
    """Restore a backup."""
    if not backup_id:
      raise ValueError('Unspecified Backup.')
  
    backup = db.get(db.Key(backup_id))
    if not backup:
      raise ValueError('Invalid Backup id.')
  
    input_reader_to_use = None
  
    kinds = set(kinds)
    if not (backup.blob_files or kinds):
      raise ValueError('No kinds were selected')
    backup_kinds = set(backup.kinds)
    difference = kinds.difference(backup_kinds)
    if difference:
      raise ValueError('Backup does not have kind[s] %s' % ', '.join(difference))
  
    if run_as_a_service:
      raise ValueError(
          'Please use the Managed Import/Export service or remove '
          'run_as_a_service=T. '
          'https://cloud.google.com/datastore/docs/export-import-entities')
  
    job_name = 'datastore_backup_restore_%s' % re.sub(r'[^\w]', '_',
                                                      backup.name)
    job_operation = None
    try:
      operation_name = 'Restoring %s from backup: %s' % (
          ', '.join(kinds) if kinds else 'all', backup.name)
      job_operation = utils.StartOperation(operation_name)
  
  
      kinds = list(kinds) if len(backup_kinds) != len(kinds) else []
      mapper_params.update({
          'files': get_backup_files(backup, kinds),
          'kind_filter': kinds,
          'original_app': backup.original_app,
  
  
          parameters.DYNAMIC_RATE_INITIAL_QPS_PARAM: 500,
          parameters.DYNAMIC_RATE_BUMP_FACTOR_PARAM: 1.5,
          parameters.DYNAMIC_RATE_BUMP_TIME_PARAM: 300,
      })
  
      if backup.filesystem == FILES_API_GS_FILESYSTEM:
        input_reader_to_use = (input_readers.__name__ +
                               '.GoogleCloudStorageRecordInputReader')
        if not is_readable_gs_handle(backup.gs_handle,
                                     mapper_params.get('account_id')):
          raise ValueError('Backup not readable.')
  
        if not mapper_params['files']:
          raise ValueError('No blob objects in restore.')
        bucket = parse_gs_handle(mapper_params['files'][0])[0]
  
  
  
  
  
  
        mapper_params['input_reader'] = copy.copy(mapper_params)
        mapper_params['input_reader'].update({
  
  
  
  
            'objects': [parse_gs_handle(f)[1] for f in mapper_params['files']],
            'bucket_name': bucket,
        })
      elif backup.filesystem == FILES_API_BLOBSTORE_FILESYSTEM:
        input_reader_to_use = __name__ + '.BlobstoreRecordsReader'
      else:
        raise ValueError('Unknown backup filesystem.')
  
      mapreduce_params = {
          'backup_name': backup.name,
          'force_ops_writes': True,
      }
      shard_count = min(max(utils.MAPREDUCE_MIN_SHARDS,
                            len(mapper_params['files'])),
                        utils.MAPREDUCE_MAX_SHARDS)
      job = utils.StartMap(
          job_operation.key(), job_name, BACKUP_RESTORE_HANDLER,
          input_reader_to_use, None, mapper_params, mapreduce_params,
          queue_name=queue, shard_count=shard_count)
      return ('job', job)
    except ValueError:
      raise
    except Exception:
      logging.exception('Failed to start a restore from backup job "%s".',
                        job_name)
      if job_operation:
        job_operation.status = utils.DatastoreAdminOperation.STATUS_FAILED
        job_operation.put(force_writes=True)
      raise
  
  
  class DoBackupRestoreHandler(BaseDoHandler):
    """Handler to restore backup data.
  
    Deals with requests from the admin console.
    """
    SUFFIX = 'backup_restore.do'
    _get_html_page = 'do_restore_from_backup.html'
    _get_post_html_page = SUFFIX
  
    def _ProcessPostRequest(self):
      """Triggers backup restore mapper jobs and returns their ids."""
      try:
        # AppScale: Use custom service account if specified.
        mapper_params = _get_basic_mapper_params(self)
        account_id = self.request.get('service_account_name', None)
        mapper_params['account_id'] = account_id
        mapper_params['tmp_account_id'] = account_id
        return [
            _restore(
                backup_id=self.request.get('backup_id'),
                kinds=self.request.get_all('kind'),
                run_as_a_service=self.request.get('run_as_a_service', False),
                queue=self.request.get('queue'),
                mapper_params=mapper_params
            )
        ]
      except ValueError, e:
        return [('error', e.message)]
  
  
  def _import_backup(gs_handle, account_id=None):
    """Import backup from `gs_handle` to the current project."""
    bucket_name, path = parse_gs_handle(gs_handle)
    file_content = get_gs_object(bucket_name, path, account_id)
    entities = parse_backup_info_file(file_content)
    original_backup_info = entities.next()
    entity = datastore.Entity(BackupInformation.kind())
    entity.update(original_backup_info)
    backup_info = BackupInformation.from_entity(entity)
    if original_backup_info.key().app() != os.getenv('APPLICATION_ID'):
      backup_info.original_app = original_backup_info.key().app()
  
    def tx():
      backup_info.put(force_writes=True)
      kind_files_models = []
      for entity in entities:
        kind_files = backup_info.create_kind_backup_files(
            entity.key().name(), entity['files'])
        kind_files_models.append(kind_files)
      db.put(kind_files_models, force_writes=True)
    db.run_in_transaction(tx)
    return str(backup_info.key())
  
  
  class BackupImportAndRestoreLinkHandler(BaseLinkHandler):
    """Handler that imports and restores a backup."""
  
    SUFFIX = 'backup_import_restore.create'
  
    def _ProcessPostRequest(self):
      """Handler for post requests to datastore_admin/import_backup.create."""
      account_id = self.request.get('service_account_name', None)
      _restore(
          backup_id=_import_backup(self.request.get('gs_handle'), account_id),
          kinds=self.request.get_all('kind'),
          run_as_a_service=self.request.get('run_as_a_service', False),
          queue=self.request.get('queue'),
          mapper_params=_get_basic_mapper_params(self)
      )
  
  
  class DoBackupImportHandler(BaseDoHandler):
    """Handler to deal with datastore admin requests to import backup info."""
  
    SUFFIX = 'import_backup.do'
  
    def get(self):
      self.post()
  
    def post(self):
      """Handler for post requests to datastore_admin/import_backup.do.
  
      Import is executed and user is redirected to the base-path handler.
      """
      gs_handle = self.request.get('gs_handle')
      account_id = self.request.get('service_account_name', None)
      token = self.request.get('xsrf_token')
      error = None
      if gs_handle and utils.ValidateXsrfToken(token, XSRF_ACTION):
        try:
          backup_id = _import_backup(gs_handle, account_id)
        except Exception, e:
          logging.exception('Failed to Import datastore backup information.')
          error = e.message
  
      if error:
        self.SendRedirect(params=[('error', error)])
      elif self.request.get('Restore'):
        ConfirmRestoreFromBackupHandler.Render(
            self, default_backup_id=backup_id,
            default_delete_backup_after_restore=True)
      else:
        self.SendRedirect()
  
  
  
  class BackupInformation(db.Model):
    """An entity to keep information on a datastore backup."""
  
    name = db.StringProperty()
    kinds = db.StringListProperty()
    namespaces = db.StringListProperty()
    filesystem = db.StringProperty(default=FILES_API_BLOBSTORE_FILESYSTEM)
    start_time = db.DateTimeProperty(auto_now_add=True)
    active_jobs = db.StringListProperty()
    completed_jobs = db.StringListProperty()
    complete_time = db.DateTimeProperty(default=None)
    blob_files = db.StringListProperty()
    original_app = db.StringProperty(default=None)
    gs_handle = db.TextProperty(default=None)
    destination = db.StringProperty()
  
    @classmethod
    def kind(cls):
      return utils.BACKUP_INFORMATION_KIND
  
    @classmethod
    def name_exists(cls, backup_name):
      query = BackupInformation.all(keys_only=True)
      query.filter('name =', backup_name)
      return query.get() is not None
  
    def create_kind_backup_files_key(self, kind):
      return db.Key.from_path(KindBackupFiles.kind(), kind, parent=self.key())
  
    def create_kind_backup_files(self, kind, kind_files):
      return KindBackupFiles(key=self.create_kind_backup_files_key(kind),
                             files=kind_files)
  
    def get_kind_backup_files(self, kinds=None):
      if kinds:
        return db.get([self.create_kind_backup_files_key(kind) for kind in kinds])
      else:
        return KindBackupFiles.all().ancestor(self).run()
  
  
  
  class KindBackupFiles(db.Model):
    """An entity to keep files information per kind for a backup.
  
    A key for this model should created using kind as a name and the associated
    BackupInformation as a parent.
    """
    files = db.StringListProperty(indexed=False)
  
    @property
    def backup_kind(self):
      return self.key().name()
  
    @classmethod
    def kind(cls):
      return utils.BACKUP_INFORMATION_FILES_KIND
  
  
  def BackupCompleteHandler(operation, job_id, mapreduce_state):
    """Updates BackupInformation record for a completed mapper job."""
    mapreduce_spec = mapreduce_state.mapreduce_spec
    filenames = mapreduce_spec.mapper.output_writer_class().get_filenames(
        mapreduce_state)
    _perform_backup_complete(
        operation,
        job_id,
        mapreduce_spec.mapper.params['entity_kind'],
        mapreduce_spec.params['backup_info_pk'],
        _get_gcs_path_prefix_from_params_dict(mapreduce_spec.mapper.params),
        filenames,
        mapreduce_spec.params.get('done_callback_queue'),
        mapreduce_spec.mapper.params['output_writer']['account_id'])
  
  
  @db.transactional
  def _perform_backup_complete(
      operation, job_id, kind, backup_info_pk, gcs_path_prefix, filenames, queue,
      account_id=None):
    backup_info = BackupInformation.get(backup_info_pk)
    if backup_info:
      if job_id in backup_info.active_jobs:
        backup_info.active_jobs.remove(job_id)
        backup_info.completed_jobs = list(
            set(backup_info.completed_jobs + [job_id]))
  
  
      filenames = [GCSUtil.add_gs_prefix_if_missing(name) for name in filenames]
      kind_backup_files = backup_info.get_kind_backup_files([kind])[0]
      if kind_backup_files:
        kind_backup_files.files = list(set(kind_backup_files.files + filenames))
      else:
        kind_backup_files = backup_info.create_kind_backup_files(kind, filenames)
      db.put((backup_info, kind_backup_files), force_writes=True)
      if operation.status == utils.DatastoreAdminOperation.STATUS_COMPLETED:
        deferred.defer(finalize_backup_info, backup_info.key(),
                       gcs_path_prefix,
                       account_id,
                       _url=config.DEFERRED_PATH,
                       _queue=queue,
                       _transactional=True)
    else:
      logging.warn('BackupInfo was not found for %s', backup_info_pk)
  
  
  def finalize_backup_info(backup_info_pk, gcs_path_prefix, account_id=None):
    """Finalize the state of BackupInformation and creates info file for GS."""
  
    def get_backup_info():
      return BackupInformation.get(backup_info_pk)
  
    backup_info = db.run_in_transaction(get_backup_info)
    if backup_info:
      complete_time = datetime.datetime.now()
      backup_info.complete_time = complete_time
      gs_handle = None
      if backup_info.filesystem == FILES_API_GS_FILESYSTEM:
  
  
  
  
  
        backup_info_writer = BackupInfoWriter(gcs_path_prefix, account_id)
        gs_handle = backup_info_writer.write(backup_info)[0]
  
      def set_backup_info_with_finalize_info():
        backup_info = get_backup_info()
        backup_info.complete_time = complete_time
        backup_info.gs_handle = gs_handle
        backup_info.put(force_writes=True)
      db.run_in_transaction(set_backup_info_with_finalize_info)
      logging.info('Backup %s completed', backup_info.name)
    else:
      logging.warn('Backup %s could not be found', backup_info_pk)
  
  
  def parse_backup_info_file(content):
    """Returns entities iterator from a backup_info file content."""
    reader = records.RecordsReader(cStringIO.StringIO(content))
    version = reader.read()
    if version != '1':
      raise IOError('Unsupported version')
    return (datastore.Entity.FromPb(record) for record in reader)
  
  
  class BackupInfoWriter(object):
    """A class for writing Datastore backup metadata files."""
  
    def __init__(self, gcs_path_prefix, account_id=None):
      """Construct a BackupInfoWriter.
  
      Args:
        gcs_path_prefix: (string) gcs prefix used for creating the backup.
      """
      self.__gcs_path_prefix = gcs_path_prefix
      self._account_id = account_id
  
    def write(self, backup_info):
      """Write the metadata files for the given backup_info.
  
      As a side effect, updates the backup_info in-memory entity object with the
      gs_handle to the Backup info filename. This is not saved to the datastore.
  
      Args:
        backup_info: Required BackupInformation.
  
      Returns:
        A list with Backup info filename followed by Kind info filenames.
      """
      fn = self._write_backup_info(backup_info)
      return [fn] + self._write_kind_info(backup_info)
  
    def _generate_filename(self, backup_info, suffix):
      key_str = str(backup_info.key()).replace('/', '_')
      return '/gs/%s/%s%s' % (self.__gcs_path_prefix, key_str, suffix)
  
    def _write_backup_info(self, backup_info):
      """Writes a backup_info_file.
  
      Args:
        backup_info: Required BackupInformation.
  
      Returns:
        Backup info filename.
      """
      filename = self._generate_filename(backup_info, '.backup_info')
      backup_info.gs_handle = filename
      with GCSUtil.open(filename, 'w', _account_id=self._account_id) as info_file:
        with records.RecordsWriter(info_file) as writer:
  
          writer.write('1')
  
          writer.write(db.model_to_protobuf(backup_info).SerializeToString())
  
          for kind_files in backup_info.get_kind_backup_files():
            writer.write(db.model_to_protobuf(kind_files).SerializeToString())
      return filename
  
    def _write_kind_info(self, backup_info):
      """Writes type information schema for each kind in backup_info.
  
      Args:
        backup_info: Required BackupInformation.
  
      Returns:
        A list with all created filenames.
      """
      def get_backup_files_tx():
        kind_backup_files_list = []
  
        for kind_backup_files in backup_info.get_kind_backup_files():
          kind_backup_files_list.append(kind_backup_files)
        return kind_backup_files_list
  
      kind_backup_files_list = db.run_in_transaction(get_backup_files_tx)
      filenames = []
      for kind_backup_files in kind_backup_files_list:
        backup = self._create_kind_backup(backup_info, kind_backup_files)
        filename = self._generate_filename(
            backup_info, '.%s.backup_info' % kind_backup_files.backup_kind)
        self._write_kind_backup_info_file(filename, backup, self._account_id)
        filenames.append(filename)
      return filenames
  
    def _create_kind_backup(self, backup_info, kind_backup_files):
      """Creates and populate a backup_pb2.Backup."""
      backup = backup_pb2.Backup()
      backup.backup_info.backup_name = backup_info.name
      backup.backup_info.start_timestamp = datastore_types.DatetimeToTimestamp(
          backup_info.start_time)
      backup.backup_info.end_timestamp = datastore_types.DatetimeToTimestamp(
          backup_info.complete_time)
      kind = kind_backup_files.backup_kind
      kind_info = backup.kind_info.add()
      kind_info.kind = kind
      kind_info.entity_schema.kind = kind
      kind_info.file.extend(kind_backup_files.files)
      entity_type_info = EntityTypeInfo(kind=kind)
      for sharded_aggregation in SchemaAggregationResult.load(
          backup_info.key(), kind):
        if sharded_aggregation.is_partial:
          kind_info.is_partial = True
        if sharded_aggregation.entity_type_info:
          entity_type_info.merge(sharded_aggregation.entity_type_info)
      entity_type_info.populate_entity_schema(kind_info.entity_schema)
      return backup
  
    @classmethod
    def _write_kind_backup_info_file(cls, filename, backup, account_id=None):
      """Writes a kind backup_info.
  
      Args:
        filename: The name of the file to be created as string.
        backup: apphosting.ext.datastore_admin.Backup proto.
      """
      with GCSUtil.open(filename, 'w', _account_id=account_id) as f:
        f.write(backup.SerializeToString())
  
  
  class PropertyTypeInfo(json_util.JsonMixin):
    """Type information for an entity property."""
  
    def __init__(self, name, is_repeated=False, primitive_types=None,
                 embedded_entities=None):
      """Construct a PropertyTypeInfo instance.
  
      Args:
        name: The name of the property as a string.
        is_repeated: A boolean that indicates if the property is repeated.
        primitive_types: Optional list of PrimitiveType integer values.
        embedded_entities: Optional list of EntityTypeInfo.
      """
      self.__name = name
      self.__is_repeated = is_repeated
      self.__primitive_types = set(primitive_types) if primitive_types else set()
      self.__embedded_entities = {}
      for entity in embedded_entities or ():
        if entity.kind in self.__embedded_entities:
          self.__embedded_entities[entity.kind].merge(entity)
        else:
          self.__embedded_entities[entity.kind] = entity
  
    @property
    def name(self):
      return self.__name
  
    @property
    def is_repeated(self):
      return self.__is_repeated
  
    @property
    def primitive_types(self):
      return self.__primitive_types
  
    def embedded_entities_kind_iter(self):
      return self.__embedded_entities.iterkeys()
  
    def get_embedded_entity(self, kind):
      return self.__embedded_entities.get(kind)
  
    def merge(self, other):
      """Merge a PropertyTypeInfo with this instance.
  
      Args:
        other: Required PropertyTypeInfo to merge.
  
      Returns:
        True if anything was changed. False otherwise.
  
      Raises:
        ValueError: if property names do not match.
        TypeError: if other is not instance of PropertyTypeInfo.
      """
      if not isinstance(other, PropertyTypeInfo):
        raise TypeError('Expected PropertyTypeInfo, was %r' % (other,))
  
      if other.__name != self.__name:
        raise ValueError('Property names mismatch (%s, %s)' %
                         (self.__name, other.__name))
      changed = False
      if other.__is_repeated and not self.__is_repeated:
        self.__is_repeated = True
        changed = True
      if not other.__primitive_types.issubset(self.__primitive_types):
        self.__primitive_types = self.__primitive_types.union(
            other.__primitive_types)
        changed = True
      for kind, other_embedded_entity in other.__embedded_entities.iteritems():
        embedded_entity = self.__embedded_entities.get(kind)
        if embedded_entity:
          changed = embedded_entity.merge(other_embedded_entity) or changed
        else:
          self.__embedded_entities[kind] = other_embedded_entity
          changed = True
      return changed
  
    def populate_entity_schema_field(self, entity_schema):
      """Add an populate a Field to the given entity_schema.
  
      Args:
        entity_schema: apphosting.ext.datastore_admin.EntitySchema proto.
      """
      if not (self.__primitive_types or self.__embedded_entities):
        return
  
      field = entity_schema.field.add()
      field.name = self.__name
      field_type = field.type.add()
      field_type.is_list = self.__is_repeated
      field_type.primitive_type.extend(self.__primitive_types)
      for embedded_entity in self.__embedded_entities.itervalues():
        embedded_entity_schema = field_type.embedded_schema.add()
        embedded_entity.populate_entity_schema(embedded_entity_schema)
  
    def to_json(self):
      json = dict()
      json['name'] = self.__name
      json['is_repeated'] = self.__is_repeated
      json['primitive_types'] = list(self.__primitive_types)
      json['embedded_entities'] = [e.to_json() for e in
                                   self.__embedded_entities.itervalues()]
      return json
  
    @classmethod
    def from_json(cls, json):
      return cls(json['name'], json['is_repeated'], json.get('primitive_types'),
                 [EntityTypeInfo.from_json(entity_json) for entity_json
                  in json.get('embedded_entities')])
  
  
  class EntityTypeInfo(json_util.JsonMixin):
    """Type information for an entity."""
  
    def __init__(self, kind=None, properties=None):
      """Construct an EntityTypeInfo instance.
  
      Args:
        kind: An optional kind name as string.
        properties: An optional list of PropertyTypeInfo.
      """
      self.__kind = kind
      self.__properties = {}
      for property_type_info in properties or ():
        if property_type_info.name in self.__properties:
          self.__properties[property_type_info.name].merge(property_type_info)
        else:
          self.__properties[property_type_info.name] = property_type_info
  
    @property
    def kind(self):
      return self.__kind
  
    def properties_name_iter(self):
      return self.__properties.iterkeys()
  
    def get_property(self, name):
      return self.__properties.get(name)
  
    def merge(self, other):
      """Merge an EntityTypeInfo with this instance.
  
      Args:
        other: Required EntityTypeInfo to merge.
  
      Returns:
        True if anything was changed. False otherwise.
  
      Raises:
        ValueError: if kinds do not match.
        TypeError: if other is not instance of EntityTypeInfo.
      """
      if not isinstance(other, EntityTypeInfo):
        raise TypeError('Expected EntityTypeInfo, was %r' % (other,))
  
      if other.__kind != self.__kind:
        raise ValueError('Kinds mismatch (%s, %s)' % (self.__kind, other.__kind))
      changed = False
      for name, other_property in other.__properties.iteritems():
        self_property = self.__properties.get(name)
        if self_property:
          changed = self_property.merge(other_property) or changed
        else:
          self.__properties[name] = other_property
          changed = True
      return changed
  
    def populate_entity_schema(self, entity_schema):
      """Populates the given entity_schema with values from this instance.
  
      Args:
        entity_schema: apphosting.ext.datastore_admin.EntitySchema proto.
      """
      if self.__kind:
        entity_schema.kind = self.__kind
      for property_type_info in self.__properties.itervalues():
        property_type_info.populate_entity_schema_field(entity_schema)
  
    def to_json(self):
      return {
          'kind': self.__kind,
          'properties': [p.to_json() for p in self.__properties.itervalues()]
      }
  
    @classmethod
    def from_json(cls, json):
      kind = json.get('kind')
      properties_json = json.get('properties')
      if properties_json:
        return cls(kind, [PropertyTypeInfo.from_json(p) for p in properties_json])
      else:
        return cls(kind)
  
    @classmethod
    def create_from_entity_proto(cls, entity_proto):
      """Creates and populates an EntityTypeInfo from an EntityProto."""
      properties = [cls.__get_property_type_info(property_proto) for
                    property_proto in itertools.chain(
                        entity_proto.property_list(),
                        entity_proto.raw_property_list())]
      kind = utils.get_kind_from_entity_pb(entity_proto)
      return cls(kind, properties)
  
    @classmethod
    def __get_property_type_info(cls, property_proto):
      """Returns the type mapping for the provided property."""
      name = property_proto.name()
      is_repeated = bool(property_proto.multiple())
      primitive_type = None
      entity_type = None
      if property_proto.has_meaning():
        primitive_type = MEANING_TO_PRIMITIVE_TYPE.get(property_proto.meaning())
      if primitive_type is None:
        value = property_proto.value()
        if value.has_int64value():
          primitive_type = backup_pb2.EntitySchema.INTEGER
        elif value.has_booleanvalue():
          primitive_type = backup_pb2.EntitySchema.BOOLEAN
        elif value.has_stringvalue():
          if property_proto.meaning() == entity_pb.Property.ENTITY_PROTO:
            entity_proto = entity_pb.EntityProto()
            try:
              entity_proto.ParsePartialFromString(value.stringvalue())
            except Exception:
  
              pass
            else:
              entity_type = EntityTypeInfo.create_from_entity_proto(entity_proto)
          else:
            primitive_type = backup_pb2.EntitySchema.STRING
        elif value.has_doublevalue():
          primitive_type = backup_pb2.EntitySchema.FLOAT
        elif value.has_pointvalue():
          primitive_type = backup_pb2.EntitySchema.GEO_POINT
        elif value.has_uservalue():
          primitive_type = backup_pb2.EntitySchema.USER
        elif value.has_referencevalue():
          primitive_type = backup_pb2.EntitySchema.REFERENCE
      return PropertyTypeInfo(
          name, is_repeated,
          (primitive_type,) if primitive_type is not None else None,
          (entity_type,) if entity_type else None)
  
  
  class SchemaAggregationResult(db.Model):
    """Persistent aggregated type information for a kind.
  
    An instance can be retrieved via the load method or created
    using the create method. An instance aggregates all type information
    for all seen embedded_entities via the merge method and persisted when needed
    using the model put method.
    """
  
    entity_type_info = json_util.JsonProperty(
        EntityTypeInfo, default=EntityTypeInfo(), indexed=False)
    is_partial = db.BooleanProperty(default=False)
  
    def merge(self, other):
      """Merge a SchemaAggregationResult or an EntityTypeInfo with this instance.
  
      Args:
        other: Required SchemaAggregationResult or EntityTypeInfo to merge.
  
      Returns:
        True if anything was changed. False otherwise.
      """
      if self.is_partial:
        return False
      if isinstance(other, SchemaAggregationResult):
        other = other.entity_type_info
      return self.entity_type_info.merge(other)
  
    @classmethod
    def _get_parent_key(cls, backup_id, kind_name):
      return datastore_types.Key.from_path('Kind', kind_name, parent=backup_id)
  
    @classmethod
    def create(cls, backup_id, kind_name, shard_id):
      """Create SchemaAggregationResult instance.
  
      Args:
        backup_id: Required BackupInformation Key.
        kind_name: Required kind name as string.
        shard_id: Required shard id as string.
  
      Returns:
        A new SchemaAggregationResult instance.
      """
      parent = cls._get_parent_key(backup_id, kind_name)
      return SchemaAggregationResult(
          key_name=shard_id, parent=parent,
          entity_type_info=EntityTypeInfo(kind=kind_name))
  
    @classmethod
    def load(cls, backup_id, kind_name, shard_id=None):
      """Retrieve SchemaAggregationResult from the Datastore.
  
      Args:
        backup_id: Required BackupInformation Key.
        kind_name: Required kind name as string.
        shard_id: Optional shard id as string.
  
      Returns:
        SchemaAggregationResult iterator or an entity if shard_id not None.
      """
      parent = cls._get_parent_key(backup_id, kind_name)
      if shard_id:
        key = datastore_types.Key.from_path(cls.kind(), shard_id, parent=parent)
        return SchemaAggregationResult.get(key)
      else:
        return db.Query(cls).ancestor(parent).run()
  
    @classmethod
    def kind(cls):
      return utils.BACKUP_INFORMATION_KIND_TYPE_INFO
  
  
  
  class SchemaAggregationPool(object):
    """An MR pool to aggregation type information per kind."""
  
    def __init__(self, backup_id, kind, shard_id):
      """Construct SchemaAggregationPool instance.
  
      Args:
        backup_id: Required BackupInformation Key.
        kind: Required kind name as string.
        shard_id: Required shard id as string.
      """
      self.__backup_id = backup_id
      self.__kind = kind
      self.__shard_id = shard_id
      self.__aggregation = SchemaAggregationResult.load(backup_id, kind, shard_id)
      if not self.__aggregation:
        self.__aggregation = SchemaAggregationResult.create(backup_id, kind,
                                                            shard_id)
        self.__needs_save = True
      else:
        self.__needs_save = False
  
    def merge(self, entity_type_info):
      """Merge EntityTypeInfo into aggregated type information."""
      if self.__aggregation.merge(entity_type_info):
        self.__needs_save = True
  
    def flush(self):
      """Save aggregated type information to the datastore if changed."""
      if self.__needs_save:
  
        def update_aggregation_tx():
          aggregation = SchemaAggregationResult.load(
              self.__backup_id, self.__kind, self.__shard_id)
          if aggregation:
            if aggregation.merge(self.__aggregation):
              aggregation.put(force_writes=True)
            self.__aggregation = aggregation
          else:
            self.__aggregation.put(force_writes=True)
  
        def mark_aggregation_as_partial_tx():
          aggregation = SchemaAggregationResult.load(
              self.__backup_id, self.__kind, self.__shard_id)
          if aggregation is None:
            aggregation = SchemaAggregationResult.create(
                self.__backup_id, self.__kind, self.__shard_id)
          aggregation.is_partial = True
          aggregation.put(force_writes=True)
          self.__aggregation = aggregation
  
        try:
          db.run_in_transaction(update_aggregation_tx)
  
  
  
        except (apiproxy_errors.RequestTooLargeError,
                datastore_errors.BadRequestError):
          db.run_in_transaction(mark_aggregation_as_partial_tx)
        self.__needs_save = False
  
  
  class AggregateSchema(op.Operation):
    """An MR Operation to aggregation type information for a kind.
  
    This operation will register an MR pool, SchemaAggregationPool, if
    one is not already registered and will invoke the pool's merge operation
    per entity. The pool is responsible for keeping a persistent state of
    type aggregation using the sharded db model, SchemaAggregationResult.
    """
  
    def __init__(self, entity_proto):
      self.__entity_info = EntityTypeInfo.create_from_entity_proto(entity_proto)
  
    def __call__(self, ctx):
      pool = ctx.get_pool('schema_aggregation_pool')
      if not pool:
        backup_id = datastore_types.Key(
            context.get().mapreduce_spec.params['backup_info_pk'])
        pool = SchemaAggregationPool(
            backup_id, self.__entity_info.kind, ctx.shard_id)
        ctx.register_pool('schema_aggregation_pool', pool)
      pool.merge(self.__entity_info)
  
  
  class BackupEntity(object):
    """A class which dumps the entity to the writer."""
  
    def map(self, entity_proto):
      """Backup entity map handler.
  
      Args:
        entity_proto: An instance of entity_pb.EntityProto.
  
      Yields:
        A serialized entity_pb.EntityProto as a string
      """
      yield entity_proto.SerializeToString()
      yield AggregateSchema(entity_proto)
  
  
  class RestoreEntity(object):
    """A class which restore the entity to datastore."""
  
    def __init__(self):
      self.initialized = False
      self.kind_filter = None
  
      self.app_id = None
  
    def initialize(self):
      """Initialize a restore mapper instance."""
      if self.initialized:
        return
      mapper_params = get_mapper_params_from_context()
      kind_filter = mapper_params.get('kind_filter')
      self.kind_filter = set(kind_filter) if kind_filter else None
      original_app = mapper_params.get('original_app')
      target_app = os.getenv('APPLICATION_ID')
      if original_app and target_app != original_app:
        self.app_id = target_app
      self.initialized = True
  
    def map(self, record):
      """Restore entity map handler.
  
      Args:
        record: A serialized entity_pb.EntityProto.
  
      Yields:
        A operation.db.Put for the mapped entity
      """
      self.initialize()
      pb = entity_pb.EntityProto(contents=record)
      if self.app_id:
        utils.FixKeys(pb, self.app_id)
  
  
  
      if not self.kind_filter or (
          utils.get_kind_from_entity_pb(pb) in self.kind_filter):
        yield utils.Put(pb)
        if self.app_id:
  
          yield utils.ReserveKey(datastore_types.Key._FromPb(pb.key()))
  
  
  def get_mapper_params_from_context():
    """Get mapper params from MR context. Split out for ease of testing."""
    return context.get().mapreduce_spec.mapper.params
  
  
  def validate_gcs_bucket_name(bucket_name):
    """Validate the format of the given bucket_name.
  
    Validation rules are based:
    https://developers.google.com/storage/docs/bucketnaming#requirements
  
    Args:
      bucket_name: The bucket name to validate.
  
    Raises:
      BackupValidationError: If the bucket name is invalid.
    """
    if len(bucket_name) > MAX_BUCKET_LEN:
      raise BackupValidationError(
          'Bucket name length should not be longer than %d' % MAX_BUCKET_LEN)
    if len(bucket_name) < MIN_BUCKET_LEN:
      raise BackupValidationError(
          'Bucket name length should be longer than %d' % MIN_BUCKET_LEN)
    if bucket_name.lower().startswith('goog'):
      raise BackupValidationError(
          'Bucket name should not start with a "goog" prefix')
    bucket_elements = bucket_name.split('.')
    for bucket_element in bucket_elements:
      if len(bucket_element) > MAX_BUCKET_SEGMENT_LEN:
        raise BackupValidationError(
            'Segment length of bucket name should not be longer than %d' %
            MAX_BUCKET_SEGMENT_LEN)
    if not re.match(BUCKET_PATTERN, bucket_name):
      raise BackupValidationError('Invalid bucket name "%s"' % bucket_name)
  
  
  def is_accessible_bucket_name(bucket_name, account_id=None):
    """Returns True if the application has access to the specified bucket."""
    scope = config.GoogleApiScope('devstorage.read_write')
    bucket_url = config.GsBucketURL(bucket_name)
    auth_token, _ = app_identity.get_access_token(scope, account_id)
    result = urlfetch.fetch(bucket_url, method=urlfetch.HEAD, headers={
        'Authorization': 'OAuth %s' % auth_token,
        'x-goog-api-version': '2'})
    return result and result.status_code == 200
  
  
  def verify_bucket_writable(bucket_name, account_id=None):
    """Verify the application can write to the specified bucket.
  
    Args:
      bucket_name: The bucket to verify.
  
    Raises:
      BackupValidationError: If the bucket is not writable.
    """
    path = '/gs/%s/%s' % (bucket_name, TEST_WRITE_FILENAME_PREFIX)
    try:
      gcs_stats = GCSUtil.listbucket(path, max_keys=MAX_KEYS_LIST_SIZE,
                                     _account_id=account_id)
      file_names = [f.filename for f in gcs_stats]
    except (cloudstorage.AuthorizationError, cloudstorage.ForbiddenError):
      raise BackupValidationError('Bucket "%s" not accessible' % bucket_name)
    except cloudstorage.NotFoundError:
      raise BackupValidationError('Bucket "%s" does not exist' % bucket_name)
  
  
    file_name = '/gs/%s/%s.tmp' % (bucket_name, TEST_WRITE_FILENAME_PREFIX)
    file_name_try = 0
    while True:
      if file_name_try >= MAX_TEST_FILENAME_TRIES:
  
  
        return
      if file_name not in file_names:
        break
      gen = random.randint(0, 9999)
      file_name = ('/gs/%s/%s_%s.tmp' %
                   (bucket_name, TEST_WRITE_FILENAME_PREFIX, gen))
      file_name_try += 1
    try:
      with GCSUtil.open(file_name, 'w', _account_id=account_id) as f:
        f.write('test')
    except cloudstorage.ForbiddenError:
      raise BackupValidationError('Bucket "%s" is not writable' % bucket_name)
    try:
      GCSUtil.delete(file_name, _account_id=account_id)
    except cloudstorage.Error:
      logging.warn('Failed to delete test file %s', file_name)
  
  
  def is_readable_gs_handle(gs_handle, account_id=None):
    """Return True if the application can read the specified gs_handle."""
    try:
      with GCSUtil.open(gs_handle, _account_id=account_id) as bak_file:
        bak_file.read(1)
    except (cloudstorage.ForbiddenError,
            cloudstorage.NotFoundError,
            cloudstorage.AuthorizationError):
      return False
    return True
  
  
  
  def parse_gs_handle(gs_handle):
    """Splits [/gs/]?bucket_name[/folder]*[/file]? to (bucket_name, path | '')."""
    if gs_handle.startswith('/'):
      filesystem = gs_handle[1:].split('/', 1)[0]
      if filesystem == 'gs':
        gs_handle = gs_handle[4:]
      else:
        raise BackupValidationError('Unsupported filesystem: %s' % filesystem)
    tokens = gs_handle.split('/', 1)
    return (tokens[0], '') if len(tokens) == 1 else tuple(tokens)
  
  
  def validate_and_split_gcs_path(gcs_path, account_id=None):
    bucket_name, path = parse_gs_handle(gcs_path)
    path = path.rstrip('/')
    validate_gcs_bucket_name(bucket_name)
    verify_bucket_writable(bucket_name, account_id)
    return bucket_name, path
  
  
  def list_bucket_files(bucket_name, prefix, max_keys=1000, account_id=None):
    """Returns a listing of of a bucket that matches the given prefix."""
    scope = config.GoogleApiScope('devstorage.read_only')
    bucket_url = config.GsBucketURL(bucket_name)
    url = bucket_url + '?'
    query = [('max-keys', max_keys)]
    if prefix:
      query.append(('prefix', prefix))
    url += urllib.urlencode(query)
    auth_token, _ = app_identity.get_access_token(scope, account_id)
    result = urlfetch.fetch(url, method=urlfetch.GET, headers={
        'Authorization': 'OAuth %s' % auth_token,
        'x-goog-api-version': '2'})
    if result and result.status_code == 200:
      doc = xml.dom.minidom.parseString(result.content)
      return [node.childNodes[0].data for node in doc.getElementsByTagName('Key')]
    raise BackupValidationError('Request to Google Cloud Storage failed')
  
  
  def get_gs_object(bucket_name, path, account_id=None):
    """Returns a listing of of a bucket that matches the given prefix."""
    scope = config.GoogleApiScope('devstorage.read_only')
    bucket_url = config.GsBucketURL(bucket_name)
    url = bucket_url + path
    auth_token, _ = app_identity.get_access_token(scope, account_id)
    result = urlfetch.fetch(url, method=urlfetch.GET, headers={
        'Authorization': 'OAuth %s' % auth_token,
        'x-goog-api-version': '2'})
    if result and result.status_code == 200:
      return result.content
    if result and result.status_code == 403:
      raise BackupValidationError(
          'Requested path %s is not accessible/access denied' % url)
    if result and result.status_code == 404:
      raise BackupValidationError('Requested path %s was not found' % url)
    raise BackupValidationError('Error encountered accessing requested path %s' %
                                url)
  
  
  
  
  def get_queue_names(app_id=None, max_rows=100):
    """Returns a list with all non-special queue names for app_id."""
    rpc = apiproxy_stub_map.UserRPC('taskqueue')
    request = taskqueue_service_pb.TaskQueueFetchQueuesRequest()
    response = taskqueue_service_pb.TaskQueueFetchQueuesResponse()
    if app_id:
      request.set_app_id(app_id)
    request.set_max_rows(max_rows)
    queues = ['default']
    try:
      rpc.make_call('FetchQueues', request, response)
      rpc.check_success()
  
      for queue in response.queue_list():
        if (queue.mode() == taskqueue_service_pb.TaskQueueMode.PUSH and
            not queue.queue_name().startswith('__') and
            queue.queue_name() != 'default'):
          queues.append(queue.queue_name())
    except Exception:
      logging.exception('Failed to get queue names.')
    return queues
  
  
  def handlers_list(base_path):
    return [
        (r'%s/%s' % (base_path, BackupLinkHandler.SUFFIX),
         BackupLinkHandler),
        (r'%s/%s' % (base_path, BackupImportAndRestoreLinkHandler.SUFFIX),
         BackupImportAndRestoreLinkHandler),
        (r'%s/%s' % (base_path, ConfirmBackupHandler.SUFFIX),
         ConfirmBackupHandler),
        (r'%s/%s' % (base_path, DoBackupHandler.SUFFIX), DoBackupHandler),
        (r'%s/%s' % (base_path, DoBackupRestoreHandler.SUFFIX),
         DoBackupRestoreHandler),
        (r'%s/%s' % (base_path, DoBackupDeleteHandler.SUFFIX),
         DoBackupDeleteHandler),
        (r'%s/%s' % (base_path, DoBackupAbortHandler.SUFFIX),
         DoBackupAbortHandler),
        (r'%s/%s' % (base_path, DoBackupImportHandler.SUFFIX),
         DoBackupImportHandler),
    ]
  
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 json
  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_service_account_names():
    """ AppScale: Fetch list of service accounts from IAM API. """
    project_id = app_identity.get_application_id()
    iam_location = 'https://127.0.0.1:17441'
    url = iam_location + '/v1/projects/{}/serviceAccounts'.format(project_id)
    token = app_identity.get_access_token(
        ['https://www.googleapis.com/auth/cloud-platform'])[0]
    headers = {'Authorization': 'Bearer {}'.format(token)}
    response = urlfetch.fetch(url, headers=headers, validate_certificate=False)
    try:
      accounts = json.loads(response.content)['accounts']
    except (KeyError, ValueError):
      raise ValueError('Invalid list of service accounts: '
                       '{}'.format(response.content))
  
    return tuple(account['email'] for account in accounts)
  
  
  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'),
          'service_accounts': get_service_account_names()
      }
      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', '')
        if not account_id:
          raise ValueError('Invalid service account name')
  
        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.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_info)
        except Exception, e:
          logging.exception('Failed to delete datastore backup.')
          params = [('error', e.message)]
  
      self.SendRedirect(params=params)
  
  
  class DoBackupAbortHandler(BaseDoHandler):
    """Handler to deal with datastore admin requests to abort pending backups."""
  
    SUFFIX = 'backup_abort.do'
  
    def get(self):
      self.post()
  
    def post(self):
      """Handler for post requests to datastore_admin/backup_abort.do.
  
      Abort is 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:
              operation = backup_info.parent()
              utils.AbortAdminOperation(operation.key())
              delete_backup_info(backup_info)
        except Exception, e:
          logging.exception('Failed to abort pending datastore backup.')
          params = [('error', e.message)]
  
      self.SendRedirect(params=params)
  
  
  def _restore(backup_id, kinds, run_as_a_service, queue, mapper_params):
    """Restore a backup."""
    if not backup_id:
      raise ValueError('Unspecified Backup.')
  
    backup = db.get(db.Key(backup_id))
    if not backup:
      raise ValueError('Invalid Backup id.')
  
    input_reader_to_use = None
  
    kinds = set(kinds)
    if not (backup.blob_files or kinds):
      raise ValueError('No kinds were selected')
    backup_kinds = set(backup.kinds)
    difference = kinds.difference(backup_kinds)
    if difference:
      raise ValueError('Backup does not have kind[s] %s' % ', '.join(difference))
  
    if run_as_a_service:
      raise ValueError(
          'Please use the Managed Import/Export service or remove '
          'run_as_a_service=T. '
          'https://cloud.google.com/datastore/docs/export-import-entities')
  
    job_name = 'datastore_backup_restore_%s' % re.sub(r'[^\w]', '_',
                                                      backup.name)
    job_operation = None
    try:
      operation_name = 'Restoring %s from backup: %s' % (
          ', '.join(kinds) if kinds else 'all', backup.name)
      job_operation = utils.StartOperation(operation_name)
  
  
      kinds = list(kinds) if len(backup_kinds) != len(kinds) else []
      mapper_params.update({
          'files': get_backup_files(backup, kinds),
          'kind_filter': kinds,
          'original_app': backup.original_app,
  
  
          parameters.DYNAMIC_RATE_INITIAL_QPS_PARAM: 500,
          parameters.DYNAMIC_RATE_BUMP_FACTOR_PARAM: 1.5,
          parameters.DYNAMIC_RATE_BUMP_TIME_PARAM: 300,
      })
  
      if backup.filesystem == FILES_API_GS_FILESYSTEM:
        input_reader_to_use = (input_readers.__name__ +
                               '.GoogleCloudStorageRecordInputReader')
        if not is_readable_gs_handle(backup.gs_handle):
          raise ValueError('Backup not readable.')
  
        if not mapper_params['files']:
          raise ValueError('No blob objects in restore.')
        bucket = parse_gs_handle(mapper_params['files'][0])[0]
  
  
  
  
  
  
        mapper_params['input_reader'] = copy.copy(mapper_params)
        mapper_params['input_reader'].update({
  
  
  
  
            'objects': [parse_gs_handle(f)[1] for f in mapper_params['files']],
            'bucket_name': bucket,
        })
      elif backup.filesystem == FILES_API_BLOBSTORE_FILESYSTEM:
        input_reader_to_use = __name__ + '.BlobstoreRecordsReader'
      else:
        raise ValueError('Unknown backup filesystem.')
  
      mapreduce_params = {
          'backup_name': backup.name,
          'force_ops_writes': True,
      }
      shard_count = min(max(utils.MAPREDUCE_MIN_SHARDS,
                            len(mapper_params['files'])),
                        utils.MAPREDUCE_MAX_SHARDS)
      job = utils.StartMap(
          job_operation.key(), job_name, BACKUP_RESTORE_HANDLER,
          input_reader_to_use, None, mapper_params, mapreduce_params,
          queue_name=queue, shard_count=shard_count)
      return ('job', job)
    except ValueError:
      raise
    except Exception:
      logging.exception('Failed to start a restore from backup job "%s".',
                        job_name)
      if job_operation:
        job_operation.status = utils.DatastoreAdminOperation.STATUS_FAILED
        job_operation.put(force_writes=True)
      raise
  
  
  class DoBackupRestoreHandler(BaseDoHandler):
    """Handler to restore backup data.
  
    Deals with requests from the admin console.
    """
    SUFFIX = 'backup_restore.do'
    _get_html_page = 'do_restore_from_backup.html'
    _get_post_html_page = SUFFIX
  
    def _ProcessPostRequest(self):
      """Triggers backup restore mapper jobs and returns their ids."""
      try:
        return [
            _restore(
                backup_id=self.request.get('backup_id'),
                kinds=self.request.get_all('kind'),
                run_as_a_service=self.request.get('run_as_a_service', False),
                queue=self.request.get('queue'),
                mapper_params=_get_basic_mapper_params(self)
            )
        ]
      except ValueError, e:
        return [('error', e.message)]
  
  
  def _import_backup(gs_handle):
    """Import backup from `gs_handle` to the current project."""
    bucket_name, path = parse_gs_handle(gs_handle)
    file_content = get_gs_object(bucket_name, path)
    entities = parse_backup_info_file(file_content)
    original_backup_info = entities.next()
    entity = datastore.Entity(BackupInformation.kind())
    entity.update(original_backup_info)
    backup_info = BackupInformation.from_entity(entity)
    if original_backup_info.key().app() != os.getenv('APPLICATION_ID'):
      backup_info.original_app = original_backup_info.key().app()
  
    def tx():
      backup_info.put(force_writes=True)
      kind_files_models = []
      for entity in entities:
        kind_files = backup_info.create_kind_backup_files(
            entity.key().name(), entity['files'])
        kind_files_models.append(kind_files)
      db.put(kind_files_models, force_writes=True)
    db.run_in_transaction(tx)
    return str(backup_info.key())
  
  
  class BackupImportAndRestoreLinkHandler(BaseLinkHandler):
    """Handler that imports and restores a backup."""
  
    SUFFIX = 'backup_import_restore.create'
  
    def _ProcessPostRequest(self):
      """Handler for post requests to datastore_admin/import_backup.create."""
      _restore(
          backup_id=_import_backup(self.request.get('gs_handle')),
          kinds=self.request.get_all('kind'),
          run_as_a_service=self.request.get('run_as_a_service', False),
          queue=self.request.get('queue'),
          mapper_params=_get_basic_mapper_params(self)
      )
  
  
  class DoBackupImportHandler(BaseDoHandler):
    """Handler to deal with datastore admin requests to import backup info."""
  
    SUFFIX = 'import_backup.do'
  
    def get(self):
      self.post()
  
    def post(self):
      """Handler for post requests to datastore_admin/import_backup.do.
  
      Import is executed and user is redirected to the base-path handler.
      """
      gs_handle = self.request.get('gs_handle')
      token = self.request.get('xsrf_token')
      error = None
      if gs_handle and utils.ValidateXsrfToken(token, XSRF_ACTION):
        try:
          backup_id = _import_backup(gs_handle)
        except Exception, e:
          logging.exception('Failed to Import datastore backup information.')
          error = e.message
  
      if error:
        self.SendRedirect(params=[('error', error)])
      elif self.request.get('Restore'):
        ConfirmRestoreFromBackupHandler.Render(
            self, default_backup_id=backup_id,
            default_delete_backup_after_restore=True)
      else:
        self.SendRedirect()
  
  
  
  class BackupInformation(db.Model):
    """An entity to keep information on a datastore backup."""
  
    name = db.StringProperty()
    kinds = db.StringListProperty()
    namespaces = db.StringListProperty()
    filesystem = db.StringProperty(default=FILES_API_BLOBSTORE_FILESYSTEM)
    start_time = db.DateTimeProperty(auto_now_add=True)
    active_jobs = db.StringListProperty()
    completed_jobs = db.StringListProperty()
    complete_time = db.DateTimeProperty(default=None)
    blob_files = db.StringListProperty()
    original_app = db.StringProperty(default=None)
    gs_handle = db.TextProperty(default=None)
    destination = db.StringProperty()
  
    @classmethod
    def kind(cls):
      return utils.BACKUP_INFORMATION_KIND
  
    @classmethod
    def name_exists(cls, backup_name):
      query = BackupInformation.all(keys_only=True)
      query.filter('name =', backup_name)
      return query.get() is not None
  
    def create_kind_backup_files_key(self, kind):
      return db.Key.from_path(KindBackupFiles.kind(), kind, parent=self.key())
  
    def create_kind_backup_files(self, kind, kind_files):
      return KindBackupFiles(key=self.create_kind_backup_files_key(kind),
                             files=kind_files)
  
    def get_kind_backup_files(self, kinds=None):
      if kinds:
        return db.get([self.create_kind_backup_files_key(kind) for kind in kinds])
      else:
        return KindBackupFiles.all().ancestor(self).run()
  
  
  
  class KindBackupFiles(db.Model):
    """An entity to keep files information per kind for a backup.
  
    A key for this model should created using kind as a name and the associated
    BackupInformation as a parent.
    """
    files = db.StringListProperty(indexed=False)
  
    @property
    def backup_kind(self):
      return self.key().name()
  
    @classmethod
    def kind(cls):
      return utils.BACKUP_INFORMATION_FILES_KIND
  
  
  def BackupCompleteHandler(operation, job_id, mapreduce_state):
    """Updates BackupInformation record for a completed mapper job."""
    mapreduce_spec = mapreduce_state.mapreduce_spec
    filenames = mapreduce_spec.mapper.output_writer_class().get_filenames(
        mapreduce_state)
    _perform_backup_complete(
        operation,
        job_id,
        mapreduce_spec.mapper.params['entity_kind'],
        mapreduce_spec.params['backup_info_pk'],
        _get_gcs_path_prefix_from_params_dict(mapreduce_spec.mapper.params),
        filenames,
        mapreduce_spec.params.get('done_callback_queue'),
        mapreduce_spec.mapper.params['output_writer']['account_id'])
  
  
  @db.transactional
  def _perform_backup_complete(
      operation, job_id, kind, backup_info_pk, gcs_path_prefix, filenames, queue,
      account_id=None):
    backup_info = BackupInformation.get(backup_info_pk)
    if backup_info:
      if job_id in backup_info.active_jobs:
        backup_info.active_jobs.remove(job_id)
        backup_info.completed_jobs = list(
            set(backup_info.completed_jobs + [job_id]))
  
  
      filenames = [GCSUtil.add_gs_prefix_if_missing(name) for name in filenames]
      kind_backup_files = backup_info.get_kind_backup_files([kind])[0]
      if kind_backup_files:
        kind_backup_files.files = list(set(kind_backup_files.files + filenames))
      else:
        kind_backup_files = backup_info.create_kind_backup_files(kind, filenames)
      db.put((backup_info, kind_backup_files), force_writes=True)
      if operation.status == utils.DatastoreAdminOperation.STATUS_COMPLETED:
        deferred.defer(finalize_backup_info, backup_info.key(),
                       gcs_path_prefix,
                       account_id,
                       _url=config.DEFERRED_PATH,
                       _queue=queue,
                       _transactional=True)
    else:
      logging.warn('BackupInfo was not found for %s', backup_info_pk)
  
  
  def finalize_backup_info(backup_info_pk, gcs_path_prefix, account_id=None):
    """Finalize the state of BackupInformation and creates info file for GS."""
  
    def get_backup_info():
      return BackupInformation.get(backup_info_pk)
  
    backup_info = db.run_in_transaction(get_backup_info)
    if backup_info:
      complete_time = datetime.datetime.now()
      backup_info.complete_time = complete_time
      gs_handle = None
      if backup_info.filesystem == FILES_API_GS_FILESYSTEM:
  
  
  
  
  
        backup_info_writer = BackupInfoWriter(gcs_path_prefix, account_id)
        gs_handle = backup_info_writer.write(backup_info)[0]
  
      def set_backup_info_with_finalize_info():
        backup_info = get_backup_info()
        backup_info.complete_time = complete_time
        backup_info.gs_handle = gs_handle
        backup_info.put(force_writes=True)
      db.run_in_transaction(set_backup_info_with_finalize_info)
      logging.info('Backup %s completed', backup_info.name)
    else:
      logging.warn('Backup %s could not be found', backup_info_pk)
  
  
  def parse_backup_info_file(content):
    """Returns entities iterator from a backup_info file content."""
    reader = records.RecordsReader(cStringIO.StringIO(content))
    version = reader.read()
    if version != '1':
      raise IOError('Unsupported version')
    return (datastore.Entity.FromPb(record) for record in reader)
  
  
  class BackupInfoWriter(object):
    """A class for writing Datastore backup metadata files."""
  
    def __init__(self, gcs_path_prefix, account_id=None):
      """Construct a BackupInfoWriter.
  
      Args:
        gcs_path_prefix: (string) gcs prefix used for creating the backup.
      """
      self.__gcs_path_prefix = gcs_path_prefix
      self._account_id = account_id
  
    def write(self, backup_info):
      """Write the metadata files for the given backup_info.
  
      As a side effect, updates the backup_info in-memory entity object with the
      gs_handle to the Backup info filename. This is not saved to the datastore.
  
      Args:
        backup_info: Required BackupInformation.
  
      Returns:
        A list with Backup info filename followed by Kind info filenames.
      """
      fn = self._write_backup_info(backup_info)
      return [fn] + self._write_kind_info(backup_info)
  
    def _generate_filename(self, backup_info, suffix):
      key_str = str(backup_info.key()).replace('/', '_')
      return '/gs/%s/%s%s' % (self.__gcs_path_prefix, key_str, suffix)
  
    def _write_backup_info(self, backup_info):
      """Writes a backup_info_file.
  
      Args:
        backup_info: Required BackupInformation.
  
      Returns:
        Backup info filename.
      """
      filename = self._generate_filename(backup_info, '.backup_info')
      backup_info.gs_handle = filename
      with GCSUtil.open(filename, 'w', _account_id=self._account_id) as info_file:
        with records.RecordsWriter(info_file) as writer:
  
          writer.write('1')
  
          writer.write(db.model_to_protobuf(backup_info).SerializeToString())
  
          for kind_files in backup_info.get_kind_backup_files():
            writer.write(db.model_to_protobuf(kind_files).SerializeToString())
      return filename
  
    def _write_kind_info(self, backup_info):
      """Writes type information schema for each kind in backup_info.
  
      Args:
        backup_info: Required BackupInformation.
  
      Returns:
        A list with all created filenames.
      """
      def get_backup_files_tx():
        kind_backup_files_list = []
  
        for kind_backup_files in backup_info.get_kind_backup_files():
          kind_backup_files_list.append(kind_backup_files)
        return kind_backup_files_list
  
      kind_backup_files_list = db.run_in_transaction(get_backup_files_tx)
      filenames = []
      for kind_backup_files in kind_backup_files_list:
        backup = self._create_kind_backup(backup_info, kind_backup_files)
        filename = self._generate_filename(
            backup_info, '.%s.backup_info' % kind_backup_files.backup_kind)
        self._write_kind_backup_info_file(filename, backup, self._account_id)
        filenames.append(filename)
      return filenames
  
    def _create_kind_backup(self, backup_info, kind_backup_files):
      """Creates and populate a backup_pb2.Backup."""
      backup = backup_pb2.Backup()
      backup.backup_info.backup_name = backup_info.name
      backup.backup_info.start_timestamp = datastore_types.DatetimeToTimestamp(
          backup_info.start_time)
      backup.backup_info.end_timestamp = datastore_types.DatetimeToTimestamp(
          backup_info.complete_time)
      kind = kind_backup_files.backup_kind
      kind_info = backup.kind_info.add()
      kind_info.kind = kind
      kind_info.entity_schema.kind = kind
      kind_info.file.extend(kind_backup_files.files)
      entity_type_info = EntityTypeInfo(kind=kind)
      for sharded_aggregation in SchemaAggregationResult.load(
          backup_info.key(), kind):
        if sharded_aggregation.is_partial:
          kind_info.is_partial = True
        if sharded_aggregation.entity_type_info:
          entity_type_info.merge(sharded_aggregation.entity_type_info)
      entity_type_info.populate_entity_schema(kind_info.entity_schema)
      return backup
  
    @classmethod
    def _write_kind_backup_info_file(cls, filename, backup, account_id=None):
      """Writes a kind backup_info.
  
      Args:
        filename: The name of the file to be created as string.
        backup: apphosting.ext.datastore_admin.Backup proto.
      """
      with GCSUtil.open(filename, 'w', _account_id=account_id) as f:
        f.write(backup.SerializeToString())
  
  
  class PropertyTypeInfo(json_util.JsonMixin):
    """Type information for an entity property."""
  
    def __init__(self, name, is_repeated=False, primitive_types=None,
                 embedded_entities=None):
      """Construct a PropertyTypeInfo instance.
  
      Args:
        name: The name of the property as a string.
        is_repeated: A boolean that indicates if the property is repeated.
        primitive_types: Optional list of PrimitiveType integer values.
        embedded_entities: Optional list of EntityTypeInfo.
      """
      self.__name = name
      self.__is_repeated = is_repeated
      self.__primitive_types = set(primitive_types) if primitive_types else set()
      self.__embedded_entities = {}
      for entity in embedded_entities or ():
        if entity.kind in self.__embedded_entities:
          self.__embedded_entities[entity.kind].merge(entity)
        else:
          self.__embedded_entities[entity.kind] = entity
  
    @property
    def name(self):
      return self.__name
  
    @property
    def is_repeated(self):
      return self.__is_repeated
  
    @property
    def primitive_types(self):
      return self.__primitive_types
  
    def embedded_entities_kind_iter(self):
      return self.__embedded_entities.iterkeys()
  
    def get_embedded_entity(self, kind):
      return self.__embedded_entities.get(kind)
  
    def merge(self, other):
      """Merge a PropertyTypeInfo with this instance.
  
      Args:
        other: Required PropertyTypeInfo to merge.
  
      Returns:
        True if anything was changed. False otherwise.
  
      Raises:
        ValueError: if property names do not match.
        TypeError: if other is not instance of PropertyTypeInfo.
      """
      if not isinstance(other, PropertyTypeInfo):
        raise TypeError('Expected PropertyTypeInfo, was %r' % (other,))
  
      if other.__name != self.__name:
        raise ValueError('Property names mismatch (%s, %s)' %
                         (self.__name, other.__name))
      changed = False
      if other.__is_repeated and not self.__is_repeated:
        self.__is_repeated = True
        changed = True
      if not other.__primitive_types.issubset(self.__primitive_types):
        self.__primitive_types = self.__primitive_types.union(
            other.__primitive_types)
        changed = True
      for kind, other_embedded_entity in other.__embedded_entities.iteritems():
        embedded_entity = self.__embedded_entities.get(kind)
        if embedded_entity:
          changed = embedded_entity.merge(other_embedded_entity) or changed
        else:
          self.__embedded_entities[kind] = other_embedded_entity
          changed = True
      return changed
  
    def populate_entity_schema_field(self, entity_schema):
      """Add an populate a Field to the given entity_schema.
  
      Args:
        entity_schema: apphosting.ext.datastore_admin.EntitySchema proto.
      """
      if not (self.__primitive_types or self.__embedded_entities):
        return
  
      field = entity_schema.field.add()
      field.name = self.__name
      field_type = field.type.add()
      field_type.is_list = self.__is_repeated
      field_type.primitive_type.extend(self.__primitive_types)
      for embedded_entity in self.__embedded_entities.itervalues():
        embedded_entity_schema = field_type.embedded_schema.add()
        embedded_entity.populate_entity_schema(embedded_entity_schema)
  
    def to_json(self):
      json = dict()
      json['name'] = self.__name
      json['is_repeated'] = self.__is_repeated
      json['primitive_types'] = list(self.__primitive_types)
      json['embedded_entities'] = [e.to_json() for e in
                                   self.__embedded_entities.itervalues()]
      return json
  
    @classmethod
    def from_json(cls, json):
      return cls(json['name'], json['is_repeated'], json.get('primitive_types'),
                 [EntityTypeInfo.from_json(entity_json) for entity_json
                  in json.get('embedded_entities')])
  
  
  class EntityTypeInfo(json_util.JsonMixin):
    """Type information for an entity."""
  
    def __init__(self, kind=None, properties=None):
      """Construct an EntityTypeInfo instance.
  
      Args:
        kind: An optional kind name as string.
        properties: An optional list of PropertyTypeInfo.
      """
      self.__kind = kind
      self.__properties = {}
      for property_type_info in properties or ():
        if property_type_info.name in self.__properties:
          self.__properties[property_type_info.name].merge(property_type_info)
        else:
          self.__properties[property_type_info.name] = property_type_info
  
    @property
    def kind(self):
      return self.__kind
  
    def properties_name_iter(self):
      return self.__properties.iterkeys()
  
    def get_property(self, name):
      return self.__properties.get(name)
  
    def merge(self, other):
      """Merge an EntityTypeInfo with this instance.
  
      Args:
        other: Required EntityTypeInfo to merge.
  
      Returns:
        True if anything was changed. False otherwise.
  
      Raises:
        ValueError: if kinds do not match.
        TypeError: if other is not instance of EntityTypeInfo.
      """
      if not isinstance(other, EntityTypeInfo):
        raise TypeError('Expected EntityTypeInfo, was %r' % (other,))
  
      if other.__kind != self.__kind:
        raise ValueError('Kinds mismatch (%s, %s)' % (self.__kind, other.__kind))
      changed = False
      for name, other_property in other.__properties.iteritems():
        self_property = self.__properties.get(name)
        if self_property:
          changed = self_property.merge(other_property) or changed
        else:
          self.__properties[name] = other_property
          changed = True
      return changed
  
    def populate_entity_schema(self, entity_schema):
      """Populates the given entity_schema with values from this instance.
  
      Args:
        entity_schema: apphosting.ext.datastore_admin.EntitySchema proto.
      """
      if self.__kind:
        entity_schema.kind = self.__kind
      for property_type_info in self.__properties.itervalues():
        property_type_info.populate_entity_schema_field(entity_schema)
  
    def to_json(self):
      return {
          'kind': self.__kind,
          'properties': [p.to_json() for p in self.__properties.itervalues()]
      }
  
    @classmethod
    def from_json(cls, json):
      kind = json.get('kind')
      properties_json = json.get('properties')
      if properties_json:
        return cls(kind, [PropertyTypeInfo.from_json(p) for p in properties_json])
      else:
        return cls(kind)
  
    @classmethod
    def create_from_entity_proto(cls, entity_proto):
      """Creates and populates an EntityTypeInfo from an EntityProto."""
      properties = [cls.__get_property_type_info(property_proto) for
                    property_proto in itertools.chain(
                        entity_proto.property_list(),
                        entity_proto.raw_property_list())]
      kind = utils.get_kind_from_entity_pb(entity_proto)
      return cls(kind, properties)
  
    @classmethod
    def __get_property_type_info(cls, property_proto):
      """Returns the type mapping for the provided property."""
      name = property_proto.name()
      is_repeated = bool(property_proto.multiple())
      primitive_type = None
      entity_type = None
      if property_proto.has_meaning():
        primitive_type = MEANING_TO_PRIMITIVE_TYPE.get(property_proto.meaning())
      if primitive_type is None:
        value = property_proto.value()
        if value.has_int64value():
          primitive_type = backup_pb2.EntitySchema.INTEGER
        elif value.has_booleanvalue():
          primitive_type = backup_pb2.EntitySchema.BOOLEAN
        elif value.has_stringvalue():
          if property_proto.meaning() == entity_pb.Property.ENTITY_PROTO:
            entity_proto = entity_pb.EntityProto()
            try:
              entity_proto.ParsePartialFromString(value.stringvalue())
            except Exception:
  
              pass
            else:
              entity_type = EntityTypeInfo.create_from_entity_proto(entity_proto)
          else:
            primitive_type = backup_pb2.EntitySchema.STRING
        elif value.has_doublevalue():
          primitive_type = backup_pb2.EntitySchema.FLOAT
        elif value.has_pointvalue():
          primitive_type = backup_pb2.EntitySchema.GEO_POINT
        elif value.has_uservalue():
          primitive_type = backup_pb2.EntitySchema.USER
        elif value.has_referencevalue():
          primitive_type = backup_pb2.EntitySchema.REFERENCE
      return PropertyTypeInfo(
          name, is_repeated,
          (primitive_type,) if primitive_type is not None else None,
          (entity_type,) if entity_type else None)
  
  
  class SchemaAggregationResult(db.Model):
    """Persistent aggregated type information for a kind.
  
    An instance can be retrieved via the load method or created
    using the create method. An instance aggregates all type information
    for all seen embedded_entities via the merge method and persisted when needed
    using the model put method.
    """
  
    entity_type_info = json_util.JsonProperty(
        EntityTypeInfo, default=EntityTypeInfo(), indexed=False)
    is_partial = db.BooleanProperty(default=False)
  
    def merge(self, other):
      """Merge a SchemaAggregationResult or an EntityTypeInfo with this instance.
  
      Args:
        other: Required SchemaAggregationResult or EntityTypeInfo to merge.
  
      Returns:
        True if anything was changed. False otherwise.
      """
      if self.is_partial:
        return False
      if isinstance(other, SchemaAggregationResult):
        other = other.entity_type_info
      return self.entity_type_info.merge(other)
  
    @classmethod
    def _get_parent_key(cls, backup_id, kind_name):
      return datastore_types.Key.from_path('Kind', kind_name, parent=backup_id)
  
    @classmethod
    def create(cls, backup_id, kind_name, shard_id):
      """Create SchemaAggregationResult instance.
  
      Args:
        backup_id: Required BackupInformation Key.
        kind_name: Required kind name as string.
        shard_id: Required shard id as string.
  
      Returns:
        A new SchemaAggregationResult instance.
      """
      parent = cls._get_parent_key(backup_id, kind_name)
      return SchemaAggregationResult(
          key_name=shard_id, parent=parent,
          entity_type_info=EntityTypeInfo(kind=kind_name))
  
    @classmethod
    def load(cls, backup_id, kind_name, shard_id=None):
      """Retrieve SchemaAggregationResult from the Datastore.
  
      Args:
        backup_id: Required BackupInformation Key.
        kind_name: Required kind name as string.
        shard_id: Optional shard id as string.
  
      Returns:
        SchemaAggregationResult iterator or an entity if shard_id not None.
      """
      parent = cls._get_parent_key(backup_id, kind_name)
      if shard_id:
        key = datastore_types.Key.from_path(cls.kind(), shard_id, parent=parent)
        return SchemaAggregationResult.get(key)
      else:
        return db.Query(cls).ancestor(parent).run()
  
    @classmethod
    def kind(cls):
      return utils.BACKUP_INFORMATION_KIND_TYPE_INFO
  
  
  
  class SchemaAggregationPool(object):
    """An MR pool to aggregation type information per kind."""
  
    def __init__(self, backup_id, kind, shard_id):
      """Construct SchemaAggregationPool instance.
  
      Args:
        backup_id: Required BackupInformation Key.
        kind: Required kind name as string.
        shard_id: Required shard id as string.
      """
      self.__backup_id = backup_id
      self.__kind = kind
      self.__shard_id = shard_id
      self.__aggregation = SchemaAggregationResult.load(backup_id, kind, shard_id)
      if not self.__aggregation:
        self.__aggregation = SchemaAggregationResult.create(backup_id, kind,
                                                            shard_id)
        self.__needs_save = True
      else:
        self.__needs_save = False
  
    def merge(self, entity_type_info):
      """Merge EntityTypeInfo into aggregated type information."""
      if self.__aggregation.merge(entity_type_info):
        self.__needs_save = True
  
    def flush(self):
      """Save aggregated type information to the datastore if changed."""
      if self.__needs_save:
  
        def update_aggregation_tx():
          aggregation = SchemaAggregationResult.load(
              self.__backup_id, self.__kind, self.__shard_id)
          if aggregation:
            if aggregation.merge(self.__aggregation):
              aggregation.put(force_writes=True)
            self.__aggregation = aggregation
          else:
            self.__aggregation.put(force_writes=True)
  
        def mark_aggregation_as_partial_tx():
          aggregation = SchemaAggregationResult.load(
              self.__backup_id, self.__kind, self.__shard_id)
          if aggregation is None:
            aggregation = SchemaAggregationResult.create(
                self.__backup_id, self.__kind, self.__shard_id)
          aggregation.is_partial = True
          aggregation.put(force_writes=True)
          self.__aggregation = aggregation
  
        try:
          db.run_in_transaction(update_aggregation_tx)
  
  
  
        except (apiproxy_errors.RequestTooLargeError,
                datastore_errors.BadRequestError):
          db.run_in_transaction(mark_aggregation_as_partial_tx)
        self.__needs_save = False
  
  
  class AggregateSchema(op.Operation):
    """An MR Operation to aggregation type information for a kind.
  
    This operation will register an MR pool, SchemaAggregationPool, if
    one is not already registered and will invoke the pool's merge operation
    per entity. The pool is responsible for keeping a persistent state of
    type aggregation using the sharded db model, SchemaAggregationResult.
    """
  
    def __init__(self, entity_proto):
      self.__entity_info = EntityTypeInfo.create_from_entity_proto(entity_proto)
  
    def __call__(self, ctx):
      pool = ctx.get_pool('schema_aggregation_pool')
      if not pool:
        backup_id = datastore_types.Key(
            context.get().mapreduce_spec.params['backup_info_pk'])
        pool = SchemaAggregationPool(
            backup_id, self.__entity_info.kind, ctx.shard_id)
        ctx.register_pool('schema_aggregation_pool', pool)
      pool.merge(self.__entity_info)
  
  
  class BackupEntity(object):
    """A class which dumps the entity to the writer."""
  
    def map(self, entity_proto):
      """Backup entity map handler.
  
      Args:
        entity_proto: An instance of entity_pb.EntityProto.
  
      Yields:
        A serialized entity_pb.EntityProto as a string
      """
      yield entity_proto.SerializeToString()
      yield AggregateSchema(entity_proto)
  
  
  class RestoreEntity(object):
    """A class which restore the entity to datastore."""
  
    def __init__(self):
      self.initialized = False
      self.kind_filter = None
  
      self.app_id = None
  
    def initialize(self):
      """Initialize a restore mapper instance."""
      if self.initialized:
        return
      mapper_params = get_mapper_params_from_context()
      kind_filter = mapper_params.get('kind_filter')
      self.kind_filter = set(kind_filter) if kind_filter else None
      original_app = mapper_params.get('original_app')
      target_app = os.getenv('APPLICATION_ID')
      if original_app and target_app != original_app:
        self.app_id = target_app
      self.initialized = True
  
    def map(self, record):
      """Restore entity map handler.
  
      Args:
        record: A serialized entity_pb.EntityProto.
  
      Yields:
        A operation.db.Put for the mapped entity
      """
      self.initialize()
      pb = entity_pb.EntityProto(contents=record)
      if self.app_id:
        utils.FixKeys(pb, self.app_id)
  
  
  
      if not self.kind_filter or (
          utils.get_kind_from_entity_pb(pb) in self.kind_filter):
        yield utils.Put(pb)
        if self.app_id:
  
          yield utils.ReserveKey(datastore_types.Key._FromPb(pb.key()))
  
  
  def get_mapper_params_from_context():
    """Get mapper params from MR context. Split out for ease of testing."""
    return context.get().mapreduce_spec.mapper.params
  
  
  def validate_gcs_bucket_name(bucket_name):
    """Validate the format of the given bucket_name.
  
    Validation rules are based:
    https://developers.google.com/storage/docs/bucketnaming#requirements
  
    Args:
      bucket_name: The bucket name to validate.
  
    Raises:
      BackupValidationError: If the bucket name is invalid.
    """
    if len(bucket_name) > MAX_BUCKET_LEN:
      raise BackupValidationError(
          'Bucket name length should not be longer than %d' % MAX_BUCKET_LEN)
    if len(bucket_name) < MIN_BUCKET_LEN:
      raise BackupValidationError(
          'Bucket name length should be longer than %d' % MIN_BUCKET_LEN)
    if bucket_name.lower().startswith('goog'):
      raise BackupValidationError(
          'Bucket name should not start with a "goog" prefix')
    bucket_elements = bucket_name.split('.')
    for bucket_element in bucket_elements:
      if len(bucket_element) > MAX_BUCKET_SEGMENT_LEN:
        raise BackupValidationError(
            'Segment length of bucket name should not be longer than %d' %
            MAX_BUCKET_SEGMENT_LEN)
    if not re.match(BUCKET_PATTERN, bucket_name):
      raise BackupValidationError('Invalid bucket name "%s"' % bucket_name)
  
  
  def is_accessible_bucket_name(bucket_name):
    """Returns True if the application has access to the specified bucket."""
    scope = config.GoogleApiScope('devstorage.read_write')
    bucket_url = config.GsBucketURL(bucket_name)
    auth_token, _ = app_identity.get_access_token(scope)
    result = urlfetch.fetch(bucket_url, method=urlfetch.HEAD, headers={
        'Authorization': 'OAuth %s' % auth_token,
        'x-goog-api-version': '2'})
    return result and result.status_code == 200
  
  
  def verify_bucket_writable(bucket_name, account_id=None):
    """Verify the application can write to the specified bucket.
  
    Args:
      bucket_name: The bucket to verify.
  
    Raises:
      BackupValidationError: If the bucket is not writable.
    """
    path = '/gs/%s/%s' % (bucket_name, TEST_WRITE_FILENAME_PREFIX)
    try:
      gcs_stats = GCSUtil.listbucket(path, max_keys=MAX_KEYS_LIST_SIZE,
                                     _account_id=account_id)
      file_names = [f.filename for f in gcs_stats]
    except (cloudstorage.AuthorizationError, cloudstorage.ForbiddenError):
      raise BackupValidationError('Bucket "%s" not accessible' % bucket_name)
    except cloudstorage.NotFoundError:
      raise BackupValidationError('Bucket "%s" does not exist' % bucket_name)
  
  
    file_name = '/gs/%s/%s.tmp' % (bucket_name, TEST_WRITE_FILENAME_PREFIX)
    file_name_try = 0
    while True:
      if file_name_try >= MAX_TEST_FILENAME_TRIES:
  
  
        return
      if file_name not in file_names:
        break
      gen = random.randint(0, 9999)
      file_name = ('/gs/%s/%s_%s.tmp' %
                   (bucket_name, TEST_WRITE_FILENAME_PREFIX, gen))
      file_name_try += 1
    try:
      with GCSUtil.open(file_name, 'w', _account_id=account_id) as f:
        f.write('test')
    except cloudstorage.ForbiddenError:
      raise BackupValidationError('Bucket "%s" is not writable' % bucket_name)
    try:
      GCSUtil.delete(file_name, _account_id=account_id)
    except cloudstorage.Error:
      logging.warn('Failed to delete test file %s', file_name)
  
  
  def is_readable_gs_handle(gs_handle):
    """Return True if the application can read the specified gs_handle."""
    try:
      with GCSUtil.open(gs_handle) as bak_file:
        bak_file.read(1)
    except (cloudstorage.ForbiddenError,
            cloudstorage.NotFoundError,
            cloudstorage.AuthorizationError):
      return False
    return True
  
  
  
  def parse_gs_handle(gs_handle):
    """Splits [/gs/]?bucket_name[/folder]*[/file]? to (bucket_name, path | '')."""
    if gs_handle.startswith('/'):
      filesystem = gs_handle[1:].split('/', 1)[0]
      if filesystem == 'gs':
        gs_handle = gs_handle[4:]
      else:
        raise BackupValidationError('Unsupported filesystem: %s' % filesystem)
    tokens = gs_handle.split('/', 1)
    return (tokens[0], '') if len(tokens) == 1 else tuple(tokens)
  
  
  def validate_and_split_gcs_path(gcs_path, account_id=None):
    bucket_name, path = parse_gs_handle(gcs_path)
    path = path.rstrip('/')
    validate_gcs_bucket_name(bucket_name)
    verify_bucket_writable(bucket_name, account_id)
    return bucket_name, path
  
  
  def list_bucket_files(bucket_name, prefix, max_keys=1000):
    """Returns a listing of of a bucket that matches the given prefix."""
    scope = config.GoogleApiScope('devstorage.read_only')
    bucket_url = config.GsBucketURL(bucket_name)
    url = bucket_url + '?'
    query = [('max-keys', max_keys)]
    if prefix:
      query.append(('prefix', prefix))
    url += urllib.urlencode(query)
    auth_token, _ = app_identity.get_access_token(scope)
    result = urlfetch.fetch(url, method=urlfetch.GET, headers={
        'Authorization': 'OAuth %s' % auth_token,
        'x-goog-api-version': '2'})
    if result and result.status_code == 200:
      doc = xml.dom.minidom.parseString(result.content)
      return [node.childNodes[0].data for node in doc.getElementsByTagName('Key')]
    raise BackupValidationError('Request to Google Cloud Storage failed')
  
  
  def get_gs_object(bucket_name, path):
    """Returns a listing of of a bucket that matches the given prefix."""
    scope = config.GoogleApiScope('devstorage.read_only')
    bucket_url = config.GsBucketURL(bucket_name)
    url = bucket_url + path
    auth_token, _ = app_identity.get_access_token(scope)
    result = urlfetch.fetch(url, method=urlfetch.GET, headers={
        'Authorization': 'OAuth %s' % auth_token,
        'x-goog-api-version': '2'})
    if result and result.status_code == 200:
      return result.content
    if result and result.status_code == 403:
      raise BackupValidationError(
          'Requested path %s is not accessible/access denied' % url)
    if result and result.status_code == 404:
      raise BackupValidationError('Requested path %s was not found' % url)
    raise BackupValidationError('Error encountered accessing requested path %s' %
                                url)
  
  
  
  
  def get_queue_names(app_id=None, max_rows=100):
    """Returns a list with all non-special queue names for app_id."""
    rpc = apiproxy_stub_map.UserRPC('taskqueue')
    request = taskqueue_service_pb.TaskQueueFetchQueuesRequest()
    response = taskqueue_service_pb.TaskQueueFetchQueuesResponse()
    if app_id:
      request.set_app_id(app_id)
    request.set_max_rows(max_rows)
    queues = ['default']
    try:
      rpc.make_call('FetchQueues', request, response)
      rpc.check_success()
  
      for queue in response.queue_list():
        if (queue.mode() == taskqueue_service_pb.TaskQueueMode.PUSH and
            not queue.queue_name().startswith('__') and
            queue.queue_name() != 'default'):
          queues.append(queue.queue_name())
    except Exception:
      logging.exception('Failed to get queue names.')
    return queues
  
  
  def handlers_list(base_path):
    return [
        (r'%s/%s' % (base_path, BackupLinkHandler.SUFFIX),
         BackupLinkHandler),
        (r'%s/%s' % (base_path, BackupImportAndRestoreLinkHandler.SUFFIX),
         BackupImportAndRestoreLinkHandler),
        (r'%s/%s' % (base_path, ConfirmBackupHandler.SUFFIX),
         ConfirmBackupHandler),
        (r'%s/%s' % (base_path, DoBackupHandler.SUFFIX), DoBackupHandler),
        (r'%s/%s' % (base_path, DoBackupRestoreHandler.SUFFIX),
         DoBackupRestoreHandler),
        (r'%s/%s' % (base_path, DoBackupDeleteHandler.SUFFIX),
         DoBackupDeleteHandler),
        (r'%s/%s' % (base_path, DoBackupAbortHandler.SUFFIX),
         DoBackupAbortHandler),
        (r'%s/%s' % (base_path, DoBackupImportHandler.SUFFIX),
         DoBackupImportHandler),
    ]
  
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.
  #
  
  
  
  
  """Main module for datastore admin receiver.
  
  To use, add this to app.yaml:
    builtins:
    - datastore_admin: on
  """
  
  
  
  import logging
  import operator
  import os
  
  from google.appengine.api import app_identity
  from google.appengine.api import datastore_errors
  from google.appengine.api import users
  from google.appengine.ext import deferred
  import webapp2 as webapp
  from google.appengine.ext.db import stats
  from google.appengine.ext.webapp import util
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext.datastore_admin import backup_handler
  from google.appengine.ext.datastore_admin import config
  from google.appengine.ext.datastore_admin import delete_handler
  from google.appengine.ext.datastore_admin import utils
  
  
  
  
  
  
  ENTITY_ACTIONS = {
      'Delete Entities': delete_handler.ConfirmDeleteHandler.Render,
      'Backup Entities': backup_handler.ConfirmBackupHandler.Render,
  }
  
  
  BACKUP_ACTIONS = {
      'Delete': backup_handler.ConfirmDeleteBackupHandler.Render,
      'Restore': backup_handler.ConfirmRestoreFromBackupHandler.Render,
      'Info': backup_handler.BackupInformationHandler.Render,
  }
  
  PENDING_BACKUP_ACTIONS = {
      'Abort': backup_handler.ConfirmAbortBackupHandler.Render,
      'Info': backup_handler.BackupInformationHandler.Render,
  }
  
  GET_ACTIONS = ENTITY_ACTIONS.copy()
  GET_ACTIONS.update(BACKUP_ACTIONS)
  GET_ACTIONS.update(PENDING_BACKUP_ACTIONS)
  GET_ACTIONS.update({'Import Backup Information':
                      backup_handler.ConfirmBackupImportHandler.Render})
  
  
  def _GetDatastoreStats(kinds_list, use_stats_kinds=False):
    """Retrieves stats for kinds.
  
    Args:
      kinds_list: List of known kinds.
      use_stats_kinds: If stats are available, kinds_list will be ignored and
        all kinds found in stats will be used instead.
  
    Returns:
      timestamp: records time that statistics were last updated.
      kind_dict: dictionary of kind objects with the following members:
      - kind_name: the name of this kind.
      - count: number of known entities of this type.
      - total_bytes_str: total bytes for this kind as a string.
      - average_bytes_str: average bytes per entity as a string.
    """
    global_stat = stats.GlobalStat.all().fetch(1)
    if not global_stat:
      return _KindsListToTuple(kinds_list)
  
    global_ts = global_stat[0].timestamp
  
    kind_stats = stats.KindStat.all().filter('timestamp =', global_ts).fetch(1000)
    if not kind_stats:
      return _KindsListToTuple(kinds_list)
  
    results = {}
    for kind_ent in kind_stats:
  
  
  
      if (not kind_ent.kind_name.startswith('__')
          and (use_stats_kinds or kind_ent.kind_name in kinds_list)
          and kind_ent.count > 0):
        results[kind_ent.kind_name] = _PresentatableKindStats(kind_ent)
  
    utils.CacheStats(results.values())
  
  
  
    for kind_str in kinds_list or []:
      if kind_str not in results:
        results[kind_str] = {'kind_name': kind_str}
  
    return (global_ts,
            sorted(results.values(), key=lambda x: x['kind_name']))
  
  
  def _KindsListToTuple(kinds_list):
    """Build default tuple when no datastore statistics are available. """
    return '', [{'kind_name': kind} for kind in sorted(kinds_list)]
  
  
  def _PresentatableKindStats(kind_ent):
    """Generate dict of presentable values for template."""
    count = kind_ent.count
    entity_bytes = kind_ent.entity_bytes
    total_bytes = kind_ent.bytes
    average_bytes = entity_bytes / count
    return {'kind_name': kind_ent.kind_name,
            'count': utils.FormatThousands(kind_ent.count),
            'entity_bytes_str': utils.GetPrettyBytes(entity_bytes),
            'entity_bytes': entity_bytes,
            'total_bytes_str': utils.GetPrettyBytes(total_bytes),
            'total_bytes': total_bytes,
            'average_bytes_str': utils.GetPrettyBytes(average_bytes),
           }
  
  
  class RouteByActionHandler(webapp.RequestHandler):
    """Route to the appropriate handler based on the action parameter."""
  
    def ListActions(self, error=None):
      """Handler for get requests to datastore_admin/confirm_delete."""
      use_stats_kinds = False
      kinds = []
      more_kinds = False
      try:
        kinds, more_kinds = utils.GetKinds()
        if not kinds:
          use_stats_kinds = True
          logging.warning('Found no kinds. Using datastore stats instead.')
      except datastore_errors.Error, e:
        logging.exception(e)
        use_stats_kinds = True
  
      last_stats_update, kind_stats = _GetDatastoreStats(
          kinds, use_stats_kinds=use_stats_kinds)
  
      template_params = {
          'run_as_a_service': self.request.get('run_as_a_service'),
          'datastore_admin_home': utils.GenerateHomeUrl(None),
          'offer_service': (self.request.get('service') and not
                            self.request.get('run_as_a_service')),
          'kind_stats': kind_stats,
          'more_kinds': more_kinds,
          'last_stats_update': last_stats_update,
          'app_id': self.request.get('app_id'),
          'hosting_app_id': app_identity.get_application_id(),
          'has_namespace': self.request.get('namespace', None) is not None,
          'namespace': self.request.get('namespace'),
          'action_list': sorted(ENTITY_ACTIONS.keys()),
          'backup_action_list': sorted(BACKUP_ACTIONS.keys()),
          'pending_backup_action_list': sorted(PENDING_BACKUP_ACTIONS.keys()),
          'error': error,
          'completed_operations': self.GetOperations(active=False),
          'active_operations': self.GetOperations(active=True),
          'pending_backups': self.GetPendingBackups(),
          'backups': self.GetBackups(),
          'map_reduce_path': config.MAPREDUCE_PATH + '/detail',
          'service_accounts': utils.get_service_account_names()
      }
      utils.RenderToResponse(self, 'list_actions.html', template_params)
  
    def RouteAction(self, action_dict):
      action = self.request.get('action')
      if not action:
        self.ListActions(error=self.request.get('error', None))
      elif action not in action_dict:
        error = '%s is not a valid action.' % action
        self.ListActions(error=error)
      else:
        action_dict[action](self)
  
    def get(self):
      self.RouteAction(GET_ACTIONS)
  
    def post(self):
      self.RouteAction(GET_ACTIONS)
  
    def GetOperations(self, active=False, limit=100):
      """Obtain a list of operation, ordered by last_updated."""
      query = utils.DatastoreAdminOperation.all()
      if active:
        query.filter('status = ', utils.DatastoreAdminOperation.STATUS_ACTIVE)
      else:
        query.filter('status IN ', [
            utils.DatastoreAdminOperation.STATUS_COMPLETED,
            utils.DatastoreAdminOperation.STATUS_FAILED,
            utils.DatastoreAdminOperation.STATUS_ABORTED])
      operations = query.fetch(max(10000, limit) if limit else 1000)
      operations = sorted(operations, key=operator.attrgetter('last_updated'),
                          reverse=True)
      return operations[:limit]
  
    def GetBackups(self, limit=100, deadline=10):
      """Obtain a list of backups.
  
      Args:
        limit: maximum number of backup records to retrieve.
        deadline: maximum number of seconds to spend getting backups.
  
      Returns:
        List of backups, sorted in reverse order by completion time.
      """
      backups = []
      query = backup_handler.BackupInformation.all()
      query.filter('complete_time > ', 0)
      query.order('-complete_time')
      try:
        backups.extend(query.run(deadline=deadline, limit=limit))
      except (datastore_errors.Timeout, apiproxy_errors.DeadlineExceededError):
        logging.warning('Failed to retrieve all backups within deadline.')
      return backups
  
    def GetPendingBackups(self, limit=100):
      """Obtain a list of pending backups."""
      query = backup_handler.BackupInformation.all()
      query.filter('complete_time = ', None)
      backups = query.fetch(max(10000, limit) if limit else 1000)
      backups = sorted(backups, key=operator.attrgetter('start_time'),
                       reverse=True)
      return backups[:limit]
  
  
  class StaticResourceHandler(webapp.RequestHandler):
    """Read static files from disk."""
  
  
  
  
  
  
    _BASE_FILE_PATH = os.path.dirname(__file__)
  
    _RESOURCE_MAP = {
        'static/js/compiled.js': 'text/javascript',
        'static/css/compiled.css': 'text/css',
        'static/img/help.gif': 'image/gif',
        'static/img/tip.png': 'image/png',
        'static/img/icn/icn-warning.gif': 'image/gif',
    }
  
    def get(self):
      relative_path = self.request.path.split(config.BASE_PATH + '/')[1]
      if relative_path not in self._RESOURCE_MAP:
        self.response.set_status(404)
        self.response.out.write('Resource not found.')
        return
  
      path = os.path.join(self._BASE_FILE_PATH, relative_path)
      self.response.headers['Cache-Control'] = 'public; max-age=300'
      self.response.headers['Content-Type'] = self._RESOURCE_MAP[relative_path]
      if relative_path == 'static/css/compiled.css':
  
  
        self.response.out.write(
            open(path).read().replace('url(/img/', 'url(../img/'))
      else:
        self.response.out.write(open(path).read())
  
  
  class LoginRequiredHandler(webapp.RequestHandler):
    """Handle federated login identity selector page."""
  
    def get(self):
      target = self.request.get('continue')
      if not target:
        self.error(400)
        return
  
  
      login_url = users.create_login_url(target)
      self.redirect(login_url)
  
  
  def CreateApplication():
    """Create new WSGIApplication and register all handlers.
  
    Returns:
      an instance of webapp.WSGIApplication with all mapreduce handlers
      registered.
    """
    return webapp.WSGIApplication(
        backup_handler.handlers_list(config.BASE_PATH) +
        [(r'%s/%s' % (config.BASE_PATH,
                      delete_handler.ConfirmDeleteHandler.SUFFIX),
          delete_handler.ConfirmDeleteHandler),
         (r'%s/%s' % (config.BASE_PATH, delete_handler.DoDeleteHandler.SUFFIX),
          delete_handler.DoDeleteHandler),
         (r'%s/%s' % (config.BASE_PATH, utils.MapreduceDoneHandler.SUFFIX),
          utils.MapreduceDoneHandler),
         (config.DEFERRED_PATH, deferred.TaskHandler),
         (r'%s/static.*' % config.BASE_PATH, StaticResourceHandler),
         (r'/_ah/login_required', LoginRequiredHandler),
         (r'.*', RouteByActionHandler)])
  
  
  APP = CreateApplication()
  
  
  def main():
    util.run_wsgi_app(APP)
  
  
  if __name__ == '__main__':
    main()
  
files: AppServer/google/appengine/ext/datastore_admin/main.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.
  #
  
  
  
  
  """Main module for datastore admin receiver.
  
  To use, add this to app.yaml:
    builtins:
    - datastore_admin: on
  """
  
  
  
  import logging
  import operator
  import os
  
  from google.appengine.api import app_identity
  from google.appengine.api import datastore_errors
  from google.appengine.api import users
  from google.appengine.ext import deferred
  import webapp2 as webapp
  from google.appengine.ext.db import stats
  from google.appengine.ext.webapp import util
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext.datastore_admin import backup_handler
  from google.appengine.ext.datastore_admin import config
  from google.appengine.ext.datastore_admin import delete_handler
  from google.appengine.ext.datastore_admin import utils
  
  
  
  
  
  
  ENTITY_ACTIONS = {
      'Delete Entities': delete_handler.ConfirmDeleteHandler.Render,
      'Backup Entities': backup_handler.ConfirmBackupHandler.Render,
  }
  
  
  BACKUP_ACTIONS = {
      'Delete': backup_handler.ConfirmDeleteBackupHandler.Render,
      'Restore': backup_handler.ConfirmRestoreFromBackupHandler.Render,
      'Info': backup_handler.BackupInformationHandler.Render,
  }
  
  PENDING_BACKUP_ACTIONS = {
      'Abort': backup_handler.ConfirmAbortBackupHandler.Render,
      'Info': backup_handler.BackupInformationHandler.Render,
  }
  
  GET_ACTIONS = ENTITY_ACTIONS.copy()
  GET_ACTIONS.update(BACKUP_ACTIONS)
  GET_ACTIONS.update(PENDING_BACKUP_ACTIONS)
  GET_ACTIONS.update({'Import Backup Information':
                      backup_handler.ConfirmBackupImportHandler.Render})
  
  
  def _GetDatastoreStats(kinds_list, use_stats_kinds=False):
    """Retrieves stats for kinds.
  
    Args:
      kinds_list: List of known kinds.
      use_stats_kinds: If stats are available, kinds_list will be ignored and
        all kinds found in stats will be used instead.
  
    Returns:
      timestamp: records time that statistics were last updated.
      kind_dict: dictionary of kind objects with the following members:
      - kind_name: the name of this kind.
      - count: number of known entities of this type.
      - total_bytes_str: total bytes for this kind as a string.
      - average_bytes_str: average bytes per entity as a string.
    """
    global_stat = stats.GlobalStat.all().fetch(1)
    if not global_stat:
      return _KindsListToTuple(kinds_list)
  
    global_ts = global_stat[0].timestamp
  
    kind_stats = stats.KindStat.all().filter('timestamp =', global_ts).fetch(1000)
    if not kind_stats:
      return _KindsListToTuple(kinds_list)
  
    results = {}
    for kind_ent in kind_stats:
  
  
  
      if (not kind_ent.kind_name.startswith('__')
          and (use_stats_kinds or kind_ent.kind_name in kinds_list)
          and kind_ent.count > 0):
        results[kind_ent.kind_name] = _PresentatableKindStats(kind_ent)
  
    utils.CacheStats(results.values())
  
  
  
    for kind_str in kinds_list or []:
      if kind_str not in results:
        results[kind_str] = {'kind_name': kind_str}
  
    return (global_ts,
            sorted(results.values(), key=lambda x: x['kind_name']))
  
  
  def _KindsListToTuple(kinds_list):
    """Build default tuple when no datastore statistics are available. """
    return '', [{'kind_name': kind} for kind in sorted(kinds_list)]
  
  
  def _PresentatableKindStats(kind_ent):
    """Generate dict of presentable values for template."""
    count = kind_ent.count
    entity_bytes = kind_ent.entity_bytes
    total_bytes = kind_ent.bytes
    average_bytes = entity_bytes / count
    return {'kind_name': kind_ent.kind_name,
            'count': utils.FormatThousands(kind_ent.count),
            'entity_bytes_str': utils.GetPrettyBytes(entity_bytes),
            'entity_bytes': entity_bytes,
            'total_bytes_str': utils.GetPrettyBytes(total_bytes),
            'total_bytes': total_bytes,
            'average_bytes_str': utils.GetPrettyBytes(average_bytes),
           }
  
  
  class RouteByActionHandler(webapp.RequestHandler):
    """Route to the appropriate handler based on the action parameter."""
  
    def ListActions(self, error=None):
      """Handler for get requests to datastore_admin/confirm_delete."""
      use_stats_kinds = False
      kinds = []
      more_kinds = False
      try:
        kinds, more_kinds = utils.GetKinds()
        if not kinds:
          use_stats_kinds = True
          logging.warning('Found no kinds. Using datastore stats instead.')
      except datastore_errors.Error, e:
        logging.exception(e)
        use_stats_kinds = True
  
      last_stats_update, kind_stats = _GetDatastoreStats(
          kinds, use_stats_kinds=use_stats_kinds)
  
      template_params = {
          'run_as_a_service': self.request.get('run_as_a_service'),
          'datastore_admin_home': utils.GenerateHomeUrl(None),
          'offer_service': (self.request.get('service') and not
                            self.request.get('run_as_a_service')),
          'kind_stats': kind_stats,
          'more_kinds': more_kinds,
          'last_stats_update': last_stats_update,
          'app_id': self.request.get('app_id'),
          'hosting_app_id': app_identity.get_application_id(),
          'has_namespace': self.request.get('namespace', None) is not None,
          'namespace': self.request.get('namespace'),
          'action_list': sorted(ENTITY_ACTIONS.keys()),
          'backup_action_list': sorted(BACKUP_ACTIONS.keys()),
          'pending_backup_action_list': sorted(PENDING_BACKUP_ACTIONS.keys()),
          'error': error,
          'completed_operations': self.GetOperations(active=False),
          'active_operations': self.GetOperations(active=True),
          'pending_backups': self.GetPendingBackups(),
          'backups': self.GetBackups(),
          'map_reduce_path': config.MAPREDUCE_PATH + '/detail'
      }
      utils.RenderToResponse(self, 'list_actions.html', template_params)
  
    def RouteAction(self, action_dict):
      action = self.request.get('action')
      if not action:
        self.ListActions(error=self.request.get('error', None))
      elif action not in action_dict:
        error = '%s is not a valid action.' % action
        self.ListActions(error=error)
      else:
        action_dict[action](self)
  
    def get(self):
      self.RouteAction(GET_ACTIONS)
  
    def post(self):
      self.RouteAction(GET_ACTIONS)
  
    def GetOperations(self, active=False, limit=100):
      """Obtain a list of operation, ordered by last_updated."""
      query = utils.DatastoreAdminOperation.all()
      if active:
        query.filter('status = ', utils.DatastoreAdminOperation.STATUS_ACTIVE)
      else:
        query.filter('status IN ', [
            utils.DatastoreAdminOperation.STATUS_COMPLETED,
            utils.DatastoreAdminOperation.STATUS_FAILED,
            utils.DatastoreAdminOperation.STATUS_ABORTED])
      operations = query.fetch(max(10000, limit) if limit else 1000)
      operations = sorted(operations, key=operator.attrgetter('last_updated'),
                          reverse=True)
      return operations[:limit]
  
    def GetBackups(self, limit=100, deadline=10):
      """Obtain a list of backups.
  
      Args:
        limit: maximum number of backup records to retrieve.
        deadline: maximum number of seconds to spend getting backups.
  
      Returns:
        List of backups, sorted in reverse order by completion time.
      """
      backups = []
      query = backup_handler.BackupInformation.all()
      query.filter('complete_time > ', 0)
      query.order('-complete_time')
      try:
        backups.extend(query.run(deadline=deadline, limit=limit))
      except (datastore_errors.Timeout, apiproxy_errors.DeadlineExceededError):
        logging.warning('Failed to retrieve all backups within deadline.')
      return backups
  
    def GetPendingBackups(self, limit=100):
      """Obtain a list of pending backups."""
      query = backup_handler.BackupInformation.all()
      query.filter('complete_time = ', None)
      backups = query.fetch(max(10000, limit) if limit else 1000)
      backups = sorted(backups, key=operator.attrgetter('start_time'),
                       reverse=True)
      return backups[:limit]
  
  
  class StaticResourceHandler(webapp.RequestHandler):
    """Read static files from disk."""
  
  
  
  
  
  
    _BASE_FILE_PATH = os.path.dirname(__file__)
  
    _RESOURCE_MAP = {
        'static/js/compiled.js': 'text/javascript',
        'static/css/compiled.css': 'text/css',
        'static/img/help.gif': 'image/gif',
        'static/img/tip.png': 'image/png',
        'static/img/icn/icn-warning.gif': 'image/gif',
    }
  
    def get(self):
      relative_path = self.request.path.split(config.BASE_PATH + '/')[1]
      if relative_path not in self._RESOURCE_MAP:
        self.response.set_status(404)
        self.response.out.write('Resource not found.')
        return
  
      path = os.path.join(self._BASE_FILE_PATH, relative_path)
      self.response.headers['Cache-Control'] = 'public; max-age=300'
      self.response.headers['Content-Type'] = self._RESOURCE_MAP[relative_path]
      if relative_path == 'static/css/compiled.css':
  
  
        self.response.out.write(
            open(path).read().replace('url(/img/', 'url(../img/'))
      else:
        self.response.out.write(open(path).read())
  
  
  class LoginRequiredHandler(webapp.RequestHandler):
    """Handle federated login identity selector page."""
  
    def get(self):
      target = self.request.get('continue')
      if not target:
        self.error(400)
        return
  
  
      login_url = users.create_login_url(target)
      self.redirect(login_url)
  
  
  def CreateApplication():
    """Create new WSGIApplication and register all handlers.
  
    Returns:
      an instance of webapp.WSGIApplication with all mapreduce handlers
      registered.
    """
    return webapp.WSGIApplication(
        backup_handler.handlers_list(config.BASE_PATH) +
        [(r'%s/%s' % (config.BASE_PATH,
                      delete_handler.ConfirmDeleteHandler.SUFFIX),
          delete_handler.ConfirmDeleteHandler),
         (r'%s/%s' % (config.BASE_PATH, delete_handler.DoDeleteHandler.SUFFIX),
          delete_handler.DoDeleteHandler),
         (r'%s/%s' % (config.BASE_PATH, utils.MapreduceDoneHandler.SUFFIX),
          utils.MapreduceDoneHandler),
         (config.DEFERRED_PATH, deferred.TaskHandler),
         (r'%s/static.*' % config.BASE_PATH, StaticResourceHandler),
         (r'/_ah/login_required', LoginRequiredHandler),
         (r'.*', RouteByActionHandler)])
  
  
  APP = CreateApplication()
  
  
  def main():
    util.run_wsgi_app(APP)
  
  
  if __name__ == '__main__':
    main()
  
files: AppServer/google/appengine/ext/datastore_admin/main.py
files: {% extends "base.html" %}
  {% block title %}Confirm Import of backup information{% endblock %}
  {% block body %}
    <h2>Datastore Admin: Import backup information</h2>
    {% if error %}
      <div class="ae-errorbox">{{ error|escape }}</div>
      <a href="{{ datastore_admin_home }}"><button>Back</button></a>
    {% else %}
      {% if not other_backup_info_files and not selected_backup_info_file %}
        <div class="ae-errorbox">No backup information files were found.</div>
        <a href="{{ datastore_admin_home }}"><button>Back</button></a>
      {% else %}
        <form action="{{base_path}}/{{form_target}}" method="post" style="width:39.39em;">
          <input type="hidden" name="xsrf_token" value="{{xsrf_token|escape}}">
          {% if not other_backup_info_files %}
            <b>Confirm import of:</b>
            <input type="hidden" name="gs_handle" value="{{selected_backup_info_file}}">{{selected_backup_info_file|escape}}
          {% else %}
            {% if selected_backup_info_file %}
              <b>Confirm import of:</b><br/>
              <input type="radio" name="gs_handle" value="{{selected_backup_info_file}}" checked>{{selected_backup_info_file|escape}}</br>
            {% else %}
              {% if backup_info_specified %}
                <b>Requested backup was not found, available backups:</b><br/>
              {% else %}
                <b>Select an available backup:</b><br/>
              {% endif %}
            {% endif %}
            {% for backup_info_file in other_backup_info_files %}
              <input type="radio" name="gs_handle" value="{{backup_info_file}}">{{backup_info_file|escape}}</br>
            {% endfor %}
          {% endif %}
          {% if run_as_a_service %}
            <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
          {% endif %}
          <input type="hidden" name="service_account_name" value="{{ service_account_name }}">
          <input type="Submit" name="Import" value="Add to backup list">
          <input type="Submit" name="Restore" value="Restore from backup">
          <a href="{{ datastore_admin_home }}">Cancel</a>
        </form>
      {% endif %}
    {% endif %}
  {% endblock %}
  
  
files: AppServer/google/appengine/ext/datastore_admin/templates/confirm_backup_import.html
files: {% extends "base.html" %}
  {% block title %}Confirm Import of backup information{% endblock %}
  {% block body %}
    <h2>Datastore Admin: Import backup information</h2>
    {% if error %}
      <div class="ae-errorbox">{{ error|escape }}</div>
      <a href="{{ datastore_admin_home }}"><button>Back</button></a>
    {% else %}
      {% if not other_backup_info_files and not selected_backup_info_file %}
        <div class="ae-errorbox">No backup information files were found.</div>
        <a href="{{ datastore_admin_home }}"><button>Back</button></a>
      {% else %}
        <form action="{{base_path}}/{{form_target}}" method="post" style="width:39.39em;">
          <input type="hidden" name="xsrf_token" value="{{xsrf_token|escape}}">
          {% if not other_backup_info_files %}
            <b>Confirm import of:</b>
            <input type="hidden" name="gs_handle" value="{{selected_backup_info_file}}">{{selected_backup_info_file|escape}}
          {% else %}
            {% if selected_backup_info_file %}
              <b>Confirm import of:</b><br/>
              <input type="radio" name="gs_handle" value="{{selected_backup_info_file}}" checked>{{selected_backup_info_file|escape}}</br>
            {% else %}
              {% if backup_info_specified %}
                <b>Requested backup was not found, available backups:</b><br/>
              {% else %}
                <b>Select an available backup:</b><br/>
              {% endif %}
            {% endif %}
            {% for backup_info_file in other_backup_info_files %}
              <input type="radio" name="gs_handle" value="{{backup_info_file}}">{{backup_info_file|escape}}</br>
            {% endfor %}
          {% endif %}
          {% if run_as_a_service %}
            <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
          {% endif %}
          <input type="Submit" name="Import" value="Add to backup list">
          <input type="Submit" name="Restore" value="Restore from backup">
          <a href="{{ datastore_admin_home }}">Cancel</a>
        </form>
      {% endif %}
    {% endif %}
  {% endblock %}
  
  
files: AppServer/google/appengine/ext/datastore_admin/templates/confirm_backup_import.html
files: {% extends "base.html" %}
  {% block title %}Confirm Restore from a backup copy{% endblock %}
  {% block body %}
    <h2>Datastore Admin: Restore from a backup copy</h2>
    {% if backup %}
      Are you sure you want to restore from {{ backup.name }}?
      <br/>Restore does not delete any new entities added since the backup.
      However, it will replace existing entities with the backed-up version for the following kinds:
      <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 %}
        <input type="hidden" name="backup_id" value="{{backup.key}}">
        <input type="hidden" name="xsrf_token" value="{{xsrf_token|escape}}">
        <input type="hidden" name="delete_backup_after_restore" value="{{ delete_backup_after_restore|escape }}">
        <ul>
          {% for kind in backup.kinds %}
            <li><input type="checkbox" name="kind" value="{{kind}}" checked="checked"/>{{ kind }}</li>
          {% endfor %}
        </ul>
        {% if notreadonly_warning %}
        <div>
          <h4 style="padding:0;margin-top:1em;margin-bottom:0.46em;">WARNING</h4>
          {% if original_app_warning %}
            <ul><li>
          {% else %}
            <p>
          {% endif %}
          The application is not in read-only mode and restore may
          conflict with other changes.
          To place the application in read-only mode,
          please follow the instructions at
          <a href="https://developers.google.com/appengine/docs/adminconsole/applicationsettings#Disable_Datastore_Writes" target="_blank">
          Disable Datastore Writes</a>
          {% if not original_app_warning %}
            </p>
            </div>
          {% endif %}
        {% endif %}
        {% if original_app_warning %}
          {% if not notreadonly_warning %}
            <div>
            <h4 style="padding:0;margin-top:1em;margin-bottom:0.46em;">WARNING</h4>
            <p>
          {% else %}
            <li>
          {% endif %}
          The backup data comes from a different application, {{original_app_warning|escape}}.
          {% if not notreadonly_warning %}
            </p>
          {% else %}
            </li>
          {% endif %}
          </div>
        {% endif %}
        <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>
        <p>
          <label for="service_account_name">
            Service account name
            <img class="ae-help-icon" src="{{ base_path }}/static/img/help.gif" height="14" width="14" alt="help"
                 title="You can define additional service accounts using the dashboard.">
          </label>
          <select id="service_account_name" name="service_account_name">
            <option value="">--Choose an option--</option>
            {% for account_name in service_accounts %}
            <option value="{{ account_name }}">{{ account_name }}</option>
            {% endfor %}
          </select>
        </p>
        <input type="submit" value="Restore">
        <a href="{{ datastore_admin_home }}">Cancel</a>
      </form>
    {% else %}
      <div class="ae-errorbox">Backup was not found.</div>
    {% endif %}
  {% endblock %}
  
files: AppServer/google/appengine/ext/datastore_admin/templates/confirm_restore_from_backup.html
files: {% extends "base.html" %}
  {% block title %}Confirm Restore from a backup copy{% endblock %}
  {% block body %}
    <h2>Datastore Admin: Restore from a backup copy</h2>
    {% if backup %}
      Are you sure you want to restore from {{ backup.name }}?
      <br/>Restore does not delete any new entities added since the backup.
      However, it will replace existing entities with the backed-up version for the following kinds:
      <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 %}
        <input type="hidden" name="backup_id" value="{{backup.key}}">
        <input type="hidden" name="xsrf_token" value="{{xsrf_token|escape}}">
        <input type="hidden" name="delete_backup_after_restore" value="{{ delete_backup_after_restore|escape }}">
        <ul>
          {% for kind in backup.kinds %}
            <li><input type="checkbox" name="kind" value="{{kind}}" checked="checked"/>{{ kind }}</li>
          {% endfor %}
        </ul>
        {% if notreadonly_warning %}
        <div>
          <h4 style="padding:0;margin-top:1em;margin-bottom:0.46em;">WARNING</h4>
          {% if original_app_warning %}
            <ul><li>
          {% else %}
            <p>
          {% endif %}
          The application is not in read-only mode and restore may
          conflict with other changes.
          To place the application in read-only mode,
          please follow the instructions at
          <a href="https://developers.google.com/appengine/docs/adminconsole/applicationsettings#Disable_Datastore_Writes" target="_blank">
          Disable Datastore Writes</a>
          {% if not original_app_warning %}
            </p>
            </div>
          {% endif %}
        {% endif %}
        {% if original_app_warning %}
          {% if not notreadonly_warning %}
            <div>
            <h4 style="padding:0;margin-top:1em;margin-bottom:0.46em;">WARNING</h4>
            <p>
          {% else %}
            <li>
          {% endif %}
          The backup data comes from a different application, {{original_app_warning|escape}}.
          {% if not notreadonly_warning %}
            </p>
          {% else %}
            </li>
          {% endif %}
          </div>
        {% endif %}
        <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>
        <input type="submit" value="Restore">
        <a href="{{ datastore_admin_home }}">Cancel</a>
      </form>
    {% else %}
      <div class="ae-errorbox">Backup was not found.</div>
    {% endif %}
  {% endblock %}
  
files: AppServer/google/appengine/ext/datastore_admin/templates/confirm_restore_from_backup.html
files: {% extends "base.html" %}
  {% block body %}
    <h2>Datastore Admin of {{hosting_app_id}}</h2>
    {% if run_as_a_service or offer_service %}
      <div class="ae-promo ae-rounded-sml">
        <strong>
          The managed backup service is deprecated, and will no longer be available as of
          Mar 27th, 2018.
        </strong>
      </div>
    {% endif %}
    {% if run_as_a_service %}
      <div class="ae-message ae-rounded-sml">
        Backup/Restore running as a service.
        <a href="{{datastore_admin_home}}?service=T">Disable</a>
      </div>
    {% else %}
      {% if offer_service %}
        <div class="ae-message ae-rounded-sml">
          This application may run backup/restore as a service.
          <a href="{{datastore_admin_home}}?run_as_a_service=True">Enable</a>
        </div>
      {% endif %}
    {% endif %}
    {% if error %}
    <div class="ae-errorbox">
      {{ error|escape }}
    </div>
    {% endif %}
    <div class="ae-table-caption">
      <div class="ae-paginate">
        <div class="g-section g-tpl-50-50 g-split">
          <div class="g-unit g-first">
            <strong>Entities</strong>
          </div>
          <div class="g-unit">
            <span id="ae-dash-quota-refresh-info">
            {% if last_stats_update %}
              Entity statistics last updated
              {{ last_stats_update|date:"M j, Y g:i a T" }}
            {% else %}
              Entity statistics are temporarily unavailable and should appear within 48 hours.
            {% endif %}
            <a target="_blank"
               href="https://cloud.google.com/appengine/docs/standard/python/console/datastore-statistics">
              <img class="ae-help-icon" src="{{ base_path }}/static/img/help.gif"
                height="14" width="14" alt="help">
            </a>
            </span>
          </div>
        </div>
      </div>
    </div>
    {% if kind_stats %}
    <form action="{{ base_path }}/" method="post">
      {% if run_as_a_service %}
        <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
      {% endif %}
      {% if has_namespace %}
      <input type="hidden" name="namespace" value="{{ namespace|escape }}">
      {% endif %}
      <input type="hidden" name="app_id" value="{{ app_id|escape }}">
      <table id="kinds" class="ae-table ae-table-striped">
        <thead>
          <tr>
            <th scope="col" class="cbc tct-selectall"></th>
            <th scope="col" class="id">Entity Kind</th>
            <th scope="col" class="id"># Entities</th>
            <th scope="col" class="id">Avg. Size/Entity</th>
            <th scope="col" class="id">Entities Size</th>
            <th scope="col" class="id">Total Size</th>
          </tr>
        </thead>
        <tbody>
          {% for kind in kind_stats %}
            <tr>
              <td class="cbc">
                <input type="checkbox" name="kind" value="{{ kind.kind_name }}"
                       id="{{ kind.kind_name }}">
              </td>
              <td class="id">
                <label for="{{ kind.kind_name }}">{{ kind.kind_name }}</label>
              </td>
              {% if kind.count %}
                <td class="id">{{ kind.count }}</td>
                <td class="id">{{ kind.average_bytes_str }}</td>
                <td class="id">{{ kind.entity_bytes_str }}</td>
                <td class="id">{{ kind.total_bytes_str }}</td>
              {% else %}
                <td class="id" style="color: #666667;">Stats not available</td>
                <td class="id" style="color: #666667;">Stats not available</td>
                <td class="id" style="color: #666667;">Stats not available</td>
                <td class="id" style="color: #666667;">Stats not available</td>
              {% endif %}
            </tr>
          {% endfor %}
          {% if more_kinds %}
            <tr>
              <td colspan="6" class="cbc">
                Not all kinds could be displayed, possibly because there were too many. To back up a kind that does not appear in this list, use a <a href="https://developers.google.com/appengine/articles/scheduled_backups" target="_blank">scheduled backup</a>.
              </td>
            </tr>
          {% endif %}
        </tbody>
        <tfoot>
          <tr>
            <td colspan="6">
              {% for action in action_list %}
                <input type="submit" name="action"
                  class="goog-button" value="{{ action }}">
              {% endfor %}
            </td>
          </tr>
        </tfoot>
      </table>
    </form>
    {% else %}
      <table class="ae-table">
        <tbody>
          <tr>
            <td>
              <div class="ae-disabled"
                  style="padding-top: 0.75em; padding-bottom: 0.75em;
                         text-align: center;">
                Datastore is empty.
              </div>
            </td>
          </tr>
        </tbody>
      </table>
    {% endif %}
    {% if pending_backups %}
      <p>
      <div class="ae-table-caption">
        <div class="ae-paginate">
          <div class="g-section g-tpl-50-50 g-split">
            <div class="g-unit g-first">
              <strong>Pending Backups</strong>
            </div>
          </div>
        </div>
      </div>
      <form action="{{ base_path }}/" method="post">
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        <input type="hidden" name="app_id" value="{{ app_id|escape }}">
        <table id="pending_backups" class="ae-table ae-table-striped">
          <thead>
            <tr>
              <th scope="col" class="cbc tct-selectall"></th>
              <th scope="col" class="id">Name</th>
              <th scope="col" class="id">Start-Time</th>
            </tr>
          </thead>
          <tbody>
            {% for pending_backup in pending_backups %}
              <tr>
                <td class="cbc">
                  <input type="checkbox" name="backup_id" value="{{ pending_backup.key }}" id="{{ pending_backup.key }}">
                </td>
                <td class="id">
                  <span title="Kinds: {{ pending_backup.kinds|join:', '|truncatewords:20|escape}}">{{ pending_backup.name }}{% if pending_backup.original_app %} (Imported from {{pending_backup.original_app|escape}}){% endif %}</span>
                </td>
                <td class="id">
                  {{ pending_backup.start_time }}
                </td>
              </tr>
            {% endfor %}
          </tbody>
          <tfoot>
            <tr>
              <td colspan="4" nowrap="nowrap">
                {% for pending_backup_action in pending_backup_action_list %}
                    <input type="submit" name="action" style="" class="goog-button" value="{{ pending_backup_action }}">
                {% endfor %}
              </td>
            </tr>
          </tfoot>
        </table>
      </form>
    {% endif %}
    <p>
    <div class="ae-table-caption">
      <div class="ae-paginate">
        <div class="g-section g-tpl-50-50 g-split">
          <div class="g-unit g-first">
            <strong>Backups</strong>
          </div>
        </div>
      </div>
    </div>
    {% if backups %}
      <form action="{{ base_path }}/" method="post">
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        <input type="hidden" name="app_id" value="{{ app_id|escape }}">
        <table id="backups" class="ae-table ae-table-striped">
          <thead>
            <tr>
              <th scope="col" class="cbc tct-selectall"></th>
              <th scope="col" class="id">Name</th>
              <th scope="col" class="id">Start-Time</th>
              <th scope="col" class="id">End-Time</th>
            </tr>
          </thead>
          <tbody>
            {% for backup in backups %}
              <tr>
                <td class="cbc">
                  <input type="checkbox" name="backup_id" value="{{ backup.key }}" id="{{ backup.key }}">
                </td>
                <td class="id">
                  <span title="Kinds: {{ backup.kinds|join:', '|truncatewords:20|escape }}">{{ backup.name }}{% if backup.original_app %} (Imported from {{backup.original_app|escape}}){% endif %}</span>
                </td>
                <td class="id">
                  {{ backup.start_time }}
                </td>
                <td class="id">
                  {{ backup.complete_time }}
                </td>
              </tr>
            {% endfor %}
          </tbody>
          <tfoot>
            <tr>
              <td colspan="4" nowrap="nowrap">
                {% for backup_action in backup_action_list %}
                    <input type="submit" name="action" style="" class="goog-button" value="{{ backup_action }}">
                {% endfor %}
              </td>
            </tr>
          </tfoot>
        </table>
      </form>
    {% endif %}
    <div class="ae-table-caption">
      <form action="{{ base_path }}/" method="post" >
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        <div style="display: block">
          <label for="service_account_name">
            Service account name
            <img class="ae-help-icon" src="{{ base_path }}/static/img/help.gif" height="14" width="14" alt="help"
                 title="You can define additional service accounts using the dashboard.">
          </label>
          <select id="service_account_name" name="service_account_name">
            <option value="">--Choose an option--</option>
            {% for account_name in service_accounts %}
            <option value="{{ account_name }}">{{ account_name }}</option>
            {% endfor %}
          </select>
        </div>
        <input type="text" name="gs_handle" value="" size="50" placeholder="Bucket or info file"
               title="Google Cloud Storage path of a backup info file or a bucket name"/>
        <input type="submit" name="action" value="Import Backup Information" title="Import Backup Information or restore from Google Cloud Storage"/>
      </form>
    </div>
    {% if active_operations %}
      <p>
      <div class="ae-table-caption">
        <div class="ae-paginate">
          <div class="g-section g-tpl-50-50 g-split">
            <div class="g-unit g-first">
              <strong>Active Operations</strong>
            </div>
          </div>
        </div>
      </div>
      <table id="operations" class="ae-table ae-table-striped">
        <thead>
          <tr>
            <th scope="col" class="id">Description</th>
            <th scope="col" class="id">Jobs</th>
          </tr>
        </thead>
        <tbody>
          {% for operation in active_operations %}
            <tr>
              <td class="id">
                {{ operation.description }}
              </td>
              <td class="id">
                ({{operation.completed_jobs}} steps completed,
                {{operation.active_jobs}} active)
                {% for job_id in operation.active_job_ids %}
                 <br/><a href="{{ map_reduce_path }}?mapreduce_id={{job_id}}">Job {{job_id}}</a>
                {% endfor %}
              </td>
            </tr>
          {% endfor %}
        </tbody>
      </table>
    {% endif %}
    {% if completed_operations %}
      <p>
      <div class="ae-table-caption">
        <div class="ae-paginate">
          <div class="g-section g-tpl-50-50 g-split">
            <div class="g-unit g-first">
              <strong>Completed Operations</strong>
            </div>
          </div>
        </div>
      </div>
      <table id="operations" class="ae-table ae-table-striped">
        <thead>
          <tr>
            <th scope="col" class="id">Description</th>
            <th scope="col" class="id">Completed steps</th>
            <th scope="col" class="id">Status</th>
            <th scope="col" class="id">Status info</th>
          </tr>
        </thead>
        <tbody>
          {% for operation in completed_operations %}
            <tr>
              <td class="id">
                {{ operation.description }}
              </td>
              <td class="id">
                {{operation.completed_jobs}} steps
              </td>
              <td class="id">
                {{operation.status}}
              </td>
              <td class="id">
                {{operation.status_info}}
              </td>
            </tr>
          {% endfor %}
        </tbody>
      </table>
    {% endif %}
  {% endblock %}
  
files: AppServer/google/appengine/ext/datastore_admin/templates/list_actions.html
files: {% extends "base.html" %}
  {% block body %}
    <h2>Datastore Admin of {{hosting_app_id}}</h2>
    {% if run_as_a_service or offer_service %}
      <div class="ae-promo ae-rounded-sml">
        <strong>
          The managed backup service is deprecated, and will no longer be available as of
          Mar 27th, 2018.
        </strong>
      </div>
    {% endif %}
    {% if run_as_a_service %}
      <div class="ae-message ae-rounded-sml">
        Backup/Restore running as a service.
        <a href="{{datastore_admin_home}}?service=T">Disable</a>
      </div>
    {% else %}
      {% if offer_service %}
        <div class="ae-message ae-rounded-sml">
          This application may run backup/restore as a service.
          <a href="{{datastore_admin_home}}?run_as_a_service=True">Enable</a>
        </div>
      {% endif %}
    {% endif %}
    {% if error %}
    <div class="ae-errorbox">
      {{ error|escape }}
    </div>
    {% endif %}
    <div class="ae-table-caption">
      <div class="ae-paginate">
        <div class="g-section g-tpl-50-50 g-split">
          <div class="g-unit g-first">
            <strong>Entities</strong>
          </div>
          <div class="g-unit">
            <span id="ae-dash-quota-refresh-info">
            {% if last_stats_update %}
              Entity statistics last updated
              {{ last_stats_update|date:"M j, Y g:i a T" }}
            {% else %}
              Entity statistics are temporarily unavailable and should appear within 48 hours.
            {% endif %}
            <a target="_blank"
               href="https://cloud.google.com/appengine/docs/standard/python/console/datastore-statistics">
              <img class="ae-help-icon" src="{{ base_path }}/static/img/help.gif"
                height="14" width="14" alt="help">
            </a>
            </span>
          </div>
        </div>
      </div>
    </div>
    {% if kind_stats %}
    <form action="{{ base_path }}/" method="post">
      {% if run_as_a_service %}
        <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
      {% endif %}
      {% if has_namespace %}
      <input type="hidden" name="namespace" value="{{ namespace|escape }}">
      {% endif %}
      <input type="hidden" name="app_id" value="{{ app_id|escape }}">
      <table id="kinds" class="ae-table ae-table-striped">
        <thead>
          <tr>
            <th scope="col" class="cbc tct-selectall"></th>
            <th scope="col" class="id">Entity Kind</th>
            <th scope="col" class="id"># Entities</th>
            <th scope="col" class="id">Avg. Size/Entity</th>
            <th scope="col" class="id">Entities Size</th>
            <th scope="col" class="id">Total Size</th>
          </tr>
        </thead>
        <tbody>
          {% for kind in kind_stats %}
            <tr>
              <td class="cbc">
                <input type="checkbox" name="kind" value="{{ kind.kind_name }}"
                       id="{{ kind.kind_name }}">
              </td>
              <td class="id">
                <label for="{{ kind.kind_name }}">{{ kind.kind_name }}</label>
              </td>
              {% if kind.count %}
                <td class="id">{{ kind.count }}</td>
                <td class="id">{{ kind.average_bytes_str }}</td>
                <td class="id">{{ kind.entity_bytes_str }}</td>
                <td class="id">{{ kind.total_bytes_str }}</td>
              {% else %}
                <td class="id" style="color: #666667;">Stats not available</td>
                <td class="id" style="color: #666667;">Stats not available</td>
                <td class="id" style="color: #666667;">Stats not available</td>
                <td class="id" style="color: #666667;">Stats not available</td>
              {% endif %}
            </tr>
          {% endfor %}
          {% if more_kinds %}
            <tr>
              <td colspan="6" class="cbc">
                Not all kinds could be displayed, possibly because there were too many. To back up a kind that does not appear in this list, use a <a href="https://developers.google.com/appengine/articles/scheduled_backups" target="_blank">scheduled backup</a>.
              </td>
            </tr>
          {% endif %}
        </tbody>
        <tfoot>
          <tr>
            <td colspan="6">
              {% for action in action_list %}
                <input type="submit" name="action"
                  class="goog-button" value="{{ action }}">
              {% endfor %}
            </td>
          </tr>
        </tfoot>
      </table>
    </form>
    {% else %}
      <table class="ae-table">
        <tbody>
          <tr>
            <td>
              <div class="ae-disabled"
                  style="padding-top: 0.75em; padding-bottom: 0.75em;
                         text-align: center;">
                Datastore is empty.
              </div>
            </td>
          </tr>
        </tbody>
      </table>
    {% endif %}
    {% if pending_backups %}
      <p>
      <div class="ae-table-caption">
        <div class="ae-paginate">
          <div class="g-section g-tpl-50-50 g-split">
            <div class="g-unit g-first">
              <strong>Pending Backups</strong>
            </div>
          </div>
        </div>
      </div>
      <form action="{{ base_path }}/" method="post">
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        <input type="hidden" name="app_id" value="{{ app_id|escape }}">
        <table id="pending_backups" class="ae-table ae-table-striped">
          <thead>
            <tr>
              <th scope="col" class="cbc tct-selectall"></th>
              <th scope="col" class="id">Name</th>
              <th scope="col" class="id">Start-Time</th>
            </tr>
          </thead>
          <tbody>
            {% for pending_backup in pending_backups %}
              <tr>
                <td class="cbc">
                  <input type="checkbox" name="backup_id" value="{{ pending_backup.key }}" id="{{ pending_backup.key }}">
                </td>
                <td class="id">
                  <span title="Kinds: {{ pending_backup.kinds|join:', '|truncatewords:20|escape}}">{{ pending_backup.name }}{% if pending_backup.original_app %} (Imported from {{pending_backup.original_app|escape}}){% endif %}</span>
                </td>
                <td class="id">
                  {{ pending_backup.start_time }}
                </td>
              </tr>
            {% endfor %}
          </tbody>
          <tfoot>
            <tr>
              <td colspan="4" nowrap="nowrap">
                {% for pending_backup_action in pending_backup_action_list %}
                    <input type="submit" name="action" style="" class="goog-button" value="{{ pending_backup_action }}">
                {% endfor %}
              </td>
            </tr>
          </tfoot>
        </table>
      </form>
    {% endif %}
    <p>
    <div class="ae-table-caption">
      <div class="ae-paginate">
        <div class="g-section g-tpl-50-50 g-split">
          <div class="g-unit g-first">
            <strong>Backups</strong>
          </div>
        </div>
      </div>
    </div>
    {% if backups %}
      <form action="{{ base_path }}/" method="post">
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        <input type="hidden" name="app_id" value="{{ app_id|escape }}">
        <table id="backups" class="ae-table ae-table-striped">
          <thead>
            <tr>
              <th scope="col" class="cbc tct-selectall"></th>
              <th scope="col" class="id">Name</th>
              <th scope="col" class="id">Start-Time</th>
              <th scope="col" class="id">End-Time</th>
            </tr>
          </thead>
          <tbody>
            {% for backup in backups %}
              <tr>
                <td class="cbc">
                  <input type="checkbox" name="backup_id" value="{{ backup.key }}" id="{{ backup.key }}">
                </td>
                <td class="id">
                  <span title="Kinds: {{ backup.kinds|join:', '|truncatewords:20|escape }}">{{ backup.name }}{% if backup.original_app %} (Imported from {{backup.original_app|escape}}){% endif %}</span>
                </td>
                <td class="id">
                  {{ backup.start_time }}
                </td>
                <td class="id">
                  {{ backup.complete_time }}
                </td>
              </tr>
            {% endfor %}
          </tbody>
          <tfoot>
            <tr>
              <td colspan="4" nowrap="nowrap">
                {% for backup_action in backup_action_list %}
                    <input type="submit" name="action" style="" class="goog-button" value="{{ backup_action }}">
                {% endfor %}
              </td>
            </tr>
          </tfoot>
        </table>
      </form>
    {% endif %}
    <div class="ae-table-caption">
      <form action="{{ base_path }}/" method="post" >
        {% if run_as_a_service %}
          <input type="hidden" name="run_as_a_service" value="{{ run_as_a_service|escape }}">
        {% endif %}
        <input type="submit" name="action" value="Import Backup Information" title="Import Backup Information or restore from Google Cloud Storage"/>
        <input type="text" name="gs_handle" value="" size="50" title="Google Cloud Storage path of a backup info file or a bucket name"/>
      </form>
    </div>
    {% if active_operations %}
      <p>
      <div class="ae-table-caption">
        <div class="ae-paginate">
          <div class="g-section g-tpl-50-50 g-split">
            <div class="g-unit g-first">
              <strong>Active Operations</strong>
            </div>
          </div>
        </div>
      </div>
      <table id="operations" class="ae-table ae-table-striped">
        <thead>
          <tr>
            <th scope="col" class="id">Description</th>
            <th scope="col" class="id">Jobs</th>
          </tr>
        </thead>
        <tbody>
          {% for operation in active_operations %}
            <tr>
              <td class="id">
                {{ operation.description }}
              </td>
              <td class="id">
                ({{operation.completed_jobs}} steps completed,
                {{operation.active_jobs}} active)
                {% for job_id in operation.active_job_ids %}
                 <br/><a href="{{ map_reduce_path }}?mapreduce_id={{job_id}}">Job {{job_id}}</a>
                {% endfor %}
              </td>
            </tr>
          {% endfor %}
        </tbody>
      </table>
    {% endif %}
    {% if completed_operations %}
      <p>
      <div class="ae-table-caption">
        <div class="ae-paginate">
          <div class="g-section g-tpl-50-50 g-split">
            <div class="g-unit g-first">
              <strong>Completed Operations</strong>
            </div>
          </div>
        </div>
      </div>
      <table id="operations" class="ae-table ae-table-striped">
        <thead>
          <tr>
            <th scope="col" class="id">Description</th>
            <th scope="col" class="id">Completed steps</th>
            <th scope="col" class="id">Status</th>
            <th scope="col" class="id">Status info</th>
          </tr>
        </thead>
        <tbody>
          {% for operation in completed_operations %}
            <tr>
              <td class="id">
                {{ operation.description }}
              </td>
              <td class="id">
                {{operation.completed_jobs}} steps
              </td>
              <td class="id">
                {{operation.status}}
              </td>
              <td class="id">
                {{operation.status_info}}
              </td>
            </tr>
          {% endfor %}
        </tbody>
      </table>
    {% endif %}
  {% endblock %}
  
files: AppServer/google/appengine/ext/datastore_admin/templates/list_actions.html
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.
  #
  
  
  
  
  """Used render templates for datastore admin."""
  
  
  
  import base64
  import datetime
  import json
  import logging
  import os
  import random
  import time
  
  # AppScale: Use bundled mapreduce library.
  from google.appengine.ext.mapreduce import context
  from google.appengine.ext.mapreduce import control
  from google.appengine.ext.mapreduce import model
  from google.appengine.ext.mapreduce import operation as mr_operation
  from google.appengine.ext.mapreduce import util
  
  from google.appengine.api import app_identity
  from google.appengine.api import datastore
  from google.appengine.api import datastore_errors
  from google.appengine.api import memcache
  from google.appengine.api import urlfetch
  from google.appengine.api import users
  from google.appengine.datastore import datastore_rpc
  from google.appengine.datastore import entity_pb
  from google.appengine.ext import db
  import webapp2 as webapp
  from google.appengine.ext.datastore_admin import config
  from google.appengine.ext.db import metadata
  from google.appengine.ext.db import stats
  from google.appengine.ext.webapp import _template
  from google.appengine.runtime import apiproxy_errors
  
  
  MEMCACHE_NAMESPACE = '_ah-datastore_admin'
  XSRF_VALIDITY_TIME = 600
  KINDS_AND_SIZES_VAR = 'kinds_and_sizes'
  MAPREDUCE_MIN_SHARDS = 8
  MAPREDUCE_DEFAULT_SHARDS = 32
  MAPREDUCE_MAX_SHARDS = 256
  RESERVE_KEY_POOL_MAX_SIZE = 1000
  
  DATASTORE_ADMIN_OPERATION_KIND = '_AE_DatastoreAdmin_Operation'
  BACKUP_INFORMATION_KIND = '_AE_Backup_Information'
  BACKUP_INFORMATION_FILES_KIND = '_AE_Backup_Information_Kind_Files'
  BACKUP_INFORMATION_KIND_TYPE_INFO = '_AE_Backup_Information_Kind_Type_Info'
  DATASTORE_ADMIN_KINDS = (DATASTORE_ADMIN_OPERATION_KIND,
                           BACKUP_INFORMATION_KIND,
                           BACKUP_INFORMATION_FILES_KIND,
                           BACKUP_INFORMATION_KIND_TYPE_INFO)
  
  
  MAX_RPCS = 10
  
  
  def IsKindNameVisible(kind_name):
    return not (kind_name.startswith('__') or
                kind_name in DATASTORE_ADMIN_KINDS or
                kind_name in model._MAP_REDUCE_KINDS)
  
  
  def RenderToResponse(handler, template_file, template_params):
    """Render the given template_file using template_vals and write to response.
  
    Args:
      handler: the handler whose response we should render to
      template_file: the file name only of the template file we are using
      template_params: the parameters used to render the given template
    """
    template_params = _GetDefaultParams(template_params)
  
  
  
  
  
  
  
  
    handler.response.headers['X-FRAME-OPTIONS'] = ('ALLOW-FROM %s' %
                                                   config.ADMIN_CONSOLE_URL)
    template_params['admin_console_url'] = config.ADMIN_CONSOLE_URL
  
    rendered = _template.render(_GetTemplatePath(template_file), template_params)
    handler.response.out.write(rendered)
  
  
  def _GetTemplatePath(template_file):
    """Return the expected path for the template to render.
  
    Args:
      template_file: simple file name of template to render.
  
    Returns:
      path of template to render.
    """
    return os.path.join(
        os.path.dirname(__file__), 'templates', template_file)
  
  
  def _GetDefaultParams(template_params):
    """Update template_params to always contain necessary paths and never None."""
    if not template_params:
      template_params = {}
    template_params.update({
        'base_path': config.BASE_PATH,
        'mapreduce_path': config.MAPREDUCE_PATH,
    })
    return template_params
  
  
  def CreateXsrfToken(action):
    """Generate a token to be passed with a form for XSRF protection.
  
    Args:
      action: action to restrict token to
  
    Returns:
      suitably random token which is only valid for ten minutes and, if the user
      is authenticated, is only valid for the user that generated it.
    """
    user_str = _MakeUserStr()
  
    token = base64.b64encode(
        ''.join(chr(int(random.random()*255)) for _ in range(0, 64)))
  
    memcache.set(token,
                 (user_str, action),
                 time=XSRF_VALIDITY_TIME,
                 namespace=MEMCACHE_NAMESPACE)
  
    return token
  
  
  def ValidateXsrfToken(token, action):
    """Validate a given XSRF token by retrieving it from memcache.
  
    If the token has not been evicted from memcache (past ten minutes) and the
    user strings are equal, then this is a valid token.
  
    Args:
      token: token to validate from memcache.
      action: action that token should correspond to
  
    Returns:
      True if the token exists in memcache and the user strings are equal,
      False otherwise.
    """
    user_str = _MakeUserStr()
    token_obj = memcache.get(token, namespace=MEMCACHE_NAMESPACE)
  
    if not token_obj:
      return False
  
    token_str, token_action = token_obj
    if user_str != token_str or action != token_action:
      return False
  
    return True
  
  
  def CacheStats(formatted_results):
    """Cache last retrieved kind size values in memcache.
  
    Args:
      formatted_results: list of dictionaries of the form returnned by
        main._PresentableKindStats.
    """
    kinds_and_sizes = dict((kind['kind_name'], kind['total_bytes'])
                           for kind in formatted_results)
  
    memcache.set(KINDS_AND_SIZES_VAR,
                 kinds_and_sizes,
                 namespace=MEMCACHE_NAMESPACE)
  
  
  def RetrieveCachedStats():
    """Retrieve cached kind sizes from last datastore stats call.
  
    Returns:
      Dictionary mapping kind names to total bytes.
    """
    return memcache.get(KINDS_AND_SIZES_VAR, namespace=MEMCACHE_NAMESPACE)
  
  
  def _MakeUserStr():
    """Make a user string to use to represent the user.  'noauth' by default."""
    user = users.get_current_user()
    return user.nickname() if user else 'noauth'
  
  
  def GetPrettyBytes(bytes_num, significant_digits=0):
    """Get a pretty print view of the given number of bytes.
  
    This will give a string like 'X MBytes'.
  
    Args:
      bytes_num: the original number of bytes to pretty print.
      significant_digits: number of digits to display after the decimal point.
  
    Returns:
      A string that has the pretty print version of the given bytes.
      If bytes_num is to big the string 'Alot' will be returned.
    """
    byte_prefixes = ['', 'K', 'M', 'G', 'T', 'P', 'E']
    for i in range(0, 7):
      exp = i * 10
      if bytes_num < 1<<(exp + 10):
        if i == 0:
          formatted_bytes = str(bytes_num)
        else:
          formatted_bytes = '%.*f' % (significant_digits,
                                      (bytes_num * 1.0 / (1<<exp)))
        if formatted_bytes != '1':
          plural = 's'
        else:
          plural = ''
        return '%s %sByte%s' % (formatted_bytes, byte_prefixes[i], plural)
  
    logging.error('Number too high to convert: %d', bytes_num)
    return 'Alot'
  
  
  def FormatThousands(value):
    """Format a numerical value, inserting commas as thousands separators.
  
    Args:
      value: An integer, float, or string representation thereof.
        If the argument is a float, it is converted to a string using '%.2f'.
  
    Returns:
      A string with groups of 3 digits before the decimal point (if any)
      separated by commas.
  
    NOTE: We don't deal with whitespace, and we don't insert
    commas into long strings of digits after the decimal point.
    """
    if isinstance(value, float):
      value = '%.2f' % value
    else:
      value = str(value)
    if '.' in value:
      head, tail = value.split('.', 1)
      tail = '.' + tail
    elif 'e' in value:
      head, tail = value.split('e', 1)
      tail = 'e' + tail
    else:
      head = value
      tail = ''
    sign = ''
    if head.startswith('-'):
      sign = '-'
      head = head[1:]
    while len(head) > 3:
      tail = ',' + head[-3:] + tail
      head = head[:-3]
    return sign + head + tail
  
  
  def TruncDelta(delta):
    """Strips microseconds from a timedelta."""
    return datetime.timedelta(days=delta.days, seconds=delta.seconds)
  
  
  def GetPrintableStrs(namespace, kinds):
    """Returns tuples describing affected kinds and namespace.
  
    Args:
      namespace: namespace being targeted.
      kinds: list of kinds being targeted.
  
    Returns:
      (namespace_str, kind_str) tuple used for display to user.
    """
    namespace_str = namespace or ''
    if kinds:
      kind_str = 'all %s entities' % ', '.join(kinds)
    else:
      kind_str = ''
    return (namespace_str, kind_str)
  
  
  def ParseKindsAndSizes(kinds):
    """Parses kind|size list and returns template parameters.
  
    Args:
      kinds: list of kinds to process.
  
    Returns:
      sizes_known: whether or not all kind objects have known sizes.
      size_total: total size of objects with known sizes.
      len(kinds) - 2: for template rendering of greater than 3 kinds.
    """
    sizes_known = True
    size_total = 0
    kinds_and_sizes = RetrieveCachedStats()
  
    if kinds_and_sizes:
      for kind in kinds:
        if kind in kinds_and_sizes:
          size_total += kinds_and_sizes[kind]
        else:
          sizes_known = False
    else:
      sizes_known = False
  
    if size_total:
      size_total = GetPrettyBytes(size_total)
  
    return sizes_known, size_total, len(kinds) - 2
  
  
  def _CreateDatastoreConfig():
    """Create datastore config for use during datastore operations."""
    return datastore_rpc.Configuration(force_writes=True, deadline=60)
  
  
  def GenerateHomeUrl(unused_request):
    """Generates a link to the Datastore Admin main page.
  
    Primarily intended to be used for cancel buttons or links on error pages. To
    avoid any XSS security vulnerabilities the URL should not use any
    user-defined strings (unless proper precautions are taken).
  
    Args:
      unused_request: the webapp.Request object (to determine if certain query
        parameters need to be used).
  
    Returns:
      domain-relative URL for the main Datastore Admin page.
    """
    return config.BASE_PATH
  
  
  class MapreduceDoneHandler(webapp.RequestHandler):
    """Handler to delete data associated with successful MapReduce jobs."""
  
    SUFFIX = 'mapreduce_done'
  
    def post(self):
      """Mapreduce done callback to delete job data if it was successful."""
      if 'Mapreduce-Id' in self.request.headers:
        mapreduce_id = self.request.headers['Mapreduce-Id']
        mapreduce_state = model.MapreduceState.get_by_job_id(mapreduce_id)
        if not mapreduce_state:
          logging.error('Done callback for no longer valid Mapreduce Id: %s',
                        mapreduce_id)
          return
  
        mapreduce_params = mapreduce_state.mapreduce_spec.params
  
        db_config = _CreateDatastoreConfig()
        if mapreduce_state.result_status == model.MapreduceState.RESULT_SUCCESS:
          operation_key = mapreduce_params.get(
              DatastoreAdminOperation.PARAM_DATASTORE_ADMIN_OPERATION)
          if operation_key is None:
            logging.error('Done callback for job %s without operation key.',
                          mapreduce_id)
          else:
  
            def tx():
              operation = DatastoreAdminOperation.get(operation_key)
              if mapreduce_id in operation.active_job_ids:
                operation.active_jobs -= 1
                operation.completed_jobs += 1
                operation.active_job_ids.remove(mapreduce_id)
              if not operation.active_jobs:
                if operation.status == DatastoreAdminOperation.STATUS_ACTIVE:
                  operation.status = DatastoreAdminOperation.STATUS_COMPLETED
                db.delete(DatastoreAdminOperationJob.all().ancestor(operation),
                          config=db_config)
              operation.put(config=db_config)
              if 'done_callback_handler' in mapreduce_params:
                done_callback_handler = util.for_name(
                    mapreduce_params['done_callback_handler'])
                if done_callback_handler:
                  done_callback_handler(operation, mapreduce_id, mapreduce_state)
                else:
                  logging.error('done_callbackup_handler %s was not found',
                                mapreduce_params['done_callback_handler'])
            db.run_in_transaction(tx)
          if config.CLEANUP_MAPREDUCE_STATE:
            keys = []
            keys = model.ShardState.calculate_keys_by_mapreduce_state(
                mapreduce_state)
  
            keys.append(model.MapreduceControl.get_key_by_job_id(mapreduce_id))
            db.delete(keys, config=db_config)
            db.delete(mapreduce_state, config=db_config)
            logging.info('State for successful job %s was deleted.', mapreduce_id)
        else:
          logging.info('Job %s was not successful so no state was deleted.',
                       mapreduce_id)
      else:
        logging.error('Done callback called without Mapreduce Id.')
  
  
  class Error(Exception):
    """Base DatastoreAdmin error type."""
  
  
  
  class DatastoreAdminOperation(db.Model):
    """An entity to keep progress and status of datastore admin operation."""
    STATUS_CREATED = 'Created'
    STATUS_ACTIVE = 'Active'
    STATUS_COMPLETED = 'Completed'
    STATUS_FAILED = 'Failed'
    STATUS_ABORTED = 'Aborted'
  
  
    PARAM_DATASTORE_ADMIN_OPERATION = 'datastore_admin_operation'
    DEFAULT_LAST_UPDATED_VALUE = datetime.datetime(1970, 1, 1)
  
    description = db.TextProperty()
    status = db.StringProperty(default=STATUS_CREATED)
    active_jobs = db.IntegerProperty(default=0)
    active_job_ids = db.StringListProperty()
    completed_jobs = db.IntegerProperty(default=0)
    last_updated = db.DateTimeProperty(default=DEFAULT_LAST_UPDATED_VALUE,
                                       auto_now=True)
    status_info = db.StringProperty(default='', indexed=False)
    service_job_id = db.StringProperty()
  
    @classmethod
    def kind(cls):
      return DATASTORE_ADMIN_OPERATION_KIND
  
  
  class DatastoreAdminOperationJob(db.Model):
    """An entity to keep track of started jobs to ensure idempotency.
  
    This entity can be used during spawning additional jobs. It is
    always stored as a child entity of DatastoreAdminOperation.
    Entity key name is job unique id.
    """
    pass
  
  
  def StartOperation(description):
    """Start datastore admin operation.
  
    Args:
      description: operation description to be displayed to user.
  
    Returns:
      an instance of DatastoreAdminOperation.
    """
  
  
  
    operation = DatastoreAdminOperation(
        description=description,
        id=db.allocate_ids(
            db.Key.from_path(DatastoreAdminOperation.kind(), 1), 1)[0])
    operation.put(config=_CreateDatastoreConfig())
    return operation
  
  
  @db.non_transactional(allow_existing=False)
  def StartMap(operation_key,
               job_name,
               handler_spec,
               reader_spec,
               writer_spec,
               mapper_params,
               mapreduce_params=None,
               queue_name=None,
               shard_count=MAPREDUCE_DEFAULT_SHARDS):
    """Start map as part of datastore admin operation.
  
    Will increase number of active jobs inside the operation and start new map.
  
    Args:
      operation_key: Key of the DatastoreAdminOperation for current operation.
      job_name: Map job name.
      handler_spec: Map handler specification.
      reader_spec: Input reader specification.
      writer_spec: Output writer specification.
      mapper_params: Custom mapper parameters.
      mapreduce_params: Custom mapreduce parameters.
      queue_name: the name of the queue that will be used by the M/R.
      shard_count: the number of shards the M/R will try to use.
  
    Returns:
      resulting map job id as string.
    """
  
    if not mapreduce_params:
      mapreduce_params = {}
    mapreduce_params[DatastoreAdminOperation.PARAM_DATASTORE_ADMIN_OPERATION] = (
        str(operation_key))
    mapreduce_params['done_callback'] = '%s/%s' % (config.BASE_PATH,
                                                   MapreduceDoneHandler.SUFFIX)
    if queue_name is not None:
      mapreduce_params['done_callback_queue'] = queue_name
    mapreduce_params['force_writes'] = 'True'
  
    @db.transactional(xg=True)
    def tx():
      """Start MapReduce job and update datastore admin state.
  
      Returns:
        result MapReduce job id as a string.
      """
      job_id = control.start_map(
          job_name, handler_spec, reader_spec,
          mapper_params,
          output_writer_spec=writer_spec,
          mapreduce_parameters=mapreduce_params,
          base_path=config.MAPREDUCE_PATH,
          shard_count=shard_count,
          in_xg_transaction=True,
          queue_name=queue_name)
      operation = DatastoreAdminOperation.get(operation_key)
      operation.status = DatastoreAdminOperation.STATUS_ACTIVE
      operation.active_jobs += 1
      operation.active_job_ids = list(set(operation.active_job_ids + [job_id]))
      operation.put(config=_CreateDatastoreConfig())
      return job_id
  
    return tx()
  
  
  def RunMapForKinds(operation_key,
                     kinds,
                     job_name_template,
                     handler_spec,
                     reader_spec,
                     writer_spec,
                     mapper_params,
                     mapreduce_params=None,
                     queue_name=None,
                     max_shard_count=None):
    """Run mapper job for all entities in specified kinds.
  
    Args:
      operation_key: The key of the DatastoreAdminOperation to record all jobs.
      kinds: list of entity kinds as strings.
      job_name_template: template for naming individual mapper jobs. Can
        reference %(kind)s and %(namespace)s formatting variables.
      handler_spec: mapper handler specification.
      reader_spec: reader specification.
      writer_spec: writer specification.
      mapper_params: custom parameters to pass to mapper.
      mapreduce_params: dictionary parameters relevant to the whole job.
      queue_name: the name of the queue that will be used by the M/R.
      max_shard_count: maximum value for shards count.
  
    Returns:
      Ids of all started mapper jobs as list of strings.
    """
    jobs = []
    try:
      for kind in kinds:
        mapper_params['entity_kind'] = kind
        job_name = job_name_template % {'kind': kind, 'namespace':
                                        mapper_params.get('namespace', '')}
        shard_count = GetShardCount(kind, max_shard_count)
        jobs.append(StartMap(operation_key, job_name, handler_spec, reader_spec,
                             writer_spec, mapper_params, mapreduce_params,
                             queue_name=queue_name, shard_count=shard_count))
      return jobs
  
    except BaseException, ex:
      AbortAdminOperation(operation_key,
                          _status=DatastoreAdminOperation.STATUS_FAILED,
                          _status_info='%s: %s' % (ex.__class__.__name__, ex))
      raise
  
  
  def GetShardCount(kind, max_shard_count=None):
    stat = stats.KindStat.all().filter('kind_name =', kind).get()
    if stat:
  
      shard_count = min(max(MAPREDUCE_MIN_SHARDS,
                            stat.bytes // (32 * 1024 * 1024)),
                        MAPREDUCE_MAX_SHARDS)
      if max_shard_count and max_shard_count < shard_count:
        shard_count = max_shard_count
      return shard_count
  
    return MAPREDUCE_DEFAULT_SHARDS
  
  
  def AbortAdminOperation(operation_key,
                          _status=DatastoreAdminOperation.STATUS_ABORTED,
                          _status_info=''):
    """Aborts active jobs."""
    operation = DatastoreAdminOperation.get(operation_key)
    operation.status = _status
    operation.status_info = _status_info
    operation.put(config=_CreateDatastoreConfig())
    for job in operation.active_job_ids:
      logging.info('Aborting Job %s', job)
      model.MapreduceControl.abort(job, config=_CreateDatastoreConfig())
  
  
  def get_kind_from_entity_pb(entity):
    element_list = entity.key().path().element_list()
    return element_list[-1].type() if element_list else None
  
  
  def FixKeys(entity_proto, app_id):
    """Go over keys in the given entity and update the application id.
  
    Args:
      entity_proto: An EntityProto to be fixed up. All identifiable keys in the
        proto will have the 'app' field reset to match app_id.
      app_id: The desired application id, typically os.getenv('APPLICATION_ID').
    """
  
    def FixKey(mutable_key):
      mutable_key.set_app(app_id)
  
    def FixPropertyList(property_list):
      for prop in property_list:
        prop_value = prop.mutable_value()
        if prop_value.has_referencevalue():
          FixKey(prop_value.mutable_referencevalue())
        elif prop.meaning() == entity_pb.Property.ENTITY_PROTO:
          embedded_entity_proto = entity_pb.EntityProto()
          try:
            embedded_entity_proto.ParsePartialFromString(prop_value.stringvalue())
          except Exception:
            logging.exception('Failed to fix-keys for property %s of %s',
                              prop.name(),
                              entity_proto.key())
          else:
            FixKeys(embedded_entity_proto, app_id)
            prop_value.set_stringvalue(
                embedded_entity_proto.SerializePartialToString())
  
  
    if entity_proto.has_key() and entity_proto.key().path().element_size():
      FixKey(entity_proto.mutable_key())
  
    FixPropertyList(entity_proto.property_list())
    FixPropertyList(entity_proto.raw_property_list())
  
  
  class ReserveKeyPool(object):
    """Mapper pool which buffers keys with ids to reserve.
  
    Runs AllocateIds rpc(s) when flushed.
    """
  
    def __init__(self):
      self.keys = []
  
    def reserve_key(self, key):
      for id_or_name in key.to_path()[1::2]:
        if isinstance(id_or_name, (int, long)):
          self.keys.append(key)
          if len(self.keys) >= RESERVE_KEY_POOL_MAX_SIZE:
            self.flush()
          return
  
    def flush(self):
      if self.keys:
  
        datastore._GetConnection()._reserve_keys(self.keys)
        self.keys = []
  
  
  class ReserveKey(mr_operation.Operation):
    """Mapper operation to reserve key ids."""
  
    def __init__(self, key):
      self.key = key
      self.app_id = key.app()
      self.pool_id = 'reserve_key_%s_pool' % self.app_id
  
    def __call__(self, ctx):
      pool = ctx.get_pool(self.pool_id)
      if not pool:
        pool = ReserveKeyPool()
        ctx.register_pool(self.pool_id, pool)
      pool.reserve_key(self.key)
  
  
  class PutPool(context.Pool):
    """A trimmed copy of the MutationPool class.
  
    Properties:
      puts: a list of entities to put to datastore.
      max_entity_count: maximum number of entities before flushing it to db.
    """
    POOL_NAME = 'put_pool'
  
    def __init__(self, max_entity_count=context.MAX_ENTITY_COUNT):
      """Constructor.
  
      Args:
        max_entity_count: maximum number of entities before flushing it to db.
      """
      self.max_entity_count = max_entity_count
      self.puts = []
  
    def Put(self, entity):
      """Registers entity to put to datastore.
  
      Args:
        entity: The EntityProto for the entity to be put.
      """
      if len(self.puts) >= self.max_entity_count:
        self.flush()
      self.puts.append(entity)
  
    def flush(self):
      """Flush all puts to datastore."""
      if self.puts:
        datastore_rpc.Connection(config=_CreateDatastoreConfig()).put(self.puts)
      self.puts = []
  
  
  class Put(mr_operation.Operation):
    """Mapper operation to batch puts."""
  
    def __init__(self, entity):
      """Constructor.
  
      Args:
        entity: The EntityProto of the entity to put.
      """
      self.entity = entity
  
    def __call__(self, ctx):
      pool = ctx.get_pool(PutPool.POOL_NAME)
      if not pool:
        pool = PutPool(
            max_entity_count=(context.MAX_ENTITY_COUNT/(2**ctx.task_retry_count)))
        ctx.register_pool(PutPool.POOL_NAME, pool)
      pool.Put(self.entity)
  
  
  def GetKinds(all_ns=True, deadline=40):
    """Obtain a list of all kind names from the datastore.
  
    Args:
      all_ns: If true, list kind names for all namespaces.
              If false, list kind names only for the current namespace.
      deadline: maximum number of seconds to spend getting kinds.
  
    Returns:
      kinds: an alphabetized list of kinds for the specified namespace(s).
      more_kinds: a boolean indicating whether there may be additional kinds
          not included in 'kinds' (e.g. because the query deadline was reached).
    """
    if all_ns:
      kinds, more_kinds = GetKindsForAllNamespaces(deadline)
    else:
      kinds, more_kinds = GetKindsForCurrentNamespace(deadline)
    return kinds, more_kinds
  
  
  def GetKindsForAllNamespaces(deadline):
    """Obtain a list of all kind names from the datastore.
  
    Pulls kinds from all namespaces. The result is deduped and alphabetized.
  
    Args:
      deadline: maximum number of seconds to spend getting kinds.
  
    Returns:
      kinds: an alphabetized list of kinds for the specified namespace(s).
      more_kinds: a boolean indicating whether there may be additional kinds
          not included in 'kinds' (e.g. because the query deadline was reached).
    """
    start = time.time()
    kind_name_set = set()
  
    def ReadFromKindIters(kind_iter_list):
      """Read kinds from a list of iterators.
  
      Reads a kind from each iterator in kind_iter_list, adds it to
      kind_name_set, and removes any completed iterators.
  
      Args:
        kind_iter_list: a list of iterators of kinds.
      """
      completed = []
      for kind_iter in kind_iter_list:
        try:
          kind_name = kind_iter.next().kind_name
          if IsKindNameVisible(kind_name):
            kind_name_set.add(kind_name)
        except StopIteration:
          completed.append(kind_iter)
      for kind_iter in completed:
        kind_iter_list.remove(kind_iter)
  
    more_kinds = False
    try:
      namespace_iter = metadata.Namespace.all().run(batch_size=1000,
                                                    deadline=deadline)
      kind_iter_list = []
      for ns in namespace_iter:
  
  
        remaining = deadline - (time.time() - start)
  
        if remaining <= 0:
          raise datastore_errors.Timeout
        kind_iter_list.append(metadata.Kind.all(namespace=ns.namespace_name)
                              .run(batch_size=1000, deadline=remaining))
        while len(kind_iter_list) == MAX_RPCS:
          ReadFromKindIters(kind_iter_list)
      while kind_iter_list:
        ReadFromKindIters(kind_iter_list)
    except (datastore_errors.Timeout, apiproxy_errors.DeadlineExceededError):
      more_kinds = True
      logging.warning('Failed to retrieve all kinds within deadline.')
    return sorted(kind_name_set), more_kinds
  
  
  def GetKindsForCurrentNamespace(deadline):
    """Obtain a list of all kind names from the datastore.
  
    Pulls kinds from the current namespace only. The result is alphabetized.
  
    Args:
      deadline: maximum number of seconds to spend getting kinds.
  
    Returns:
      kinds: an alphabetized list of kinds for the specified namespace(s).
      more_kinds: a boolean indicating whether there may be additional kinds
          not included in 'kinds' (e.g. because the query limit was reached).
    """
    more_kinds = False
    kind_names = []
    try:
      kinds = metadata.Kind.all().order('__key__').run(batch_size=1000,
                                                       deadline=deadline)
      for kind in kinds:
        kind_name = kind.kind_name
        if IsKindNameVisible(kind_name):
          kind_names.append(kind_name)
    except (datastore_errors.Timeout, apiproxy_errors.DeadlineExceededError):
      more_kinds = True
      logging.warning('Failed to retrieve all kinds within deadline.')
    return kind_names, more_kinds
  
  
  def get_service_account_names():
    """ AppScale: Fetch list of service accounts from IAM API. """
    project_id = app_identity.get_application_id()
    iam_location = 'https://127.0.0.1:17441'
    url = iam_location + '/v1/projects/{}/serviceAccounts'.format(project_id)
    token = app_identity.get_access_token(
        ['https://www.googleapis.com/auth/cloud-platform'])[0]
    headers = {'Authorization': 'Bearer {}'.format(token)}
    response = urlfetch.fetch(url, headers=headers, validate_certificate=False)
    try:
      accounts = json.loads(response.content)['accounts']
    except (KeyError, ValueError):
      raise ValueError('Invalid list of service accounts: '
                       '{}'.format(response.content))
  
    return tuple(account['email'] for account in accounts)
  
files: AppServer/google/appengine/ext/datastore_admin/utils.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.
  #
  
  
  
  
  """Used render templates for datastore admin."""
  
  
  
  import base64
  import datetime
  import logging
  import os
  import random
  import time
  
  # AppScale: Use bundled mapreduce library.
  from google.appengine.ext.mapreduce import context
  from google.appengine.ext.mapreduce import control
  from google.appengine.ext.mapreduce import model
  from google.appengine.ext.mapreduce import operation as mr_operation
  from google.appengine.ext.mapreduce import util
  
  from google.appengine.api import datastore
  from google.appengine.api import datastore_errors
  from google.appengine.api import memcache
  from google.appengine.api import users
  from google.appengine.datastore import datastore_rpc
  from google.appengine.datastore import entity_pb
  from google.appengine.ext import db
  import webapp2 as webapp
  from google.appengine.ext.datastore_admin import config
  from google.appengine.ext.db import metadata
  from google.appengine.ext.db import stats
  from google.appengine.ext.webapp import _template
  from google.appengine.runtime import apiproxy_errors
  
  
  MEMCACHE_NAMESPACE = '_ah-datastore_admin'
  XSRF_VALIDITY_TIME = 600
  KINDS_AND_SIZES_VAR = 'kinds_and_sizes'
  MAPREDUCE_MIN_SHARDS = 8
  MAPREDUCE_DEFAULT_SHARDS = 32
  MAPREDUCE_MAX_SHARDS = 256
  RESERVE_KEY_POOL_MAX_SIZE = 1000
  
  DATASTORE_ADMIN_OPERATION_KIND = '_AE_DatastoreAdmin_Operation'
  BACKUP_INFORMATION_KIND = '_AE_Backup_Information'
  BACKUP_INFORMATION_FILES_KIND = '_AE_Backup_Information_Kind_Files'
  BACKUP_INFORMATION_KIND_TYPE_INFO = '_AE_Backup_Information_Kind_Type_Info'
  DATASTORE_ADMIN_KINDS = (DATASTORE_ADMIN_OPERATION_KIND,
                           BACKUP_INFORMATION_KIND,
                           BACKUP_INFORMATION_FILES_KIND,
                           BACKUP_INFORMATION_KIND_TYPE_INFO)
  
  
  MAX_RPCS = 10
  
  
  def IsKindNameVisible(kind_name):
    return not (kind_name.startswith('__') or
                kind_name in DATASTORE_ADMIN_KINDS or
                kind_name in model._MAP_REDUCE_KINDS)
  
  
  def RenderToResponse(handler, template_file, template_params):
    """Render the given template_file using template_vals and write to response.
  
    Args:
      handler: the handler whose response we should render to
      template_file: the file name only of the template file we are using
      template_params: the parameters used to render the given template
    """
    template_params = _GetDefaultParams(template_params)
  
  
  
  
  
  
  
  
    handler.response.headers['X-FRAME-OPTIONS'] = ('ALLOW-FROM %s' %
                                                   config.ADMIN_CONSOLE_URL)
    template_params['admin_console_url'] = config.ADMIN_CONSOLE_URL
  
    rendered = _template.render(_GetTemplatePath(template_file), template_params)
    handler.response.out.write(rendered)
  
  
  def _GetTemplatePath(template_file):
    """Return the expected path for the template to render.
  
    Args:
      template_file: simple file name of template to render.
  
    Returns:
      path of template to render.
    """
    return os.path.join(
        os.path.dirname(__file__), 'templates', template_file)
  
  
  def _GetDefaultParams(template_params):
    """Update template_params to always contain necessary paths and never None."""
    if not template_params:
      template_params = {}
    template_params.update({
        'base_path': config.BASE_PATH,
        'mapreduce_path': config.MAPREDUCE_PATH,
    })
    return template_params
  
  
  def CreateXsrfToken(action):
    """Generate a token to be passed with a form for XSRF protection.
  
    Args:
      action: action to restrict token to
  
    Returns:
      suitably random token which is only valid for ten minutes and, if the user
      is authenticated, is only valid for the user that generated it.
    """
    user_str = _MakeUserStr()
  
    token = base64.b64encode(
        ''.join(chr(int(random.random()*255)) for _ in range(0, 64)))
  
    memcache.set(token,
                 (user_str, action),
                 time=XSRF_VALIDITY_TIME,
                 namespace=MEMCACHE_NAMESPACE)
  
    return token
  
  
  def ValidateXsrfToken(token, action):
    """Validate a given XSRF token by retrieving it from memcache.
  
    If the token has not been evicted from memcache (past ten minutes) and the
    user strings are equal, then this is a valid token.
  
    Args:
      token: token to validate from memcache.
      action: action that token should correspond to
  
    Returns:
      True if the token exists in memcache and the user strings are equal,
      False otherwise.
    """
    user_str = _MakeUserStr()
    token_obj = memcache.get(token, namespace=MEMCACHE_NAMESPACE)
  
    if not token_obj:
      return False
  
    token_str, token_action = token_obj
    if user_str != token_str or action != token_action:
      return False
  
    return True
  
  
  def CacheStats(formatted_results):
    """Cache last retrieved kind size values in memcache.
  
    Args:
      formatted_results: list of dictionaries of the form returnned by
        main._PresentableKindStats.
    """
    kinds_and_sizes = dict((kind['kind_name'], kind['total_bytes'])
                           for kind in formatted_results)
  
    memcache.set(KINDS_AND_SIZES_VAR,
                 kinds_and_sizes,
                 namespace=MEMCACHE_NAMESPACE)
  
  
  def RetrieveCachedStats():
    """Retrieve cached kind sizes from last datastore stats call.
  
    Returns:
      Dictionary mapping kind names to total bytes.
    """
    return memcache.get(KINDS_AND_SIZES_VAR, namespace=MEMCACHE_NAMESPACE)
  
  
  def _MakeUserStr():
    """Make a user string to use to represent the user.  'noauth' by default."""
    user = users.get_current_user()
    return user.nickname() if user else 'noauth'
  
  
  def GetPrettyBytes(bytes_num, significant_digits=0):
    """Get a pretty print view of the given number of bytes.
  
    This will give a string like 'X MBytes'.
  
    Args:
      bytes_num: the original number of bytes to pretty print.
      significant_digits: number of digits to display after the decimal point.
  
    Returns:
      A string that has the pretty print version of the given bytes.
      If bytes_num is to big the string 'Alot' will be returned.
    """
    byte_prefixes = ['', 'K', 'M', 'G', 'T', 'P', 'E']
    for i in range(0, 7):
      exp = i * 10
      if bytes_num < 1<<(exp + 10):
        if i == 0:
          formatted_bytes = str(bytes_num)
        else:
          formatted_bytes = '%.*f' % (significant_digits,
                                      (bytes_num * 1.0 / (1<<exp)))
        if formatted_bytes != '1':
          plural = 's'
        else:
          plural = ''
        return '%s %sByte%s' % (formatted_bytes, byte_prefixes[i], plural)
  
    logging.error('Number too high to convert: %d', bytes_num)
    return 'Alot'
  
  
  def FormatThousands(value):
    """Format a numerical value, inserting commas as thousands separators.
  
    Args:
      value: An integer, float, or string representation thereof.
        If the argument is a float, it is converted to a string using '%.2f'.
  
    Returns:
      A string with groups of 3 digits before the decimal point (if any)
      separated by commas.
  
    NOTE: We don't deal with whitespace, and we don't insert
    commas into long strings of digits after the decimal point.
    """
    if isinstance(value, float):
      value = '%.2f' % value
    else:
      value = str(value)
    if '.' in value:
      head, tail = value.split('.', 1)
      tail = '.' + tail
    elif 'e' in value:
      head, tail = value.split('e', 1)
      tail = 'e' + tail
    else:
      head = value
      tail = ''
    sign = ''
    if head.startswith('-'):
      sign = '-'
      head = head[1:]
    while len(head) > 3:
      tail = ',' + head[-3:] + tail
      head = head[:-3]
    return sign + head + tail
  
  
  def TruncDelta(delta):
    """Strips microseconds from a timedelta."""
    return datetime.timedelta(days=delta.days, seconds=delta.seconds)
  
  
  def GetPrintableStrs(namespace, kinds):
    """Returns tuples describing affected kinds and namespace.
  
    Args:
      namespace: namespace being targeted.
      kinds: list of kinds being targeted.
  
    Returns:
      (namespace_str, kind_str) tuple used for display to user.
    """
    namespace_str = namespace or ''
    if kinds:
      kind_str = 'all %s entities' % ', '.join(kinds)
    else:
      kind_str = ''
    return (namespace_str, kind_str)
  
  
  def ParseKindsAndSizes(kinds):
    """Parses kind|size list and returns template parameters.
  
    Args:
      kinds: list of kinds to process.
  
    Returns:
      sizes_known: whether or not all kind objects have known sizes.
      size_total: total size of objects with known sizes.
      len(kinds) - 2: for template rendering of greater than 3 kinds.
    """
    sizes_known = True
    size_total = 0
    kinds_and_sizes = RetrieveCachedStats()
  
    if kinds_and_sizes:
      for kind in kinds:
        if kind in kinds_and_sizes:
          size_total += kinds_and_sizes[kind]
        else:
          sizes_known = False
    else:
      sizes_known = False
  
    if size_total:
      size_total = GetPrettyBytes(size_total)
  
    return sizes_known, size_total, len(kinds) - 2
  
  
  def _CreateDatastoreConfig():
    """Create datastore config for use during datastore operations."""
    return datastore_rpc.Configuration(force_writes=True, deadline=60)
  
  
  def GenerateHomeUrl(unused_request):
    """Generates a link to the Datastore Admin main page.
  
    Primarily intended to be used for cancel buttons or links on error pages. To
    avoid any XSS security vulnerabilities the URL should not use any
    user-defined strings (unless proper precautions are taken).
  
    Args:
      unused_request: the webapp.Request object (to determine if certain query
        parameters need to be used).
  
    Returns:
      domain-relative URL for the main Datastore Admin page.
    """
    return config.BASE_PATH
  
  
  class MapreduceDoneHandler(webapp.RequestHandler):
    """Handler to delete data associated with successful MapReduce jobs."""
  
    SUFFIX = 'mapreduce_done'
  
    def post(self):
      """Mapreduce done callback to delete job data if it was successful."""
      if 'Mapreduce-Id' in self.request.headers:
        mapreduce_id = self.request.headers['Mapreduce-Id']
        mapreduce_state = model.MapreduceState.get_by_job_id(mapreduce_id)
        if not mapreduce_state:
          logging.error('Done callback for no longer valid Mapreduce Id: %s',
                        mapreduce_id)
          return
  
        mapreduce_params = mapreduce_state.mapreduce_spec.params
  
        db_config = _CreateDatastoreConfig()
        if mapreduce_state.result_status == model.MapreduceState.RESULT_SUCCESS:
          operation_key = mapreduce_params.get(
              DatastoreAdminOperation.PARAM_DATASTORE_ADMIN_OPERATION)
          if operation_key is None:
            logging.error('Done callback for job %s without operation key.',
                          mapreduce_id)
          else:
  
            def tx():
              operation = DatastoreAdminOperation.get(operation_key)
              if mapreduce_id in operation.active_job_ids:
                operation.active_jobs -= 1
                operation.completed_jobs += 1
                operation.active_job_ids.remove(mapreduce_id)
              if not operation.active_jobs:
                if operation.status == DatastoreAdminOperation.STATUS_ACTIVE:
                  operation.status = DatastoreAdminOperation.STATUS_COMPLETED
                db.delete(DatastoreAdminOperationJob.all().ancestor(operation),
                          config=db_config)
              operation.put(config=db_config)
              if 'done_callback_handler' in mapreduce_params:
                done_callback_handler = util.for_name(
                    mapreduce_params['done_callback_handler'])
                if done_callback_handler:
                  done_callback_handler(operation, mapreduce_id, mapreduce_state)
                else:
                  logging.error('done_callbackup_handler %s was not found',
                                mapreduce_params['done_callback_handler'])
            db.run_in_transaction(tx)
          if config.CLEANUP_MAPREDUCE_STATE:
            keys = []
            keys = model.ShardState.calculate_keys_by_mapreduce_state(
                mapreduce_state)
  
            keys.append(model.MapreduceControl.get_key_by_job_id(mapreduce_id))
            db.delete(keys, config=db_config)
            db.delete(mapreduce_state, config=db_config)
            logging.info('State for successful job %s was deleted.', mapreduce_id)
        else:
          logging.info('Job %s was not successful so no state was deleted.',
                       mapreduce_id)
      else:
        logging.error('Done callback called without Mapreduce Id.')
  
  
  class Error(Exception):
    """Base DatastoreAdmin error type."""
  
  
  
  class DatastoreAdminOperation(db.Model):
    """An entity to keep progress and status of datastore admin operation."""
    STATUS_CREATED = 'Created'
    STATUS_ACTIVE = 'Active'
    STATUS_COMPLETED = 'Completed'
    STATUS_FAILED = 'Failed'
    STATUS_ABORTED = 'Aborted'
  
  
    PARAM_DATASTORE_ADMIN_OPERATION = 'datastore_admin_operation'
    DEFAULT_LAST_UPDATED_VALUE = datetime.datetime(1970, 1, 1)
  
    description = db.TextProperty()
    status = db.StringProperty(default=STATUS_CREATED)
    active_jobs = db.IntegerProperty(default=0)
    active_job_ids = db.StringListProperty()
    completed_jobs = db.IntegerProperty(default=0)
    last_updated = db.DateTimeProperty(default=DEFAULT_LAST_UPDATED_VALUE,
                                       auto_now=True)
    status_info = db.StringProperty(default='', indexed=False)
    service_job_id = db.StringProperty()
  
    @classmethod
    def kind(cls):
      return DATASTORE_ADMIN_OPERATION_KIND
  
  
  class DatastoreAdminOperationJob(db.Model):
    """An entity to keep track of started jobs to ensure idempotency.
  
    This entity can be used during spawning additional jobs. It is
    always stored as a child entity of DatastoreAdminOperation.
    Entity key name is job unique id.
    """
    pass
  
  
  def StartOperation(description):
    """Start datastore admin operation.
  
    Args:
      description: operation description to be displayed to user.
  
    Returns:
      an instance of DatastoreAdminOperation.
    """
  
  
  
    operation = DatastoreAdminOperation(
        description=description,
        id=db.allocate_ids(
            db.Key.from_path(DatastoreAdminOperation.kind(), 1), 1)[0])
    operation.put(config=_CreateDatastoreConfig())
    return operation
  
  
  @db.non_transactional(allow_existing=False)
  def StartMap(operation_key,
               job_name,
               handler_spec,
               reader_spec,
               writer_spec,
               mapper_params,
               mapreduce_params=None,
               queue_name=None,
               shard_count=MAPREDUCE_DEFAULT_SHARDS):
    """Start map as part of datastore admin operation.
  
    Will increase number of active jobs inside the operation and start new map.
  
    Args:
      operation_key: Key of the DatastoreAdminOperation for current operation.
      job_name: Map job name.
      handler_spec: Map handler specification.
      reader_spec: Input reader specification.
      writer_spec: Output writer specification.
      mapper_params: Custom mapper parameters.
      mapreduce_params: Custom mapreduce parameters.
      queue_name: the name of the queue that will be used by the M/R.
      shard_count: the number of shards the M/R will try to use.
  
    Returns:
      resulting map job id as string.
    """
  
    if not mapreduce_params:
      mapreduce_params = {}
    mapreduce_params[DatastoreAdminOperation.PARAM_DATASTORE_ADMIN_OPERATION] = (
        str(operation_key))
    mapreduce_params['done_callback'] = '%s/%s' % (config.BASE_PATH,
                                                   MapreduceDoneHandler.SUFFIX)
    if queue_name is not None:
      mapreduce_params['done_callback_queue'] = queue_name
    mapreduce_params['force_writes'] = 'True'
  
    @db.transactional(xg=True)
    def tx():
      """Start MapReduce job and update datastore admin state.
  
      Returns:
        result MapReduce job id as a string.
      """
      job_id = control.start_map(
          job_name, handler_spec, reader_spec,
          mapper_params,
          output_writer_spec=writer_spec,
          mapreduce_parameters=mapreduce_params,
          base_path=config.MAPREDUCE_PATH,
          shard_count=shard_count,
          in_xg_transaction=True,
          queue_name=queue_name)
      operation = DatastoreAdminOperation.get(operation_key)
      operation.status = DatastoreAdminOperation.STATUS_ACTIVE
      operation.active_jobs += 1
      operation.active_job_ids = list(set(operation.active_job_ids + [job_id]))
      operation.put(config=_CreateDatastoreConfig())
      return job_id
  
    return tx()
  
  
  def RunMapForKinds(operation_key,
                     kinds,
                     job_name_template,
                     handler_spec,
                     reader_spec,
                     writer_spec,
                     mapper_params,
                     mapreduce_params=None,
                     queue_name=None,
                     max_shard_count=None):
    """Run mapper job for all entities in specified kinds.
  
    Args:
      operation_key: The key of the DatastoreAdminOperation to record all jobs.
      kinds: list of entity kinds as strings.
      job_name_template: template for naming individual mapper jobs. Can
        reference %(kind)s and %(namespace)s formatting variables.
      handler_spec: mapper handler specification.
      reader_spec: reader specification.
      writer_spec: writer specification.
      mapper_params: custom parameters to pass to mapper.
      mapreduce_params: dictionary parameters relevant to the whole job.
      queue_name: the name of the queue that will be used by the M/R.
      max_shard_count: maximum value for shards count.
  
    Returns:
      Ids of all started mapper jobs as list of strings.
    """
    jobs = []
    try:
      for kind in kinds:
        mapper_params['entity_kind'] = kind
        job_name = job_name_template % {'kind': kind, 'namespace':
                                        mapper_params.get('namespace', '')}
        shard_count = GetShardCount(kind, max_shard_count)
        jobs.append(StartMap(operation_key, job_name, handler_spec, reader_spec,
                             writer_spec, mapper_params, mapreduce_params,
                             queue_name=queue_name, shard_count=shard_count))
      return jobs
  
    except BaseException, ex:
      AbortAdminOperation(operation_key,
                          _status=DatastoreAdminOperation.STATUS_FAILED,
                          _status_info='%s: %s' % (ex.__class__.__name__, ex))
      raise
  
  
  def GetShardCount(kind, max_shard_count=None):
    stat = stats.KindStat.all().filter('kind_name =', kind).get()
    if stat:
  
      shard_count = min(max(MAPREDUCE_MIN_SHARDS,
                            stat.bytes // (32 * 1024 * 1024)),
                        MAPREDUCE_MAX_SHARDS)
      if max_shard_count and max_shard_count < shard_count:
        shard_count = max_shard_count
      return shard_count
  
    return MAPREDUCE_DEFAULT_SHARDS
  
  
  def AbortAdminOperation(operation_key,
                          _status=DatastoreAdminOperation.STATUS_ABORTED,
                          _status_info=''):
    """Aborts active jobs."""
    operation = DatastoreAdminOperation.get(operation_key)
    operation.status = _status
    operation.status_info = _status_info
    operation.put(config=_CreateDatastoreConfig())
    for job in operation.active_job_ids:
      logging.info('Aborting Job %s', job)
      model.MapreduceControl.abort(job, config=_CreateDatastoreConfig())
  
  
  def get_kind_from_entity_pb(entity):
    element_list = entity.key().path().element_list()
    return element_list[-1].type() if element_list else None
  
  
  def FixKeys(entity_proto, app_id):
    """Go over keys in the given entity and update the application id.
  
    Args:
      entity_proto: An EntityProto to be fixed up. All identifiable keys in the
        proto will have the 'app' field reset to match app_id.
      app_id: The desired application id, typically os.getenv('APPLICATION_ID').
    """
  
    def FixKey(mutable_key):
      mutable_key.set_app(app_id)
  
    def FixPropertyList(property_list):
      for prop in property_list:
        prop_value = prop.mutable_value()
        if prop_value.has_referencevalue():
          FixKey(prop_value.mutable_referencevalue())
        elif prop.meaning() == entity_pb.Property.ENTITY_PROTO:
          embedded_entity_proto = entity_pb.EntityProto()
          try:
            embedded_entity_proto.ParsePartialFromString(prop_value.stringvalue())
          except Exception:
            logging.exception('Failed to fix-keys for property %s of %s',
                              prop.name(),
                              entity_proto.key())
          else:
            FixKeys(embedded_entity_proto, app_id)
            prop_value.set_stringvalue(
                embedded_entity_proto.SerializePartialToString())
  
  
    if entity_proto.has_key() and entity_proto.key().path().element_size():
      FixKey(entity_proto.mutable_key())
  
    FixPropertyList(entity_proto.property_list())
    FixPropertyList(entity_proto.raw_property_list())
  
  
  class ReserveKeyPool(object):
    """Mapper pool which buffers keys with ids to reserve.
  
    Runs AllocateIds rpc(s) when flushed.
    """
  
    def __init__(self):
      self.keys = []
  
    def reserve_key(self, key):
      for id_or_name in key.to_path()[1::2]:
        if isinstance(id_or_name, (int, long)):
          self.keys.append(key)
          if len(self.keys) >= RESERVE_KEY_POOL_MAX_SIZE:
            self.flush()
          return
  
    def flush(self):
      if self.keys:
  
        datastore._GetConnection()._reserve_keys(self.keys)
        self.keys = []
  
  
  class ReserveKey(mr_operation.Operation):
    """Mapper operation to reserve key ids."""
  
    def __init__(self, key):
      self.key = key
      self.app_id = key.app()
      self.pool_id = 'reserve_key_%s_pool' % self.app_id
  
    def __call__(self, ctx):
      pool = ctx.get_pool(self.pool_id)
      if not pool:
        pool = ReserveKeyPool()
        ctx.register_pool(self.pool_id, pool)
      pool.reserve_key(self.key)
  
  
  class PutPool(context.Pool):
    """A trimmed copy of the MutationPool class.
  
    Properties:
      puts: a list of entities to put to datastore.
      max_entity_count: maximum number of entities before flushing it to db.
    """
    POOL_NAME = 'put_pool'
  
    def __init__(self, max_entity_count=context.MAX_ENTITY_COUNT):
      """Constructor.
  
      Args:
        max_entity_count: maximum number of entities before flushing it to db.
      """
      self.max_entity_count = max_entity_count
      self.puts = []
  
    def Put(self, entity):
      """Registers entity to put to datastore.
  
      Args:
        entity: The EntityProto for the entity to be put.
      """
      if len(self.puts) >= self.max_entity_count:
        self.flush()
      self.puts.append(entity)
  
    def flush(self):
      """Flush all puts to datastore."""
      if self.puts:
        datastore_rpc.Connection(config=_CreateDatastoreConfig()).put(self.puts)
      self.puts = []
  
  
  class Put(mr_operation.Operation):
    """Mapper operation to batch puts."""
  
    def __init__(self, entity):
      """Constructor.
  
      Args:
        entity: The EntityProto of the entity to put.
      """
      self.entity = entity
  
    def __call__(self, ctx):
      pool = ctx.get_pool(PutPool.POOL_NAME)
      if not pool:
        pool = PutPool(
            max_entity_count=(context.MAX_ENTITY_COUNT/(2**ctx.task_retry_count)))
        ctx.register_pool(PutPool.POOL_NAME, pool)
      pool.Put(self.entity)
  
  
  def GetKinds(all_ns=True, deadline=40):
    """Obtain a list of all kind names from the datastore.
  
    Args:
      all_ns: If true, list kind names for all namespaces.
              If false, list kind names only for the current namespace.
      deadline: maximum number of seconds to spend getting kinds.
  
    Returns:
      kinds: an alphabetized list of kinds for the specified namespace(s).
      more_kinds: a boolean indicating whether there may be additional kinds
          not included in 'kinds' (e.g. because the query deadline was reached).
    """
    if all_ns:
      kinds, more_kinds = GetKindsForAllNamespaces(deadline)
    else:
      kinds, more_kinds = GetKindsForCurrentNamespace(deadline)
    return kinds, more_kinds
  
  
  def GetKindsForAllNamespaces(deadline):
    """Obtain a list of all kind names from the datastore.
  
    Pulls kinds from all namespaces. The result is deduped and alphabetized.
  
    Args:
      deadline: maximum number of seconds to spend getting kinds.
  
    Returns:
      kinds: an alphabetized list of kinds for the specified namespace(s).
      more_kinds: a boolean indicating whether there may be additional kinds
          not included in 'kinds' (e.g. because the query deadline was reached).
    """
    start = time.time()
    kind_name_set = set()
  
    def ReadFromKindIters(kind_iter_list):
      """Read kinds from a list of iterators.
  
      Reads a kind from each iterator in kind_iter_list, adds it to
      kind_name_set, and removes any completed iterators.
  
      Args:
        kind_iter_list: a list of iterators of kinds.
      """
      completed = []
      for kind_iter in kind_iter_list:
        try:
          kind_name = kind_iter.next().kind_name
          if IsKindNameVisible(kind_name):
            kind_name_set.add(kind_name)
        except StopIteration:
          completed.append(kind_iter)
      for kind_iter in completed:
        kind_iter_list.remove(kind_iter)
  
    more_kinds = False
    try:
      namespace_iter = metadata.Namespace.all().run(batch_size=1000,
                                                    deadline=deadline)
      kind_iter_list = []
      for ns in namespace_iter:
  
  
        remaining = deadline - (time.time() - start)
  
        if remaining <= 0:
          raise datastore_errors.Timeout
        kind_iter_list.append(metadata.Kind.all(namespace=ns.namespace_name)
                              .run(batch_size=1000, deadline=remaining))
        while len(kind_iter_list) == MAX_RPCS:
          ReadFromKindIters(kind_iter_list)
      while kind_iter_list:
        ReadFromKindIters(kind_iter_list)
    except (datastore_errors.Timeout, apiproxy_errors.DeadlineExceededError):
      more_kinds = True
      logging.warning('Failed to retrieve all kinds within deadline.')
    return sorted(kind_name_set), more_kinds
  
  
  def GetKindsForCurrentNamespace(deadline):
    """Obtain a list of all kind names from the datastore.
  
    Pulls kinds from the current namespace only. The result is alphabetized.
  
    Args:
      deadline: maximum number of seconds to spend getting kinds.
  
    Returns:
      kinds: an alphabetized list of kinds for the specified namespace(s).
      more_kinds: a boolean indicating whether there may be additional kinds
          not included in 'kinds' (e.g. because the query limit was reached).
    """
    more_kinds = False
    kind_names = []
    try:
      kinds = metadata.Kind.all().order('__key__').run(batch_size=1000,
                                                       deadline=deadline)
      for kind in kinds:
        kind_name = kind.kind_name
        if IsKindNameVisible(kind_name):
          kind_names.append(kind_name)
    except (datastore_errors.Timeout, apiproxy_errors.DeadlineExceededError):
      more_kinds = True
      logging.warning('Failed to retrieve all kinds within deadline.')
    return kind_names, more_kinds
  
files: AppServer/google/appengine/ext/datastore_admin/utils.py