feat: add support for PITR-lite backups (#1)

* Backup changes

* Basic tests

* Add system tests

* Fix system tests

* Add retention period to backup systests

Author: zoercai

google/cloud/spanner_v1/backup.py

@@ -51,14 +51,23 @@ class Backup(object):
     :param expire_time: (Optional) The expire time that will be used to
                         create the backup. Required if the create method
                         needs to be called.
+
+    :type version_time: :class:`datetime.datetime`
+    :param version_time: (Optional) The version time that was specified for
+                        the externally consistent copy of the database. If
+                        not present, it is the same as the `create_time` of
+                        the backup.
     """
 
-    def __init__(self, backup_id, instance, database="", expire_time=None):
+    def __init__(
+        self, backup_id, instance, database="", expire_time=None, version_time=None
+    ):
         self.backup_id = backup_id
         self._instance = instance
         self._database = database
         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
@@ -109,6 +118,16 @@ def create_time(self):
         """
         return self._create_time
 
+    @property
+    def version_time(self):
+        """Version time of this backup.
+
+        :rtype: :class:`datetime.datetime`
+        :returns: a datetime object representing the version time of
+            this backup
+        """
+        return self._version_time
+
     @property
     def size_bytes(self):
         """Size of this backup in bytes.
@@ -190,7 +209,11 @@ def create(self):
             raise ValueError("database not set")
         api = self._instance._client.database_admin_api
         metadata = _metadata_with_prefix(self.name)
-        backup = BackupPB(database=self._database, expire_time=self.expire_time,)
+        backup = BackupPB(
+            database=self._database,
+            expire_time=self.expire_time,
+            version_time=self.version_time,
+        )
 
         future = api.create_backup(
             parent=self._instance.name,
@@ -228,6 +251,7 @@ def reload(self):
         self._database = pb.database
         self._expire_time = pb.expire_time
         self._create_time = pb.create_time
+        self._version_time = pb.version_time
         self._size_bytes = pb.size_bytes
         self._state = BackupPB.State(pb.state)
         self._referencing_databases = pb.referencing_databases

google/cloud/spanner_v1/instance.py

@@ -400,7 +400,7 @@ def list_databases(self, page_size=None):
         )
         return page_iter
 
-    def backup(self, backup_id, database="", expire_time=None):
+    def backup(self, backup_id, database="", expire_time=None, version_time=None):
         """Factory to create a backup within this instance.
 
         :type backup_id: str
@@ -415,13 +415,29 @@ def backup(self, backup_id, database="", expire_time=None):
         :param expire_time:
             Optional. The expire time that will be used when creating the backup.
             Required if the create method needs to be called.
+
+        :type version_time: :class:`datetime.datetime`
+        :param version_time:
+            Optional. The version time that will be used to create the externally
+            consistent copy of the database. If not present, it is the same as
+            the `create_time` of the backup.
         """
         try:
             return Backup(
-                backup_id, self, database=database.name, expire_time=expire_time
+                backup_id,
+                self,
+                database=database.name,
+                expire_time=expire_time,
+                version_time=version_time,
             )
         except AttributeError:
-            return Backup(backup_id, self, database=database, expire_time=expire_time)
+            return Backup(
+                backup_id,
+                self,
+                database=database,
+                expire_time=expire_time,
+                version_time=version_time,
+            )
 
     def list_backups(self, filter_="", page_size=None):
         """List backups for the instance.

tests/system/test_system.py

@@ -613,6 +613,22 @@ def setUpClass(cls):
         op1.result(SPANNER_OPERATION_TIMEOUT_IN_SECONDS)  # raises on failure / timeout.
         op2.result(SPANNER_OPERATION_TIMEOUT_IN_SECONDS)  # raises on failure / timeout.
 
+        # Add retention period for backups
+        retention_period = "7d"
+        ddl_statements = DDL_STATEMENTS + [
+            "ALTER DATABASE {}"
+            " SET OPTIONS (version_retention_period = '{}')".format(
+                cls.DATABASE_NAME, retention_period
+            )
+        ]
+        db = Config.INSTANCE.database(
+            cls.DATABASE_NAME, pool=pool, ddl_statements=ddl_statements
+        )
+        operation = db.update_ddl(ddl_statements)
+        # We want to make sure the operation completes.
+        operation.result(240)  # raises on failure / timeout.
+        db.reload()
+
         current_config = Config.INSTANCE.configuration_name
         same_config_instance_id = "same-config" + unique_resource_id("-")
         create_time = str(int(time.time()))
@@ -685,9 +701,16 @@ def test_backup_workflow(self):
         backup_id = "backup_id" + unique_resource_id("_")
         expire_time = datetime.utcnow() + timedelta(days=3)
         expire_time = expire_time.replace(tzinfo=UTC)
+        version_time = datetime.utcnow() - timedelta(seconds=5)
+        version_time = version_time.replace(tzinfo=UTC)
 
         # Create backup.
-        backup = instance.backup(backup_id, database=self._db, expire_time=expire_time)
+        backup = instance.backup(
+            backup_id,
+            database=self._db,
+            expire_time=expire_time,
+            version_time=version_time,
+        )
         operation = backup.create()
         self.to_delete.append(backup)
 
@@ -702,6 +725,7 @@ def test_backup_workflow(self):
         self.assertEqual(self._db.name, backup._database)
         self.assertEqual(expire_time, backup.expire_time)
         self.assertIsNotNone(backup.create_time)
+        self.assertEqual(version_time, backup.version_time)
         self.assertIsNotNone(backup.size_bytes)
         self.assertIsNotNone(backup.state)
 
@@ -722,6 +746,80 @@ def test_backup_workflow(self):
         backup.delete()
         self.assertFalse(backup.exists())
 
+    def test_backup_version_time_defaults_to_create_time(self):
+        from datetime import datetime
+        from datetime import timedelta
+        from pytz import UTC
+
+        instance = Config.INSTANCE
+        backup_id = "backup_id" + unique_resource_id("_")
+        expire_time = datetime.utcnow() + timedelta(days=3)
+        expire_time = expire_time.replace(tzinfo=UTC)
+
+        # Create backup.
+        backup = instance.backup(backup_id, database=self._db, expire_time=expire_time,)
+        operation = backup.create()
+        self.to_delete.append(backup)
+
+        # Check metadata.
+        metadata = operation.metadata
+        self.assertEqual(backup.name, metadata.name)
+        self.assertEqual(self._db.name, metadata.database)
+        operation.result()
+
+        # Check backup object.
+        backup.reload()
+        self.assertEqual(self._db.name, backup._database)
+        self.assertIsNotNone(backup.create_time)
+        self.assertEqual(backup.create_time, backup.version_time)
+
+        backup.delete()
+        self.assertFalse(backup.exists())
+
+    def test_create_backup_invalid_version_time_past(self):
+        from datetime import datetime
+        from datetime import timedelta
+        from pytz import UTC
+
+        backup_id = "backup_id" + unique_resource_id("_")
+        expire_time = datetime.utcnow() + timedelta(days=3)
+        expire_time = expire_time.replace(tzinfo=UTC)
+        version_time = datetime.utcnow() - timedelta(days=10)
+        version_time = version_time.replace(tzinfo=UTC)
+
+        backup = Config.INSTANCE.backup(
+            backup_id,
+            database=self._db,
+            expire_time=expire_time,
+            version_time=version_time,
+        )
+
+        with self.assertRaises(exceptions.InvalidArgument):
+            op = backup.create()
+            op.result()
+
+    def test_create_backup_invalid_version_time_future(self):
+        from datetime import datetime
+        from datetime import timedelta
+        from pytz import UTC
+
+        backup_id = "backup_id" + unique_resource_id("_")
+        expire_time = datetime.utcnow() + timedelta(days=3)
+        expire_time = expire_time.replace(tzinfo=UTC)
+        version_time = datetime.utcnow() + timedelta(days=2)
+        version_time = version_time.replace(tzinfo=UTC)
+
+        backup = Config.INSTANCE.backup(
+            backup_id,
+            database=self._db,
+            expire_time=expire_time,
+            version_time=version_time,
+        )
+
+        with self.assertRaises(exceptions.InvalidArgument):
+            op = backup.create()
+            op.result()
+
     def test_restore_to_diff_instance(self):
         from datetime import datetime
         from datetime import timedelta
@@ -818,9 +916,14 @@ def test_list_backups(self):
         instance = Config.INSTANCE
         expire_time_1 = datetime.utcnow() + timedelta(days=21)
         expire_time_1 = expire_time_1.replace(tzinfo=UTC)
+        version_time_1 = datetime.utcnow() - timedelta(minutes=5)
+        version_time_1 = version_time_1.replace(tzinfo=UTC)
 
         backup1 = Config.INSTANCE.backup(
-            backup_id_1, database=self._dbs[0], expire_time=expire_time_1
+            backup_id_1,
+            database=self._dbs[0],
+            expire_time=expire_time_1,
+            version_time=version_time_1,
         )
 
         expire_time_2 = datetime.utcnow() + timedelta(days=1)
@@ -860,6 +963,13 @@ def test_list_backups(self):
         for backup in instance.list_backups(filter_=filter_):
             self.assertEqual(backup.name, backup2.name)
 
+        # List backups filtered by version time.
+        filter_ = 'version_time > "{0}"'.format(
+            create_time_compare.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+        )
+        for backup in instance.list_backups(filter_=filter_):
+            self.assertEqual(backup.name, backup2.name)
+
         # List backups filtered by expire time.
         filter_ = 'expire_time > "{0}"'.format(
             expire_time_1.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

tests/unit/test_backup.py

@@ -266,19 +266,32 @@ def test_create_database_not_set(self):
 
     def test_create_success(self):
         from google.cloud.spanner_admin_database_v1 import Backup
+        from datetime import datetime
+        from datetime import timedelta
+        from pytz import UTC
 
         op_future = object()
         client = _Client()
         api = client.database_admin_api = self._make_database_admin_api()
         api.create_backup.return_value = op_future
 
         instance = _Instance(self.INSTANCE_NAME, client=client)
-        timestamp = self._make_timestamp()
+        version_timestamp = datetime.utcnow() - timedelta(minutes=5)
+        version_timestamp = version_timestamp.replace(tzinfo=UTC)
+        expire_timestamp = self._make_timestamp()
         backup = self._make_one(
-            self.BACKUP_ID, instance, database=self.DATABASE_NAME, expire_time=timestamp
+            self.BACKUP_ID,
+            instance,
+            database=self.DATABASE_NAME,
+            expire_time=expire_timestamp,
+            version_time=version_timestamp,
         )
 
-        backup_pb = Backup(database=self.DATABASE_NAME, expire_time=timestamp,)
+        backup_pb = Backup(
+            database=self.DATABASE_NAME,
+            expire_time=expire_timestamp,
+            version_time=version_timestamp,
+        )
 
         future = backup.create()
         self.assertIs(future, op_future)
@@ -437,6 +450,7 @@ def test_reload_success(self):
             name=self.BACKUP_NAME,
             database=self.DATABASE_NAME,
             expire_time=timestamp,
+            version_time=timestamp,
             create_time=timestamp,
             size_bytes=10,
             state=1,
@@ -452,6 +466,7 @@ def test_reload_success(self):
         self.assertEqual(backup.database, self.DATABASE_NAME)
         self.assertEqual(backup.expire_time, timestamp)
         self.assertEqual(backup.create_time, timestamp)
+        self.assertEqual(backup.version_time, timestamp)
         self.assertEqual(backup.size_bytes, 10)
         self.assertEqual(backup.state, Backup.State.CREATING)
         self.assertEqual(backup.referencing_databases, [])