refactor: split GDriveConnection into a Drive client, layout, and one class per job

benchmarks/download_files_parallel.py

@@ -44,8 +44,8 @@ def benchmark_gdrive_load_state():
 
     connection = manager_do1._connection_router.connections[0]
 
-    personal_syftbox_folder_id = connection.get_personal_syftbox_folder_id()
-    file_metadatas = connection.get_file_metadatas_from_folder(
+    personal_syftbox_folder_id = connection.layout.get_personal_syftbox_folder_id()
+    file_metadatas = connection.layout.get_file_metadatas_from_folder(
         personal_syftbox_folder_id
     )
     # valid_fname_objs = connection._get_valid_events_from_file_metadatas(file_metadatas)

packages/syft-enclave/src/syft_enclaves/utils.py

@@ -3,7 +3,7 @@
     MockDriveBackingStore,
     MockDriveService,
 )
-from syft.sync.connections.drive.gdrive_transport import GDriveConnection
+from syft.sync.connections.drive.drive_cached_client import DriveCachedClient
 from syft.sync.syftbox_manager import SyftboxManager
 from syft.sync.transport.protocols import Transport
 
@@ -66,7 +66,7 @@ def create_clients(configs) -> tuple:
 def _mock_transport(backing_store: MockDriveBackingStore, email: str) -> Transport:
     service = MockDriveService(backing_store, email)
     return SyftboxManager._drive_transport(
-        GDriveConnection.from_service(email, service)
+        DriveCachedClient.from_service(email, service)
     )
 
 

packages/syft-rds/src/syft_rds/client.py

@@ -394,9 +394,8 @@ def _share_any_datasets_with_peer(self, peer_email: str) -> None:
                     )
             except Exception:
                 # One dataset failing (missing folder, quota, network) must not
-                # stop us sharing the rest with this peer. "alreadyShared" is
-                # already handled in _batch_add_permissions, so anything
-                # reaching here is a real failure worth a traceback.
+                # stop us sharing the rest with this peer. Anything reaching
+                # here is a real failure worth a traceback.
                 logger.exception("Failed to share dataset %r with %s", tag, peer_email)
 
     # ------------------------------------------------------------------ #
@@ -818,9 +817,7 @@ def _create_and_upload_collection(
 
         Returns ``(folder_id, content_hash)``.
         """
-        from syft.sync.connections.drive.gdrive_transport import (
-            CollectionFolder,
-        )
+        from syft.sync.connections.drive.drive_collections import CollectionFolder
 
         content_hash = CollectionFolder.compute_hash(files)
         folder_id = self.sync_engine.create_collection_folder(

packages/syft-rds/tests/test_dataset_upload_private.py

@@ -1,14 +1,11 @@
-from syft.sync.connections.drive.gdrive_transport import (
-    GDriveConnection,
-    collection_name_query,
-)
+from syft.sync.connections.drive.drive_collections import collection_name_query
 from syft_rds import SyftRDSClient
 from syft_datasets.dataset_manager import (
     DATASET_COLLECTION_PREFIX,
     PRIVATE_DATASET_COLLECTION_PREFIX,
 )
 from dataset_test_utils import create_tmp_dataset_files
-from syft.sync.connections.drive.drive_connections import gdrive_connection_of
+from syft.sync.connections.drive.drive_connections import drive_cached_client_of
 
 
 class TestDatasetUploadPrivate:
@@ -146,11 +143,11 @@ def test_ds_cannot_find_private_folders_via_gdrive_query(self):
         )
 
         # Get the DS's raw GDrive connection and search for private folders
-        ds_connection: GDriveConnection = gdrive_connection_of(
+        ds_cached_client = drive_cached_client_of(
             ds_manager.sync_engine._connection_router.transport
         )
         results = (
-            ds_connection.drive_service.files()
+            ds_cached_client.client.drive_service.files()
             .list(
                 q=f"{collection_name_query(PRIVATE_DATASET_COLLECTION_PREFIX)} and trashed=false",
                 fields="files(id,name)",
@@ -160,11 +157,11 @@ def test_ds_cannot_find_private_folders_via_gdrive_query(self):
         assert len(results.get("files", [])) == 0
 
         # DO should find it with the same query
-        do_connection: GDriveConnection = gdrive_connection_of(
+        do_cached_client = drive_cached_client_of(
             do_manager.sync_engine._connection_router.transport
         )
         do_results = (
-            do_connection.drive_service.files()
+            do_cached_client.client.drive_service.files()
             .list(
                 q=f"{collection_name_query(PRIVATE_DATASET_COLLECTION_PREFIX)} and trashed=false",
                 fields="files(id,name)",

packages/syft-rds/tests/test_sync_manager.py

@@ -8,7 +8,7 @@
 import pytest
 
 from syft.sync.connections.drive import mock_drive_service
-from syft.sync.connections.drive.gdrive_transport import GDriveConnection
+from syft.sync.connections.drive.drive_cached_client import DriveCachedClient
 from syft.sync.syftbox_manager import SyftboxManager
 from syft_datasets import Dataset
 from syft_datasets.dataset_manager import DATASET_COLLECTION_PREFIX
@@ -22,7 +22,7 @@
     create_tmp_dataset_files,
     create_tmp_dataset_files_with_parquet,
 )
-from syft.sync.connections.drive.drive_connections import gdrive_connection_of
+from syft.sync.connections.drive.drive_connections import drive_cached_client_of
 
 
 def path_for_job(do_email: str, ds_email: str, filename: str = "test.job") -> str:
@@ -233,9 +233,9 @@ def test_dataset_only_mock_data_uploaded():
     # Sync so DS receives the dataset
     ds_manager.sync()
 
-    files = gdrive_connection_of(
+    files = drive_cached_client_of(
         ds_manager.sync_engine._connection_router.transport
-    ).drive_service._backing_store.files
+    ).client.drive_service._backing_store.files
     list(files.keys())
     file_objs = list(files.values())
     file_objs_ex_dataset_yaml = [
@@ -284,11 +284,13 @@ def test_jobs():
     # we do this by not always syncing on a file change, currently this logic is a bit of
     # a short cut, but we could do this based on timing eventually (if there are items in the
     # queue for longer than a certain time we start pushing)
-    connection_do = gdrive_connection_of(
+    do_cached_client = drive_cached_client_of(
         do_manager.sync_engine._connection_router.transport
     )
-    inbox_folder_id = connection_do._get_own_datasite_inbox_id(ds_manager.email)
-    inbox_file_metadatas = connection_do.get_file_metadatas_from_folder(inbox_folder_id)
+    inbox_folder_id = do_cached_client.get_own_datasite_inbox_id(ds_manager.email)
+    inbox_file_metadatas = do_cached_client.get_file_metadatas_from_folder(
+        inbox_folder_id
+    )
     assert len(inbox_file_metadatas) == 1
 
     do_manager.sync()
@@ -1040,9 +1042,9 @@ def test_ds_dataset_cache_aware_sync():
     assert remote_hash is not None
 
     # Get the mock backing store and directories for creating second pair
-    mock_backing_store = gdrive_connection_of(
+    mock_backing_store = drive_cached_client_of(
         ds_manager.sync_engine._connection_router.transport
-    ).drive_service._backing_store
+    ).client.drive_service._backing_store
     ds_folder = ds_manager.syftbox_folder
     do_folder = do_manager.syftbox_folder
     ds_email = ds_manager.email
@@ -1059,12 +1061,12 @@ def test_ds_dataset_cache_aware_sync():
     )
 
     # Replace mock backing store to share dataset collections
-    gdrive_connection_of(
+    drive_cached_client_of(
         ds_manager2.sync_engine._connection_router.transport
-    ).drive_service._backing_store = mock_backing_store
-    gdrive_connection_of(
+    ).client.drive_service._backing_store = mock_backing_store
+    drive_cached_client_of(
         do_manager2.sync_engine._connection_router.transport
-    ).drive_service._backing_store = mock_backing_store
+    ).client.drive_service._backing_store = mock_backing_store
 
     # Load peers (already approved in shared backing store)
     ds_manager2.load_peers()
@@ -1140,9 +1142,9 @@ def test_do_dataset_cache_aware_sync():
     assert len(do_manager.datasets.get_all()) == 1
 
     # Get the mock backing store and directories for creating second pair
-    mock_backing_store = gdrive_connection_of(
+    mock_backing_store = drive_cached_client_of(
         ds_manager.sync_engine._connection_router.transport
-    ).drive_service._backing_store
+    ).client.drive_service._backing_store
     ds_folder = ds_manager.syftbox_folder
     do_folder = do_manager.syftbox_folder
     ds_email = ds_manager.email
@@ -1159,12 +1161,12 @@ def test_do_dataset_cache_aware_sync():
     )
 
     # Replace mock backing store to share dataset collections
-    gdrive_connection_of(
+    drive_cached_client_of(
         ds_manager2.sync_engine._connection_router.transport
-    ).drive_service._backing_store = mock_backing_store
-    gdrive_connection_of(
+    ).client.drive_service._backing_store = mock_backing_store
+    drive_cached_client_of(
         do_manager2.sync_engine._connection_router.transport
-    ).drive_service._backing_store = mock_backing_store
+    ).client.drive_service._backing_store = mock_backing_store
 
     # Load peers (already approved in shared backing store)
     ds_manager2.load_peers()
@@ -1207,9 +1209,9 @@ def test_in_memory_connection_load_state():
     )
 
     # Get the backing store that will persist across "restarts"
-    backing_store = gdrive_connection_of(
+    backing_store = drive_cached_client_of(
         do_manager1.sync_engine._connection_router.transport
-    ).drive_service._backing_store
+    ).client.drive_service._backing_store
     ds_folder = ds_manager1.syftbox_folder
     do_folder = do_manager1.syftbox_folder
     ds_email = ds_manager1.email
@@ -1269,16 +1271,16 @@ def test_in_memory_connection_load_state():
 
     # Connect to the same backing store
     service_do = mock_drive_service.MockDriveService(backing_store, do_email)
-    do_connection2 = GDriveConnection.from_service(do_email, service_do)
+    do_cached_client2 = DriveCachedClient.from_service(do_email, service_do)
 
     service_ds = mock_drive_service.MockDriveService(backing_store, ds_email)
-    ds_connection2 = GDriveConnection.from_service(ds_email, service_ds)
+    ds_cached_client2 = DriveCachedClient.from_service(ds_email, service_ds)
 
     do_manager2 = SyftRDSClient.from_config(
-        do_config2, SyftboxManager._drive_transport(do_connection2)
+        do_config2, SyftboxManager._drive_transport(do_cached_client2)
     )
     ds_manager2 = SyftRDSClient.from_config(
-        ds_config2, SyftboxManager._drive_transport(ds_connection2)
+        ds_config2, SyftboxManager._drive_transport(ds_cached_client2)
     )
 
     # Load peers