feat: backup restore to different instance

google/cloud/bigtable/backup.py

@@ -376,17 +376,25 @@ def delete(self):
             request={"name": self.name}
         )
 
-    def restore(self, table_id):
+    def restore(self, table_id, instance_id=None):
         """Creates a new Table by restoring from this Backup. The new Table
-        must be in the same Instance as the Instance containing the Backup.
+        can be created in the same Instance as the Instance containing the
+        Backup, or another Instance whose ID can be specified in the arguments.
         The returned Table ``long-running operation`` can be used to track the
         progress of the operation and to cancel it. The ``response`` type is
         ``Table``, if successful.
 
+        :type table_id: str
         :param table_id: The ID of the Table to create and restore to.
                          This Table must not already exist.
-        :returns: An instance of
-             :class:`~google.cloud.bigtable_admin_v2.types._OperationFuture`.
+
+        :type instance_id: str
+        :param instance_id: (Optional) The ID of the Instance to restore the
+                            backup into, if different from the current one.
+
+        :rtype: :class:`~google.cloud.bigtable_admin_v2.types._OperationFuture`
+        :returns: A future to be used to poll the status of the 'restore'
+                  request.
 
         :raises: google.api_core.exceptions.AlreadyExists: If the table
                  already exists.
@@ -397,12 +405,15 @@ def restore(self, table_id):
         :raises: ValueError: If the parameters are invalid.
         """
         api = self._instance._client._table_admin_client
+        if instance_id:
+            parent = BigtableTableAdminClient.instance_path(
+                project=self._instance._client.project, instance=instance_id,
+            )
+        else:
+            parent = self._instance.name
+
         return api.restore_table(
-            request={
-                "parent": self._instance.name,
-                "table_id": table_id,
-                "backup": self.name,
-            }
+            request={"parent": parent, "table_id": table_id, "backup": self.name}
         )
 
     def get_iam_policy(self):

tests/system.py

@@ -897,12 +897,33 @@ def test_backup(self):
         restored_table_id = "test-backup-table-restored"
         restored_table = Config.INSTANCE_DATA.table(restored_table_id)
         temp_table.restore(
-            restored_table_id, cluster_id=CLUSTER_ID_DATA, backup_id=temp_backup_id
+            restored_table_id, cluster_id=CLUSTER_ID_DATA, backup_id=temp_backup_id,
         ).result()
         tables = Config.INSTANCE_DATA.list_tables()
         self.assertIn(restored_table, tables)
         restored_table.delete()
 
+        # Testing `Backup.restore()` into a different instance:
+        # Setting up another instance...
+        alt_instance_id = "gcp-" + UNIQUE_SUFFIX
+        alt_cluster_id = alt_instance_id + "-cluster"
+        alt_instance = Config.CLIENT.instance(alt_instance_id, labels=LABELS)
+        alt_cluster = alt_instance.cluster(
+            cluster_id=alt_cluster_id, location_id=LOCATION_ID, serve_nodes=SERVE_NODES,
+        )
+        if not Config.IN_EMULATOR:
+            alt_instance.create(clusters=[alt_cluster]).result(timeout=10)
+
+        # Testing `restore()`...
+        temp_backup.restore(restored_table_id, alt_instance_id).result()
+        restored_table = alt_instance.table(restored_table_id)
+        self.assertIn(restored_table, alt_instance.list_tables())
+        restored_table.delete()
+
+        # Tearing down the resources...
+        if not Config.IN_EMULATOR:
+            retry_429(alt_instance.delete)()
+
 
 class TestDataAPI(unittest.TestCase):
     @classmethod

tests/unit/test_backup.py

@@ -32,6 +32,11 @@ class TestBackup(unittest.TestCase):
     BACKUP_ID = "backup-id"
     BACKUP_NAME = CLUSTER_NAME + "/backups/" + BACKUP_ID
 
+    ALT_INSTANCE = "other-instance-id"
+    ALT_INSTANCE_NAME = "projects/" + PROJECT_ID + "/instances/" + ALT_INSTANCE
+    ALT_CLUSTER_NAME = ALT_INSTANCE_NAME + "/clusters/" + CLUSTER_ID
+    ALT_BACKUP_NAME = ALT_CLUSTER_NAME + "/backups/" + BACKUP_ID
+
     @staticmethod
     def _get_target_class():
         from google.cloud.bigtable.backup import Backup
@@ -714,7 +719,7 @@ def test_restore_cluster_not_set(self):
         with self.assertRaises(ValueError):
             backup.restore(self.TABLE_ID)
 
-    def test_restore_success(self):
+    def _restore_helper(self, instance_id=None, instance_name=None):
         op_future = object()
         client = _Client()
         api = client._table_admin_client = self._make_table_admin_client()
@@ -729,17 +734,24 @@ def test_restore_success(self):
             expire_time=timestamp,
         )
 
-        future = backup.restore(self.TABLE_ID)
+        future = backup.restore(self.TABLE_ID, instance_id)
         self.assertEqual(backup._cluster, self.CLUSTER_ID)
         self.assertIs(future, op_future)
 
         api.restore_table.assert_called_once_with(
             request={
-                "parent": self.INSTANCE_NAME,
+                "parent": instance_name or self.INSTANCE_NAME,
                 "table_id": self.TABLE_ID,
                 "backup": self.BACKUP_NAME,
             }
         )
+        api.restore_table.reset_mock()
+
+    def test_restore_default(self):
+        self._restore_helper()
+
+    def test_restore_to_another_instance(self):
+        self._restore_helper(self.ALT_INSTANCE, self.ALT_INSTANCE_NAME)
 
     def test_get_iam_policy(self):
         from google.cloud.bigtable.client import Client