def custom_transfer(transfer_client, source_ep, dest_ep, path_list, interval=DEFAULT_INTERVAL,
                    inactivity_time=DEFAULT_INACTIVITY_TIME, notify=True):
    """Perform a Globus Transfer.

    Arguments:
        transfer_client (TransferClient): An authenticated Transfer client.
        source_ep (str): The source Globus Endpoint ID.
        dest_ep (str): The destination Globus Endpoint ID.
        path_list (list of tuple of 2 str): A list of tuples containing the paths to transfer as
                ``(source, destination)``.

                **Example**::

                    [("/source/files/file.dat", "/dest/mydocs/doc.dat"),
                     ("/source/all_reports/", "/dest/reports/")]

        interval (int): Number of seconds to wait before polling Transfer status.
                Minimum ``1``. **Default**: ``DEFAULT_INTERVAL``.
        inactivity_time (int): Number of seconds a Transfer is allowed to go without progress
                before being cancelled. **Default**: ``DEFAULT_INACTIVITY_TIME``.
        notify (bool): When ``True``, trigger a notification email from Globus to the user when
                the Transfer succeeds or fails. When ``False``, disable the notification.
                **Default**: ``True``.

    Yields:
        dict: An error from the transfer, or (last) a success status.

    Accepts via ``.send()``:
        *bool*: ``True``: Continue the Transfer
                ``False``: Cancel the Transfer
                **Default**: ``True``
    """
    # TODO: (LW) Handle transfers with huge number of files
    # If a TransferData object is too large, Globus might timeout
    #   before it can be completely uploaded.
    # So, we need to be able to check the size of the TD object and, if need be, send it early.
    if interval < 1:
        interval = 1
    deadline = datetime.utcfromtimestamp(int(time.time()) + inactivity_time)
    tdata = globus_sdk.TransferData(transfer_client, source_ep, dest_ep,
                                    deadline=deadline, verify_checksum=True,
                                    notify_on_succeeded=notify, notify_on_failed=notify,
                                    notify_on_inactive=notify)
    for item in path_list:
        # Check if source path is directory or missing
        try:
            transfer_client.operation_ls(source_ep, path=item[0])
            source_is_dir = True
        except globus_sdk.exc.TransferAPIError as e:
            # If error indicates path exists but is not dir, is not dir
            if e.code == "ExternalError.DirListingFailed.NotDirectory":
                source_is_dir = False
            # Too many files in dir indicates is dir
            elif e.code == "ExternalError.DirListingFailed.SizeLimit":
                source_is_dir = True
            # Not found is real error
            elif e.code == "ClientError.NotFound":
                raise globus_sdk.GlobusError("Path '{}' not found on source endpoint '{}'"
                                             .format(item[0], source_ep))
            # Else, retry on parent dir
            else:
                try:
                    parent, item_name = os.path.split(item[0])
                    parent_ls = transfer_client.operation_ls(source_ep, path=parent)
                    type_list = [x["type"] for x in parent_ls["DATA"] if x["name"] == item_name]
                    if len(type_list) < 1:
                        raise globus_sdk.GlobusError("No items with name '{}' in path '{}' on "
                                                     "endpoint '{}'"
                                                     .format(item_name, parent, source_ep))
                    elif len(type_list) > 1:
                        raise globus_sdk.GlobusError("Multiple items with name '{}' in path '{}'"
                                                     "on endpoint '{}'"
                                                     .format(item_name, parent, source_ep))
                    item_type = type_list[0]
                    if item_type == "dir":
                        source_is_dir = True
                    elif item_type == "file":
                        source_is_dir = False
                    else:
                        raise ValueError("Path '{}' does not lead to a file or a directory ({})"
                                         .format(item[0], item_type))
                except globus_sdk.exc.TransferAPIError as e:
                    # Size limit means we can't figure out this path
                    if e.code == "ExternalError.DirListingFailed.SizeLimit":
                        raise globus_sdk.GlobusError("Unable to check type of {}".format(item[0]))
                    # Not found is still an error
                    elif e.code == "ClientError.NotFound":
                        raise globus_sdk.GlobusError("Parent path '{}' not found on source "
                                                     "endpoint '{}'".format(item[0], source_ep))
                    else:
                        raise

        # Check if dest path is directory
        dest_exists = False
        try:
            transfer_client.operation_ls(dest_ep, path=item[1])
            dest_exists = True
            dest_is_dir = True
        except globus_sdk.exc.TransferAPIError as e:
            if e.code == "ExternalError.DirListingFailed.NotDirectory":
                dest_exists = True
                dest_is_dir = False
            elif e.code == "ExternalError.DirListingFailed.SizeLimit":
                dest_exists = True
                dest_is_dir = True
            elif e.code == "ClientError.NotFound":
                # Destination will be created, not an issue if not found
                pass
            else:
                try:
                    parent, item_name = os.path.split(item[1])
                    parent_ls = transfer_client.operation_ls(dest_ep, path=parent)
                    type_list = [x["type"] for x in parent_ls["DATA"] if x["name"] == item_name]
                    if len(type_list) < 1:
                        raise globus_sdk.GlobusError("No items with name '{}' in path '{}' on "
                                                     "endpoint '{}'"
                                                     .format(item_name, parent, dest_ep))
                    elif len(type_list) > 1:
                        raise globus_sdk.GlobusError("Multiple items with name '{}' in path '{}'"
                                                     "on endpoint '{}'"
                                                     .format(item_name, parent, dest_ep))
                    item_type = type_list[0]
                    if item_type == "dir":
                        dest_exists = True
                        dest_is_dir = True
                    elif item_type == "file":
                        dest_exists = True
                        dest_is_dir = False
                    else:
                        # Assume we're overwriting whatever dest is, as if it doesn't exist
                        pass
                except globus_sdk.exc.TransferAPIError as e:
                    # Size limit means we can't figure out this path
                    if e.code == "ExternalError.DirListingFailed.SizeLimit":
                        raise globus_sdk.GlobusError("Unable to check type of {}".format(item[0]))
                    # Not found is not our problem for dest
                    elif e.code == "ClientError.NotFound":
                        pass
                    else:
                        raise
        # Transfer dir
        # Short-circuit OR/AND eval means if not dest_exists, dest_is_dir can be unassigned
        if source_is_dir and (not dest_exists or dest_is_dir):
            tdata.add_item(item[0], item[1], recursive=True)
        # Transfer non-dir
        elif not source_is_dir and (not dest_exists or not dest_is_dir):
            tdata.add_item(item[0], item[1])
        # Transfer non-dir into dir
        # TODO: Is this logic user-friendly or is it surprising?
        # Take non-dir source filename, Transfer to dest dir+filename
        elif not source_is_dir and (dest_exists and dest_is_dir):
            new_dest = os.path.join(item[1], os.path.basename(item[0]))
            tdata.add_item(item[0], new_dest)
        # Malformed - Cannot transfer dir into non-dir
        else:
            raise globus_sdk.GlobusError("Cannot transfer a directory into a file: "
                                         + str(item))

    res = transfer_client.submit_transfer(tdata)
    if res["code"] != "Accepted":
        raise globus_sdk.GlobusError("Failed to transfer files: Transfer " + res["code"])

    error_timestamps = set()
    # while Transfer is active
    while not transfer_client.task_wait(res["task_id"],
                                        timeout=interval, polling_interval=interval):
        for event in transfer_client.task_event_list(res["task_id"]):
            # Only process error events that have not been presented to the user
            # Events do not have UUIDs, so if there are multiple simultaneous errors
            #   only the last (chronologically) error will be processed
            if event["is_error"] and event["time"] not in error_timestamps:
                error_timestamps.add(event["time"])
                ret_event = event.data
                # yield value should always have success: bool
                ret_event["success"] = False
                ret_event["finished"] = False
                # User can cancel Transfer with .send(False)
                cont = yield ret_event
                if cont is False:
                    transfer_client.cancel_task(res["task_id"])
                    # Wait until Transfer is no longer active after cancellation
                    while not transfer_client.task_wait(res["task_id"],
                                                        timeout=1, polling_interval=1):
                        pass
                    break
            # If progress has been made, move deadline forward
            elif event["code"] == "PROGRESS":
                new_deadline = datetime.utcfromtimestamp(int(time.time()) + inactivity_time)
                new_doc = {
                    "DATA_TYPE": "task",
                    "deadline": str(new_deadline)
                }
                transfer_client.update_task(res["task_id"], new_doc)
    # Transfer is no longer active; now check if succeeded
    task = transfer_client.get_task(res["task_id"]).data
    task["success"] = (task["status"] == "SUCCEEDED")
    task["finished"] = True
    yield task
def custom_transfer ( transfer_client , source_ep , dest_ep , path_list , interval = DEFAULT_INTERVAL , inactivity_time = DEFAULT_INACTIVITY_TIME , notify = True ) : if interval < 1 : interval = 1 deadline = datetime . utcfromtimestamp ( int ( time . time ( ) ) + inactivity_time ) tdata = globus_sdk . TransferData ( transfer_client , source_ep , dest_ep , deadline = deadline , verify_checksum = True , notify_on_succeeded = notify , notify_on_failed = notify , notify_on_inactive = notify ) for item in path_list : try : transfer_client . operation_ls ( source_ep , path = item [ 0 ] ) source_is_dir = True except globus_sdk . exc . TransferAPIError as e : if e . code == "ExternalError.DirListingFailed.NotDirectory" : source_is_dir = False elif e . code == "ExternalError.DirListingFailed.SizeLimit" : source_is_dir = True elif e . code == "ClientError.NotFound" : raise globus_sdk . GlobusError ( "Path '{}' not found on source endpoint '{}'" . format ( item [ 0 ] , source_ep ) ) else : try : parent , item_name = os . path . split ( item [ 0 ] ) parent_ls = transfer_client . operation_ls ( source_ep , path = parent ) type_list = [ x [ "type" ] for x in parent_ls [ "DATA" ] if x [ "name" ] == item_name ] if len ( type_list ) < 1 : raise globus_sdk . GlobusError ( "No items with name '{}' in path '{}' on " "endpoint '{}'" . format ( item_name , parent , source_ep ) ) elif len ( type_list ) > 1 : raise globus_sdk . GlobusError ( "Multiple items with name '{}' in path '{}'" "on endpoint '{}'" . format ( item_name , parent , source_ep ) ) item_type = type_list [ 0 ] if item_type == "dir" : source_is_dir = True elif item_type == "file" : source_is_dir = False else : raise ValueError ( "Path '{}' does not lead to a file or a directory ({})" . format ( item [ 0 ] , item_type ) ) except globus_sdk . exc . TransferAPIError as e : if e . code == "ExternalError.DirListingFailed.SizeLimit" : raise globus_sdk . GlobusError ( "Unable to check type of {}" . format ( item [ 0 ] ) ) elif e . code == "ClientError.NotFound" : raise globus_sdk . GlobusError ( "Parent path '{}' not found on source " "endpoint '{}'" . format ( item [ 0 ] , source_ep ) ) else : raise dest_exists = False try : transfer_client . operation_ls ( dest_ep , path = item [ 1 ] ) dest_exists = True dest_is_dir = True except globus_sdk . exc . TransferAPIError as e : if e . code == "ExternalError.DirListingFailed.NotDirectory" : dest_exists = True dest_is_dir = False elif e . code == "ExternalError.DirListingFailed.SizeLimit" : dest_exists = True dest_is_dir = True elif e . code == "ClientError.NotFound" : pass else : try : parent , item_name = os . path . split ( item [ 1 ] ) parent_ls = transfer_client . operation_ls ( dest_ep , path = parent ) type_list = [ x [ "type" ] for x in parent_ls [ "DATA" ] if x [ "name" ] == item_name ] if len ( type_list ) < 1 : raise globus_sdk . GlobusError ( "No items with name '{}' in path '{}' on " "endpoint '{}'" . format ( item_name , parent , dest_ep ) ) elif len ( type_list ) > 1 : raise globus_sdk . GlobusError ( "Multiple items with name '{}' in path '{}'" "on endpoint '{}'" . format ( item_name , parent , dest_ep ) ) item_type = type_list [ 0 ] if item_type == "dir" : dest_exists = True dest_is_dir = True elif item_type == "file" : dest_exists = True dest_is_dir = False else : pass except globus_sdk . exc . TransferAPIError as e : if e . code == "ExternalError.DirListingFailed.SizeLimit" : raise globus_sdk . GlobusError ( "Unable to check type of {}" . format ( item [ 0 ] ) ) elif e . code == "ClientError.NotFound" : pass else : raise if source_is_dir and ( not dest_exists or dest_is_dir ) : tdata . add_item ( item [ 0 ] , item [ 1 ] , recursive = True ) elif not source_is_dir and ( not dest_exists or not dest_is_dir ) : tdata . add_item ( item [ 0 ] , item [ 1 ] ) elif not source_is_dir and ( dest_exists and dest_is_dir ) : new_dest = os . path . join ( item [ 1 ] , os . path . basename ( item [ 0 ] ) ) tdata . add_item ( item [ 0 ] , new_dest ) else : raise globus_sdk . GlobusError ( "Cannot transfer a directory into a file: " + str ( item ) ) res = transfer_client . submit_transfer ( tdata ) if res [ "code" ] != "Accepted" : raise globus_sdk . GlobusError ( "Failed to transfer files: Transfer " + res [ "code" ] ) error_timestamps = set ( ) while not transfer_client . task_wait ( res [ "task_id" ] , timeout = interval , polling_interval = interval ) : for event in transfer_client . task_event_list ( res [ "task_id" ] ) : if event [ "is_error" ] and event [ "time" ] not in error_timestamps : error_timestamps . add ( event [ "time" ] ) ret_event = event . data ret_event [ "success" ] = False ret_event [ "finished" ] = False cont = yield ret_event if cont is False : transfer_client . cancel_task ( res [ "task_id" ] ) while not transfer_client . task_wait ( res [ "task_id" ] , timeout = 1 , polling_interval = 1 ) : pass break elif event [ "code" ] == "PROGRESS" : new_deadline = datetime . utcfromtimestamp ( int ( time . time ( ) ) + inactivity_time ) new_doc = { "DATA_TYPE" : "task" , "deadline" : str ( new_deadline ) } transfer_client . update_task ( res [ "task_id" ] , new_doc ) task = transfer_client . get_task ( res [ "task_id" ] ) . data task [ "success" ] = ( task [ "status" ] == "SUCCEEDED" ) task [ "finished" ] = True yield task
Perform a Globus Transfer.

    Arguments:
        transfer_client (TransferClient): An authenticated Transfer client.
        source_ep (str): The source Globus Endpoint ID.
        dest_ep (str): The destination Globus Endpoint ID.
        path_list (list of tuple of 2 str): A list of tuples containing the paths to transfer as
                ``(source, destination)``.

                **Example**::

                    [("/source/files/file.dat", "/dest/mydocs/doc.dat"),
                     ("/source/all_reports/", "/dest/reports/")]

        interval (int): Number of seconds to wait before polling Transfer status.
                Minimum ``1``. **Default**: ``DEFAULT_INTERVAL``.
        inactivity_time (int): Number of seconds a Transfer is allowed to go without progress
                before being cancelled. **Default**: ``DEFAULT_INACTIVITY_TIME``.
        notify (bool): When ``True``, trigger a notification email from Globus to the user when
                the Transfer succeeds or fails. When ``False``, disable the notification.
                **Default**: ``True``.

    Yields:
        dict: An error from the transfer, or (last) a success status.

    Accepts via ``.send()``:
        *bool*: ``True``: Continue the Transfer
                ``False``: Cancel the Transfer
                **Default**: ``True``