from uuid import UUIDfrom encord import EncordUserClientfrom encord.orm.storage import CloudSyncedFolderParams# User inputSSH_PATH = "/Users/chris-encord/ssh-private-key.txt" # Specify the file path to your access keyCLOUD_SYNCED_FOLDER_NAME = "SDK Cloud-Synced Folder" # Specify a meaningful name for your Cloud-synced FolderCLOUD_SYNCED_FOLDER_DESCRIPTION = "A folder to store my files" # Specify a meaningful description for your Cloud-synced FolderINTEGRATION_UUID = "3b6299c3-f8c8-4755-ae26-d9144b215920" # Specify the unique id for your integrationREMOTE_URL = "gs://my-gcp-bucket/" # Specify the storage/file path to your cloud storage# Authenticate with Encord using the path to your private keyuser_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key( ssh_private_key_path=SSH_PATH, domain="https://api.encord.com",)# Create cloud synced folder paramscloud_synced_folder_params = CloudSyncedFolderParams( integration_uuid=UUID(INTEGRATION_UUID), remote_url=REMOTE_URL,)# Create the storage folderfolder_name = CLOUD_SYNCED_FOLDER_NAMEfolder_description = CLOUD_SYNCED_FOLDER_DESCRIPTIONfolder_metadata = {"my": "folder_metadata"}storage_folder = user_client.create_storage_folder( name=folder_name, description=folder_description, client_metadata=folder_metadata, cloud_synced_folder_params=cloud_synced_folder_params,)
The following code syncs a Cloud-sync Folder with the cloud storage bucket.The sync_private_data_with_cloud_synced_folder_get_result time out value can be adjusted to your needs.
Sync Cloud-synced Folder
Copy
from uuid import UUIDfrom encord import EncordUserClientfrom encord.orm.storage import SyncPrivateDataWithCloudSyncedFolderStatusfrom encord.storage import FoldersSortBy# User inputSSH_PATH = "/Users/chris-encord/ssh-private-key.txt" # Specify the file path to your access keyCLOUD_SYNCED_FOLDER_NAME = "SDK Cloud-Synced Folder" # Specify the name of your Cloud-synced Folder# Authenticate with Encorduser_client: EncordUserClient = EncordUserClient.create_with_ssh_private_key( ssh_private_key_path=SSH_PATH, domain="https://api.encord.com",)# Option 1: Use UUID directly# storage_folder = user_client.get_storage_folder(CLOUD_SYNCED_FOLDER_ID)# Option 2: Look up folder by namefolders = list(user_client.find_storage_folders(search=CLOUD_SYNCED_FOLDER_NAME))if not folders: print("Folder not found") exit()storage_folder = folders[0]# Start sync jobsync_job_uuid = storage_folder.sync_private_data_with_cloud_synced_folder_start()# Wait for resultresult = storage_folder.sync_private_data_with_cloud_synced_folder_get_result( sync_job_uuid, timeout_seconds=300 # You can adjust this)print(f"Sync job finished with status: {result.status}")if result.status == SyncPrivateDataWithCloudSyncedFolderStatus.DONE: print("Sync completed successfully.") if result.unit_errors: print("Some items failed to sync:") for err in result.unit_errors: print(err.object_urls)elif result.status == SyncPrivateDataWithCloudSyncedFolderStatus.PENDING: print("Sync is still in progress. Try polling later.")else: print(f"Sync failed or cancelled. Errors: {result.errors}")