[fs] basic sync tool

hail/Makefile

@@ -186,6 +186,8 @@ pytest-inter-cloud: upload-remote-test-resources install-editable
           HAIL_TEST_S3_BUCKET=hail-test-dy5rg \
           HAIL_TEST_AZURE_ACCOUNT=hailtest \
           HAIL_TEST_AZURE_CONTAINER=hail-test-4nxei \
+          HAIL_TEST_AZURE_RESGRP=hail-dev \
+          HAIL_TEST_AZURE_SUBID=22cd45fe-f996-4c51-af67-ef329d977519 \
           $(HAIL_PYTHON3) -m pytest \
             -Werror:::hail -Werror:::hailtop -Werror::ResourceWarning \
             --log-cli-level=INFO \

hail/python/hailtop/aiocloud/aioaws/fs.py

@@ -93,8 +93,11 @@ def __init__(self, head_object_resp, url: str):
         self.head_object_resp = head_object_resp
         self._url = url
 
+    def __repr__(self):
+        return f'S3HeadObjectFileStatus({self.head_object_resp}, {self._url})'
+
     def basename(self) -> str:
-        return os.path.basename(self._url.rstrip('/'))
+        return os.path.basename(self._url)
 
     def url(self) -> str:
         return self._url
@@ -121,8 +124,11 @@ def __init__(self, item: Dict[str, Any], url: str):
         self._item = item
         self._url = url
 
+    def __repr__(self):
+        return f'S3ListFilesFileStatus({self._item}, {self._url})'
+
     def basename(self) -> str:
-        return os.path.basename(self._url.rstrip('/'))
+        return os.path.basename(self._url)
 
     def url(self) -> str:
         return self._url
@@ -195,18 +201,28 @@ def __init__(self, bucket: str, key: str, item: Optional[Dict[str, Any]]):
         self._item = item
         self._status: Optional[S3ListFilesFileStatus] = None
 
+    def __repr__(self):
+        return f'S3FileListEntry({self._bucket}, {self._key}, {self._item})'
+
     def basename(self) -> str:
-        return os.path.basename(self._key.rstrip('/'))
+        object_name = self._key
+        if self._is_dir():
+            assert object_name[-1] == '/'
+            object_name = object_name[:-1]
+        return os.path.basename(object_name)
 
     async def url(self) -> str:
         return f's3://{self._bucket}/{self._key}'
 
     async def is_file(self) -> bool:
         return self._item is not None
 
-    async def is_dir(self) -> bool:
+    def _is_dir(self) -> bool:
         return self._item is None
 
+    async def is_dir(self) -> bool:
+        return self._is_dir()
+
     async def status(self) -> FileStatus:
         if self._status is None:
             if self._item is None:

hail/python/hailtop/aiocloud/aioazure/fs.py

@@ -243,8 +243,15 @@ def __init__(self, url: 'AzureAsyncFSURL', blob_props: Optional[BlobProperties])
         self._blob_props = blob_props
         self._status: Optional[AzureFileStatus] = None
 
+    def __repr__(self):
+        return f'AzureFileListEntry({self._url}, {self._blob_props})'
+
     def basename(self) -> str:
-        return os.path.basename(self._url.base.rstrip('/'))
+        url_no_sas = self._url.base
+        if self._is_dir():
+            assert url_no_sas[-1] == '/'
+            url_no_sas = url_no_sas[:-1]
+        return os.path.basename(url_no_sas)
 
     async def url(self) -> str:
         return self._url.base
@@ -255,9 +262,12 @@ async def url_full(self) -> str:
     async def is_file(self) -> bool:
         return self._blob_props is not None
 
-    async def is_dir(self) -> bool:
+    def _is_dir(self) -> bool:
         return self._blob_props is None
 
+    async def is_dir(self) -> bool:
+        return self._is_dir()
+
     async def status(self) -> FileStatus:
         if self._status is None:
             if self._blob_props is None:
@@ -271,8 +281,11 @@ def __init__(self, blob_props: BlobProperties, url: 'AzureAsyncFSURL'):
         self.blob_props = blob_props
         self._url = url
 
+    def __repr__(self):
+        return f'AzureFileStatus({self.blob_props}, {self._url})'
+
     def basename(self) -> str:
-        return os.path.basename(self._url.base.rstrip('/'))
+        return os.path.basename(self._url.base)
 
     def url(self) -> str:
         return str(self._url)
@@ -546,9 +559,11 @@ async def isfile(self, url: str) -> bool:
     @handle_public_access_error
     async def isdir(self, url: str) -> bool:
         fs_url = self.parse_url(url, error_if_bucket=True)
-        assert not fs_url.path or fs_url.path.endswith('/'), fs_url.path
         client = await self.get_container_client(fs_url)
-        async for _ in client.walk_blobs(name_starts_with=fs_url.path, include=['metadata'], delimiter='/'):
+        prefix = fs_url.path
+        if prefix[-1] != '/':
+            prefix = prefix + '/'
+        async for _ in client.walk_blobs(name_starts_with=prefix, include=['metadata'], delimiter='/'):
             return True
         return False
 

hail/python/hailtop/aiocloud/aiogoogle/client/storage_client.py

@@ -80,12 +80,13 @@ async def _cleanup_future(fut: asyncio.Future):
 
 
 class InsertObjectStream(WritableStream):
-    def __init__(self, it: FeedableAsyncIterable[bytes], request_task: asyncio.Task[aiohttp.ClientResponse]):
+    def __init__(self, it: FeedableAsyncIterable[bytes], request_task: asyncio.Task[aiohttp.ClientResponse], url: str):
         super().__init__()
         self._it = it
         self._request_task = request_task
         self._value = None
         self._exit_stack = AsyncExitStack()
+        self.url = url
 
         async def cleanup_request_task():
             if not self._request_task.cancelled():
@@ -364,7 +365,7 @@ async def insert_object(self, bucket: str, name: str, **kwargs) -> WritableStrea
                     f'https://storage.googleapis.com/upload/storage/v1/b/{bucket}/o', retry=False, **kwargs
                 )
             )
-            return InsertObjectStream(it, request_task)
+            return InsertObjectStream(it, request_task, 'gs://' + bucket + '/' + name)
 
         # Write using resumable uploads.  See:
         # https://cloud.google.com/storage/docs/performing-resumable-uploads
@@ -461,8 +462,11 @@ def __init__(self, items: Dict[str, str], url: str):
         self._items = items
         self._url = url
 
+    def __repr__(self):
+        return f'GetObjectFileStatus({self._items}, {self._url})'
+
     def basename(self) -> str:
-        return os.path.basename(self._url.rstrip('/'))
+        return os.path.basename(self._url)
 
     def url(self) -> str:
         return self._url
@@ -487,18 +491,28 @@ def __init__(self, bucket: str, name: str, items: Optional[Dict[str, Any]]):
         self._items = items
         self._status: Optional[GetObjectFileStatus] = None
 
+    def __repr__(self):
+        return f'GoogleStorageFileListEntry({self._bucket}, {self._name}, {self._items})'
+
     def basename(self) -> str:
-        return os.path.basename(self._name.rstrip('/'))
+        object_name = self._name
+        if self._is_dir():
+            assert object_name[-1] == '/'
+            object_name = object_name[:-1]
+        return os.path.basename(object_name)
 
     async def url(self) -> str:
         return f'gs://{self._bucket}/{self._name}'
 
     async def is_file(self) -> bool:
         return self._items is not None
 
-    async def is_dir(self) -> bool:
+    def _is_dir(self) -> bool:
         return self._items is None
 
+    async def is_dir(self) -> bool:
+        return self._is_dir()
+
     async def status(self) -> FileStatus:
         if self._status is None:
             if self._items is None:
@@ -856,7 +870,9 @@ async def isfile(self, url: str) -> bool:
 
     async def isdir(self, url: str) -> bool:
         fsurl = self.parse_url(url, error_if_bucket=True)
-        assert not fsurl._path or fsurl.path.endswith('/'), fsurl._path
+        prefix = fsurl._path
+        if len(prefix) > 0 and prefix[-1] != '/':
+            prefix += '/'
         params = {'prefix': fsurl._path, 'delimiter': '/', 'includeTrailingDelimiter': 'true', 'maxResults': 1}
         async for page in await self._storage_client.list_objects(fsurl._bucket, params=params):
             prefixes = page.get('prefixes')