async def copy(src_bundle, dest_bundle, request={}, start_time=None, **kwargs):
    start_time = start_time or time.time()

    src_path, src_provider = src_bundle.pop('path'), utils.make_provider(**src_bundle.pop('provider'))
    dest_path, dest_provider = dest_bundle.pop('path'), utils.make_provider(**dest_bundle.pop('provider'))

    logger.info('Starting copying {!r}, {!r} to {!r}, {!r}'
                .format(src_path, src_provider, dest_path, dest_provider))

    metadata, errors = None, []
    try:
        metadata, created = await src_provider.copy(dest_provider, src_path, dest_path, **kwargs)
    except Exception as e:
        logger.error('Copy failed with error {!r}'.format(e))
        errors = [e.__repr__()]
        raise  # Ensure sentry sees this
    else:
        logger.info('Copy succeeded')
        dest_path = WaterButlerPath.from_metadata(metadata)
    finally:
        source = LogPayload(src_bundle['nid'], src_provider, path=src_path)
        destination = LogPayload(
            dest_bundle['nid'], dest_provider, path=dest_path, metadata=metadata
        )

        await remote_logging.wait_for_log_futures(
            'copy', source=source, destination=destination, start_time=start_time,
            errors=errors, request=request, api_version='celery',
        )

    return metadata, created

async def move(src_bundle, dest_bundle, start_time=None, **kwargs):
    start_time = start_time or time.time()

    src_path, src_provider = src_bundle.pop('path'), utils.make_provider(**src_bundle.pop('provider'))
    dest_path, dest_provider = dest_bundle.pop('path'), utils.make_provider(**dest_bundle.pop('provider'))

    logger.info('Starting moving {!r}, {!r} to {!r}, {!r}'.format(src_path, src_provider, dest_path, dest_provider))

    metadata, errors = None, []
    try:
        metadata, created = await src_provider.move(dest_provider, src_path, dest_path, **kwargs)
    except Exception as e:
        logger.error('Move failed with error {!r}'.format(e))
        errors = [e.__repr__()]
        raise  # Ensure sentry sees this
    else:
        logger.info('Move succeeded')
        dest_path = WaterButlerPath.from_metadata(metadata)
    finally:
        source = LogPayload(src_bundle['nid'], src_provider, path=src_path)
        destination = LogPayload(
            dest_bundle['nid'], dest_provider, path=dest_path, metadata=metadata
        )

        await utils.log_to_callback(
            'move',
            source=source,
            destination=destination,
            start_time=start_time,
            errors=errors
        )

    return metadata, created

async def move(src_bundle, dest_bundle, callback_url, auth, start_time=None, **kwargs):
    start_time = start_time or time.time()
    src_path, src_provider = src_bundle.pop('path'), utils.make_provider(**src_bundle.pop('provider'))
    dest_path, dest_provider = dest_bundle.pop('path'), utils.make_provider(**dest_bundle.pop('provider'))

    data = {
        'errors': [],
        'action': 'move',
        'source': dict(src_bundle, **{
            'path': src_path.identifier_path if src_provider.NAME in IDENTIFIER_PATHS else '/' + src_path.raw_path,
            'name': src_path.name,
            'materialized': str(src_path),
            'provider': src_provider.NAME,
            'kind': src_path.kind,
        }),
        'destination': dict(dest_bundle, **{
            'path': dest_path.identifier_path if dest_provider.NAME in IDENTIFIER_PATHS else '/' + dest_path.raw_path,
            'name': dest_path.name,
            'materialized': str(dest_path),
            'provider': dest_provider.NAME,
            'kind': dest_path.kind,
        }),
        'auth': auth['auth'],
    }

    logger.info('Starting moving {!r}, {!r} to {!r}, {!r}'.format(src_path, src_provider, dest_path, dest_provider))

    try:
        metadata, created = await src_provider.move(dest_provider, src_path, dest_path, **kwargs)
    except Exception as e:
        logger.error('Move failed with error {!r}'.format(e))
        data.update({'errors': [e.__repr__()]})
        raise  # Ensure sentry sees this
    else:
        logger.info('Move succeeded')
        data.update({'destination': dict(src_bundle, **metadata.serialized())})
    finally:
        resp = await utils.send_signed_request('PUT', callback_url, dict(data, **{
            'time': time.time() + 60,
            'email': time.time() - start_time > settings.WAIT_TIMEOUT
        }))
        logger.info('Callback returned {!r}'.format(resp))

        resp_data = await resp.read()
        if resp.status // 100 != 2:
            raise Exception(
                'Callback failed with {!r}, got {}'.format(resp, resp_data.decode('utf-8'))
            ) from sys.exc_info()[1]

        logger.info('Callback succeeded with {}'.format(resp_data.decode('utf-8')))

    return metadata, created

    def prepare(self, *args, **kwargs):
        # TODO Find a nicer way to handle this
        if self.request.method.lower() == "options":
            return

        self.path = self.path_kwargs["path"] or "/"
        provider = self.path_kwargs["provider"]
        self.resource = self.path_kwargs["resource"]

        if self.request.method.lower() in self.VALIDATORS:
            # create must validate before accepting files
            getattr(self, self.VALIDATORS[self.request.method.lower()])()

        self.auth = yield from auth_handler.get(self.resource, provider, self.request)
        self.provider = utils.make_provider(
            provider, self.auth["auth"], self.auth["credentials"], self.auth["settings"]
        )
        self.path = yield from self.provider.validate_path(self.path)

        # The one special case
        if self.request.method == "PUT" and self.path.is_file:
            yield from self.prepare_stream()
        else:
            self.stream = None
        self.body = b""

    async def prepare(self):
        """Builds an MFR provider instance, to which it passes the the ``url`` query parameter.
        From that, the file metadata is extracted.  Also builds cached waterbutler providers.
        """
        if self.request.method == 'OPTIONS':
            return

        try:
            self.url = self.request.query_arguments['url'][0].decode('utf-8')
        except KeyError:
            raise exceptions.ProviderError('"url" is a required argument.', code=400)

        self.provider = utils.make_provider(
            settings.PROVIDER_NAME,
            self.request,
            self.url
        )

        self.metadata = await self.provider.metadata()
        self.extension_metrics.add('ext', self.metadata.ext)

        self.cache_provider = waterbutler.core.utils.make_provider(
            settings.CACHE_PROVIDER_NAME,
            {},  # User information which can be left blank
            settings.CACHE_PROVIDER_CREDENTIALS,
            settings.CACHE_PROVIDER_SETTINGS
        )

        self.local_cache_provider = waterbutler.core.utils.make_provider(
            'filesystem', {}, {}, settings.LOCAL_CACHE_PROVIDER_SETTINGS
        )

        self.source_file_id = uuid.uuid4()

 async def make_provider(self, provider, prefix='', **kwargs):
     payload = await auth_handler.fetch(
         self.request,
         dict(kwargs, provider=provider, action=self.action + prefix)
     )
     self.auth = payload
     return utils.make_provider(provider, **payload)

def _upload_parity(path, credentials, settings):
    _, name = os.path.split(path)
    provider_name = settings.get('provider')
    provider = make_provider(provider_name, {}, credentials, settings)
    with open(path, 'rb') as file_pointer:
        stream = streams.FileStreamReader(file_pointer)
        yield from provider.upload(stream, path='/' + name)

 def make_provider(self, provider, prefix='', **kwargs):
     payload = yield from auth_handler.fetch(
         self.request,
         dict(kwargs, provider=provider, action=self.action + prefix)
     )
     self.auth = payload
     self.callback_url = payload.pop('callback_url')
     return utils.make_provider(provider, **payload)

    def make_provider(self, settings):
        """Requests on different files may need to use different providers,
        instances, e.g. when different files lives in different containers
        within a provider. This helper creates a single-use provider instance
        that optionally overrides the settings.

        :param dict settings: Overridden settings
        """
        return utils.make_provider(self.provider_name, self.auth, self.credentials["storage"], self.settings["storage"])

async def _upload_parity(path, credentials, settings):
    _, name = os.path.split(path)
    provider_name = settings.get('provider')
    provider = make_provider(provider_name, {}, credentials, settings)
    with open(path, 'rb') as file_pointer:
        stream = streams.FileStreamReader(file_pointer)
        await provider.upload(
            stream,
            (await provider.validate_path('/' + name))
        )

def move(src_bundle, dest_bundle, callback_url, auth, start_time=None, **kwargs):
    start_time = start_time or time.time()
    src_path, src_provider = src_bundle.pop('path'), utils.make_provider(**src_bundle.pop('provider'))
    dest_path, dest_provider = dest_bundle.pop('path'), utils.make_provider(**dest_bundle.pop('provider'))

    data = {
        'errors': [],
        'action': 'move',
        'source': dict(src_bundle, **{
            'path': src_path.path,
            'name': src_path.name,
            'materialized': str(src_path),
            'provider': src_provider.NAME,
        }),
        'destination': dict(dest_bundle, **{
            'path': dest_path.path,
            'name': dest_path.name,
            'materialized': str(dest_path),
            'provider': dest_provider.NAME,
        }),
        'auth': auth['auth'],
    }

    logger.info('Starting moving {!r}, {!r} to {!r}, {!r}'.format(src_path, src_provider, dest_path, dest_provider))

    try:
        metadata, created = yield from src_provider.move(dest_provider, src_path, dest_path, **kwargs)
    except Exception as e:
        logger.error('Move failed with error {!r}'.format(e))
        data.update({'errors': [e.__repr__()]})
        raise  # Ensure sentry sees this
    else:
        logger.info('Move succeeded')
        data.update({'destination': dict(src_bundle, **metadata.serialized())})
    finally:
        resp = yield from utils.send_signed_request('PUT', callback_url, dict(data, **{
            'time': time.time() + 60,
            'email': time.time() - start_time > settings.WAIT_TIMEOUT
        }))
        logger.info('Callback returned {!r}'.format(resp))

    return metadata, created

    async def prepare(self, *args, **kwargs):
        method = self.request.method.lower()

        # TODO Find a nicer way to handle this
        if method == 'options':
            return

        self.arguments = {
            key: list_or_value(value)
            for key, value in self.request.query_arguments.items()
        }

        # Going with version as its the most correct term
        # TODO Change all references of revision to version @chrisseto
        # revisions will still be accepted until necessary changes are made to OSF
        self.requested_version = (self.get_query_argument('version', default=None) or
                                  self.get_query_argument('revision', default=None))

        self.path = self.path_kwargs['path'] or '/'
        provider = self.path_kwargs['provider']
        self.resource = self.path_kwargs['resource']

        # pre-validator methods perform validations that can be performed before ensuring that the
        # path given by the url is valid.  An example would be making sure that a particular query
        # parameter matches and allowed value.  We do this because validating the path requires
        # issuing one or more API calls to the provider, and some providers are quite stingy with
        # their rate limits.
        if method in self.PRE_VALIDATORS:
            getattr(self, self.PRE_VALIDATORS[method])()

        # Delay setup of the provider when method is post, as we need to evaluate the json body
        # action.
        if method != 'post':
            self.auth = await auth_handler.get(self.resource, provider, self.request,
                                               path=self.path, version=self.requested_version)
            self.provider = utils.make_provider(provider, self.auth['auth'],
                                                self.auth['credentials'], self.auth['settings'])
            self.path = await self.provider.validate_v1_path(self.path, **self.arguments)

        self.target_path = None

        # post-validator methods perform validations that expect that the path given in the url has
        # been verified for existence and type.
        if method in self.POST_VALIDATORS:
            await getattr(self, self.POST_VALIDATORS[method])()

        # The one special case
        if method == 'put' and self.target_path.is_file:
            await self.prepare_stream()
        else:
            self.stream = None
        self.body = b''

        self.add_header('X-WATERBUTLER-REQUEST-ID', str(uuid.uuid4()))

    async def prepare(self):
        self.arguments = {key: list_or_value(value) for key, value in self.request.query_arguments.items()}
        try:
            self.arguments["action"] = self.ACTION_MAP[self.request.method]
        except KeyError:
            return

        self.payload = await auth_handler.fetch(self.request, self.arguments)

        self.provider = utils.make_provider(
            self.arguments["provider"], self.payload["auth"], self.payload["credentials"], self.payload["settings"]
        )

        self.path = await self.provider.validate_path(**self.arguments)
        self.arguments["path"] = self.path  # TODO Not this

    async def prepare(self, *args, **kwargs):
        method = self.request.method.lower()

        # TODO Find a nicer way to handle this
        if method == 'options':
            return

        self.arguments = {
            key: list_or_value(value)
            for key, value in self.request.query_arguments.items()
        }

        self.path = self.path_kwargs['path'] or '/'
        provider = self.path_kwargs['provider']
        self.resource = self.path_kwargs['resource']

        # pre-validator methods perform validations that can be performed before ensuring that the
        # path given by the url is valid.  An example would be making sure that a particular query
        # parameter matches and allowed value.  We do this because validating the path requires
        # issuing one or more API calls to the provider, and some providers are quite stingy with
        # their rate limits.
        if method in self.PRE_VALIDATORS:
            getattr(self, self.PRE_VALIDATORS[method])()

        self.auth = await auth_handler.get(self.resource, provider, self.request)
        self.provider = utils.make_provider(provider, self.auth['auth'], self.auth['credentials'], self.auth['settings'])
        self.path = await self.provider.validate_v1_path(self.path, **self.arguments)

        self.target_path = None

        # post-validator methods perform validations that expect that the path given in the url has
        # been verified for existence and type.
        if method in self.POST_VALIDATORS:
            await getattr(self, self.POST_VALIDATORS[method])()

        # The one special case
        if method == 'put' and self.target_path.is_file:
            await self.prepare_stream()
        else:
            self.stream = None
        self.body = b''

    def prepare(self):
        self.arguments = {
            key: list_or_value(value)
            for key, value in self.request.query_arguments.items()
        }
        try:
            self.arguments['action'] = self.ACTION_MAP[self.request.method]
        except KeyError:
            return

        self.payload = yield from auth_handler.fetch(self.request, self.arguments)

        self.provider = utils.make_provider(
            self.arguments['provider'],
            self.payload['auth'],
            self.payload['credentials'],
            self.payload['settings'],
        )

        self.path = yield from self.provider.validate_path(**self.arguments)
        self.arguments['path'] = self.path  # TODO Not this

    def prepare(self, *args, **kwargs):
        # TODO Find a nicer way to handle this
        if self.request.method.lower() == 'options':
            return

        self.path = self.path_kwargs['path'] or '/'
        provider = self.path_kwargs['provider']
        self.resource = self.path_kwargs['resource']

        if self.request.method.lower() in self.VALIDATORS:
            # create must validate before accepting files
            getattr(self, self.VALIDATORS[self.request.method.lower()])()

        self.auth = yield from auth_handler.get(self.resource, provider, self.request)
        self.provider = utils.make_provider(provider, self.auth['auth'], self.auth['credentials'], self.auth['settings'])
        self.path = yield from self.provider.validate_path(self.path)

        # The one special case
        if self.request.method == 'PUT' and self.path.is_file:
            yield from self.prepare_stream()
        else:
            self.stream = None
        self.body = b''

    def prepare(self):
        if self.request.method == 'OPTIONS':
            return

        self.url = self.request.query_arguments['url'][0].decode('utf-8')

        self.provider = utils.make_provider(
            settings.PROVIDER_NAME,
            self.request,
            self.url
        )

        self.metadata = yield from self.provider.metadata()

        self.cache_provider = waterbutler.core.utils.make_provider(
            settings.CACHE_PROVIDER_NAME,
            {},  # User information which can be left blank
            settings.CACHE_PROVIDER_CREDENTIALS,
            settings.CACHE_PROVIDER_SETTINGS
        )

        self.local_cache_provider = waterbutler.core.utils.make_provider(
            'filesystem', {}, {}, settings.LOCAL_CACHE_PROVIDER_SETTINGS
        )

    async def move_or_copy(self):
        """Copy, move, and rename files and folders.

        **Auth actions**: ``copy``, ``move``, or ``rename``

        **Provider actions**: ``copy`` or ``move``

        *Auth actions* come from the ``action`` body parameter in the request and are used by the
        auth handler.

        *Provider actions* are determined from the *auth action*.  A "rename" is a special case of
        the "move" provider action that implies that the destination resource, provider, and parent
        path will all be the same as the source.
        """

        # Force the json body to load into memory
        await self.request.body

        auth_action = self.json.get('action', 'null')
        if auth_action not in ('copy', 'move', 'rename'):
            raise exceptions.InvalidParameters('Auth action must be "copy", "move", or "rename", '
                                               'not "{}"'.format(auth_action))

        # Provider setup is delayed so the provider action can be updated from the auth action.
        provider = self.path_kwargs.get('provider', '')
        provider_action = auth_action
        if auth_action == 'rename':
            if not self.json.get('rename', ''):
                raise exceptions.InvalidParameters('"rename" field is required for renaming')
            provider_action = 'move'

        self.auth = await auth_handler.get(
            self.resource,
            provider,
            self.request,
            action=auth_action,
            auth_type=AuthType.SOURCE,
            path=self.path,
            version=self.requested_version,
        )
        self.provider = make_provider(
            provider,
            self.auth['auth'],
            self.auth['credentials'],
            self.auth['settings']
        )
        self.path = await self.provider.validate_v1_path(self.path, **self.arguments)

        if auth_action == 'rename':  # 'rename' implies the file/folder does not change location
            self.dest_auth = self.auth
            self.dest_provider = self.provider
            self.dest_path = self.path.parent
            self.dest_resource = self.resource
        else:
            path = self.json.get('path', None)
            if path is None:
                raise exceptions.InvalidParameters('"path" field is required for moves or copies')
            if not path.endswith('/'):
                raise exceptions.InvalidParameters(
                    '"path" field requires a trailing slash to indicate it is a folder'
                )

            # TODO optimize for same provider and resource

            # for copy action, `auth_action` is the same as `provider_action`
            if auth_action == 'copy' and self.path.is_root and not self.json.get('rename'):
                raise exceptions.InvalidParameters('"rename" field is required for copying root')

            # Note: attached to self so that _send_hook has access to these
            self.dest_resource = self.json.get('resource', self.resource)
            self.dest_auth = await auth_handler.get(
                self.dest_resource,
                self.json.get('provider', self.provider.NAME),
                self.request,
                action=auth_action,
                auth_type=AuthType.DESTINATION,
            )
            self.dest_provider = make_provider(
                self.json.get('provider', self.provider.NAME),
                self.dest_auth['auth'],
                self.dest_auth['credentials'],
                self.dest_auth['settings']
            )
            self.dest_path = await self.dest_provider.validate_path(**self.json)

        if not getattr(self.provider, 'can_intra_' + provider_action)(self.dest_provider, self.path):
            # this weird signature syntax courtesy of py3.4 not liking trailing commas on kwargs
            conflict = self.json.get('conflict', DEFAULT_CONFLICT)
            result = await getattr(tasks, provider_action).adelay(
                rename=self.json.get('rename'),
                conflict=conflict,
                request=remote_logging._serialize_request(self.request),
                *self.build_args()
            )
            metadata, created = await tasks.wait_on_celery(result)
        else:
            metadata, created = (
                await tasks.backgrounded(
                    getattr(self.provider, provider_action),
                    self.dest_provider,
#.........这里部分代码省略.........

    def move_or_copy(self):
        # Force the json body to load into memory
        yield self.request.body

        if self.json.get('action') not in ('copy', 'move', 'rename'):
            # Note: null is used as the default to avoid python specific error messages
            raise exceptions.InvalidParameters('Action must be copy, move or rename, not {}'.format(self.json.get('action', 'null')))

        if self.json['action'] == 'rename':
            action = 'move'
            self.dest_auth = self.auth
            self.dest_provider = self.provider
            self.dest_path = self.path.parent
        else:
            if 'path' not in self.json:
                raise exceptions.InvalidParameters('Path is required for moves or copies')

            action = self.json['action']

            # Note: attached to self so that _send_hook has access to these
            self.dest_resource = self.json.get('resource', self.resource)

            # TODO optimize for same provider and resource
            self.dest_auth = yield from auth_handler.get(
                self.dest_resource,
                self.json.get('provider', self.provider.NAME),
                self.request
            )

            self.dest_provider = make_provider(
                self.json.get('provider', self.provider.NAME),
                self.dest_auth['auth'],
                self.dest_auth['credentials'],
                self.dest_auth['settings']
            )

            self.dest_path = yield from self.dest_provider.validate_path(self.json['path'])

        if not getattr(self.provider, 'can_intra_' + action)(self.dest_provider, self.path):
            result = yield from getattr(tasks, action).adelay(*self.build_args(self.dest_provider, self.dest_path))
            metadata, created = yield from tasks.wait_on_celery(result)
        else:
            metadata, created = (
                yield from tasks.backgrounded(
                    getattr(self.provider, action),
                    self.dest_provider,
                    self.path,
                    self.dest_path,
                    rename=self.json.get('rename'),
                    conflict=self.json.get('conflict', 'replace'),
                )
            )

        metadata = metadata.serialized()

        if created:
            self.set_status(201)
        else:
            self.set_status(200)

        self.write(metadata)

    def move_or_copy(self):
        # Force the json body to load into memory
        yield self.request.body

        if self.json.get("action") not in ("copy", "move", "rename"):
            # Note: null is used as the default to avoid python specific error messages
            raise exceptions.InvalidParameters(
                "Action must be copy, move or rename, not {}".format(self.json.get("action", "null"))
            )

        if self.json["action"] == "rename":
            if not self.json.get("rename"):
                raise exceptions.InvalidParameters("Rename is required for renaming")
            action = "move"
            self.dest_auth = self.auth
            self.dest_provider = self.provider
            self.dest_path = self.path.parent
            self.dest_resource = self.resource
        else:
            if "path" not in self.json:
                raise exceptions.InvalidParameters("Path is required for moves or copies")

            action = self.json["action"]

            # Note: attached to self so that _send_hook has access to these
            self.dest_resource = self.json.get("resource", self.resource)

            # TODO optimize for same provider and resource
            self.dest_auth = yield from auth_handler.get(
                self.dest_resource, self.json.get("provider", self.provider.NAME), self.request
            )

            self.dest_provider = make_provider(
                self.json.get("provider", self.provider.NAME),
                self.dest_auth["auth"],
                self.dest_auth["credentials"],
                self.dest_auth["settings"],
            )

            self.dest_path = yield from self.dest_provider.validate_path(self.json["path"])

        if not getattr(self.provider, "can_intra_" + action)(self.dest_provider, self.path):
            # this weird signature syntax courtesy of py3.4 not liking trailing commas on kwargs
            result = yield from getattr(tasks, action).adelay(
                rename=self.json.get("rename"), conflict=self.json.get("conflict", DEFAULT_CONFLICT), *self.build_args()
            )
            metadata, created = yield from tasks.wait_on_celery(result)
        else:
            metadata, created = (
                yield from tasks.backgrounded(
                    getattr(self.provider, action),
                    self.dest_provider,
                    self.path,
                    self.dest_path,
                    rename=self.json.get("rename"),
                    conflict=self.json.get("conflict", DEFAULT_CONFLICT),
                )
            )

        if created:
            self.set_status(201)
        else:
            self.set_status(200)

        self.write({"data": metadata.json_api_serialized(self.dest_resource)})