feat: add support for spanner copy backup feature

google/cloud/spanner_v1/backup.py

@@ -21,6 +21,8 @@
 from google.cloud.spanner_admin_database_v1 import Backup as BackupPB
 from google.cloud.spanner_admin_database_v1 import CreateBackupEncryptionConfig
 from google.cloud.spanner_admin_database_v1 import CreateBackupRequest
+from google.cloud.spanner_admin_database_v1 import CopyBackupEncryptionConfig
+from google.cloud.spanner_admin_database_v1 import CopyBackupRequest
 from google.cloud.spanner_v1._helpers import _metadata_with_prefix
 
 _BACKUP_NAME_RE = re.compile(
@@ -77,19 +79,30 @@ def __init__(
         expire_time=None,
         version_time=None,
         encryption_config=None,
+        source_backup=None,
     ):
         self.backup_id = backup_id
         self._instance = instance
         self._database = database
+        self._source_backup = source_backup
         self._expire_time = expire_time
         self._create_time = None
         self._version_time = version_time
         self._size_bytes = None
         self._state = None
         self._referencing_databases = None
         self._encryption_info = None
+        self._max_expire_time = None
+        self._referencing_backups = None
         if type(encryption_config) == dict:
-            self._encryption_config = CreateBackupEncryptionConfig(**encryption_config)
+            if source_backup:
+                self._encryption_config = CopyBackupEncryptionConfig(
+                    **encryption_config
+                )
+            else:
+                self._encryption_config = CreateBackupEncryptionConfig(
+                    **encryption_config
+                )
         else:
             self._encryption_config = encryption_config
 
@@ -185,6 +198,26 @@ def encryption_info(self):
         """
         return self._encryption_info
 
+    @property
+    def max_expire_time(self):
+        """The max allowed expiration time of the backup.
+
+        :rtype: :class:`datetime.datetime`
+        :returns: a datetime object representing the max expire time of
+            this backup
+        """
+        return self._max_expire_time
+
+    @property
+    def referencing_backups(self):
+        """The names of the destination backups being created by copying this source backup.
+
+        :rtype: list of strings
+        :returns: a list of backup path strings which specify the backups that are
+            referencing this copy backup
+        """
+        return self._referencing_backups
+
     @classmethod
     def from_pb(cls, backup_pb, instance):
         """Create an instance of this class from a protobuf message.
@@ -223,7 +256,7 @@ def from_pb(cls, backup_pb, instance):
         return cls(backup_id, instance)
 
     def create(self):
-        """Create this backup within its instance.
+        """Create this backup or backup copy within its instance.
 
         :rtype: :class:`~google.api_core.operation.Operation`
         :returns: a future used to poll the status of the create request
@@ -234,6 +267,32 @@ def create(self):
         """
         if not self._expire_time:
             raise ValueError("expire_time not set")
+
+        api = self._instance._client.database_admin_api
+        metadata = _metadata_with_prefix(self.name)
+
+        if self._source_backup:
+            if (
+                self._encryption_config
+                and self._encryption_config.kms_key_name
+                and self._encryption_config.encryption_type
+                != CopyBackupEncryptionConfig.EncryptionType.CUSTOMER_MANAGED_ENCRYPTION
+            ):
+                raise ValueError(
+                    "kms_key_name only used with CUSTOMER_MANAGED_ENCRYPTION"
+                )
+
+            request = CopyBackupRequest(
+                parent=self._instance.name,
+                backup_id=self.backup_id,
+                source_backup=self._source_backup,
+                expire_time=self._expire_time,
+                encryption_config=self._encryption_config,
+            )
+
+            future = api.copy_backup(request=request, metadata=metadata,)
+            return future
+
         if not self._database:
             raise ValueError("database not set")
         if (
@@ -243,8 +302,7 @@ def create(self):
             != CreateBackupEncryptionConfig.EncryptionType.CUSTOMER_MANAGED_ENCRYPTION
         ):
             raise ValueError("kms_key_name only used with CUSTOMER_MANAGED_ENCRYPTION")
-        api = self._instance._client.database_admin_api
-        metadata = _metadata_with_prefix(self.name)
+
         backup = BackupPB(
             database=self._database,
             expire_time=self.expire_time,
@@ -294,6 +352,8 @@ def reload(self):
         self._state = BackupPB.State(pb.state)
         self._referencing_databases = pb.referencing_databases
         self._encryption_info = pb.encryption_info
+        self._max_expire_time = pb.max_expire_time
+        self._referencing_backups = pb.referencing_backups
 
     def update_expire_time(self, new_expire_time):
         """Update the expire time of this backup.

google/cloud/spanner_v1/instance.py

@@ -551,6 +551,41 @@ def backup(
                 encryption_config=encryption_config,
             )
 
+    def copy_backup(
+        self, backup_id, source_backup, expire_time=None, encryption_config=None,
+    ):
+        """Factory to create a copy backup within this instance.
+
+            :type backup_id: str
+            :param backup_id: The ID of the backup copy.
+
+            :type source_backup_id: str
+            :param backup_id: The ID of the source backup to be copied.
+
+            :type expire_time: :class:`datetime.datetime`
+            :param expire_time:
+                Optional. The expire time that will be used when creating the copy backup.
+                Required if the create method needs to be called.
+
+            :type encryption_config:
+                :class:`~google.cloud.spanner_admin_database_v1.types.CopyBackupEncryptionConfig`
+                or :class:`dict`
+            :param encryption_config:
+                (Optional) Encryption configuration for the backup.
+                If a dict is provided, it must be of the same form as the protobuf
+                message :class:`~google.cloud.spanner_admin_database_v1.types.CopyBackupEncryptionConfig`
+
+            :rtype: :class:`~google.cloud.spanner_v1.backup.Backup`
+            :returns: a copy backup owned by this instance.
+            """
+        return Backup(
+            backup_id,
+            self,
+            source_backup=source_backup,
+            expire_time=expire_time,
+            encryption_config=encryption_config,
+        )
+
     def list_backups(self, filter_="", page_size=None):
         """List backups for the instance.
 

tests/system/_helpers.py

@@ -39,6 +39,9 @@
 DATABASE_OPERATION_TIMEOUT_IN_SECONDS = int(
     os.getenv("SPANNER_DATABASE_OPERATION_TIMEOUT_IN_SECONDS", 60)
 )
+BACKUP_OPERATION_TIMEOUT_IN_SECONDS = int(
+    os.getenv("SPANNER_BACKUP_OPERATION_TIMEOUT_IN_SECONDS", 1200)
+)
 
 USE_EMULATOR_ENVVAR = "SPANNER_EMULATOR_HOST"
 USE_EMULATOR = os.getenv(USE_EMULATOR_ENVVAR) is not None

tests/system/conftest.py

@@ -12,7 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import datetime
 import time
+from google.cloud.spanner_admin_database_v1.types.backup import (
+    CreateBackupEncryptionConfig,
+)
 
 import pytest
 
@@ -67,6 +71,11 @@ def database_operation_timeout():
     return _helpers.DATABASE_OPERATION_TIMEOUT_IN_SECONDS
 
 
+@pytest.fixture(scope="session")
+def backup_operation_timeout():
+    return _helpers.BACKUP_OPERATION_TIMEOUT_IN_SECONDS
+
+
 @pytest.fixture(scope="session")
 def shared_instance_id():
     if _helpers.CREATE_INSTANCE:
@@ -148,6 +157,30 @@ def shared_database(shared_instance, database_operation_timeout):
     database.drop()
 
 
+@pytest.fixture(scope="session")
+def shared_backup(shared_instance, shared_database, backup_operation_timeout):
+    backup_name = _helpers.unique_id("test_backup")
+    expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
+        days=3
+    )
+    source_encryption_enum = CreateBackupEncryptionConfig.EncryptionType
+    source_encryption_config = CreateBackupEncryptionConfig(
+        encryption_type=source_encryption_enum.GOOGLE_DEFAULT_ENCRYPTION,
+    )
+    backup = shared_instance.backup(
+        backup_name,
+        database=shared_database,
+        expire_time=expire_time,
+        encryption_config=source_encryption_config,
+    )
+    operation = backup.create()
+    operation.result(backup_operation_timeout)  # raises on failure / timeout.
+
+    yield backup
+
+    backup.delete()
+
+
 @pytest.fixture(scope="function")
 def databases_to_delete():
     to_delete = []

tests/system/test_backup_api.py

@@ -199,6 +199,63 @@ def test_backup_workflow(
     assert not backup.exists()
 
 
+def test_copy_backup_workflow(
+    shared_instance, shared_backup, backups_to_delete,
+):
+    from google.cloud.spanner_admin_database_v1 import (
+        CreateBackupEncryptionConfig,
+        CopyBackupEncryptionConfig,
+        EncryptionInfo,
+    )
+
+    backup_id = _helpers.unique_id("backup_id", separator="_")
+    expire_time = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
+        days=3
+    )
+    copy_encryption_enum = CopyBackupEncryptionConfig.EncryptionType
+    copy_encryption_config = CopyBackupEncryptionConfig(
+        encryption_type=copy_encryption_enum.GOOGLE_DEFAULT_ENCRYPTION,
+    )
+
+    # Create backup.
+    shared_backup.reload()
+    # Create a copy backup
+    copy_backup = shared_instance.copy_backup(
+        backup_id=backup_id,
+        source_backup=shared_backup.name,
+        expire_time=expire_time,
+        encryption_config=copy_encryption_config,
+    )
+    operation = copy_backup.create()
+    backups_to_delete.append(copy_backup)
+
+    # Check metadata.
+    metadata = operation.metadata
+    assert copy_backup.name == metadata.name
+    operation.result()  # blocks indefinitely
+
+    # Check backup object.
+    copy_backup.reload()
+    assert expire_time == copy_backup.expire_time
+    assert copy_backup.create_time is not None
+    assert copy_backup.size_bytes is not None
+    assert copy_backup.state is not None
+    assert (
+        EncryptionInfo.Type.GOOGLE_DEFAULT_ENCRYPTION
+        == copy_backup.encryption_info.encryption_type
+    )
+
+    # Update with valid argument.
+    valid_expire_time = datetime.datetime.now(
+        datetime.timezone.utc
+    ) + datetime.timedelta(days=7)
+    copy_backup.update_expire_time(valid_expire_time)
+    assert valid_expire_time == copy_backup.expire_time
+
+    copy_backup.delete()
+    assert not copy_backup.exists()
+
+
 def test_backup_create_w_version_time_dflt_to_create_time(
     shared_instance, shared_database, backups_to_delete, databases_to_delete,
 ):
@@ -289,6 +346,25 @@ def test_backup_create_w_invalid_version_time_future(
         op.result()  # blocks indefinitely
 
 
+def test_copy_backup_create_w_invalid_expire_time(
+    shared_instance, shared_backup,
+):
+    backup_id = _helpers.unique_id("backup_id", separator="_")
+    invalid_expire_time = datetime.datetime.now(datetime.timezone.utc)
+
+    shared_backup.reload()
+
+    copy_backup = shared_instance.copy_backup(
+        backup_id=backup_id,
+        source_backup=shared_backup.name,
+        expire_time=invalid_expire_time,
+    )
+
+    with pytest.raises(exceptions.InvalidArgument):
+        operation = copy_backup.create()
+        operation.result()  # blocks indefinitely
+
+
 def test_database_restore_to_diff_instance(
     shared_instance,
     shared_database,