migrate use of upload_fobj to use transfer

[skshetry]

Migrate away from upload_fobj to fs.utils.transfer
Also extend fs.utils.transfer to support file object upload and bytes upload.

Pre-requisite for merging fs.upload and fs.upload_fobj together. fs.upload_fobj won't be able to support callback.get_size as we don't have that information in the fs.upload as there's no src_fs and src_info to query to, so we are achieving that consistency in terms of callback support with the use of fs.utils.transfer.

Also pre-requisite for #6546.

[skshetry]

This is a bit overloaded though, but moving with it for now.

[skshetry]

fs.utils.transfer is the only place where we use upload_fobj now.

[skshetry]

@pmrowla, what do you think of this? Does transfer fit here, or should it use more primitive functions like pipe_file?
(although I want to use less API as possible)

Thinking about it again, I have doubts about this here, I only did it for simplifying the exception handling. 😅

[pmrowla]

The upload_fobj for reference objects isn't really a file transfer, it's just a write() of the serialized reference data from memory.

It seems to me like we should still have a distinction between a file transfer and a direct binary data write/upload, since in this case there really isn't a from_fs and from_info that we are transferring.

---

One alternative would maybe be adding something like ReferenceHashFile.to_memfile() where it returns a memfs://some_tmp_pathname path to a MemoryFileSystem entry (containing the serialized data). And then we could do an actual "file transfer" using the memfs and memfs_path_info without needing the content field?

(but this seems a bit more convoluted than just having an fs.utils.write method)

[skshetry]

yeah, initially I started with two methods, but at the end fs.utils.write is going to be equivalent to upload_fobj, so I don't see much difference there. With transfer, I tried to push the problem of getting size for the callback support inside the transfer itself, rather than putting that burden on the caller.

We do the following when transferring files straight to the remote, which is just a file transfer at the end:

 with fs.open(path_info, mode="rb", chunk_size=fs.CHUNK_SIZE) as stream:
         stream = HashedStreamReader(stream)
         upload_odb.fs.upload_fobj(
             stream, tmp_info, desc=path_info.name, size=fs.getsize(path_info)
         )

But maybe the best thing to do here is put that burden on the caller itself, rather than trying to generalize it in transfer.

[skshetry]

Closing this PR as we'll likely put the burden of doing callback.set_size on the caller itself.

See #6558 (comment).