feat: add PITR-lite support

google/cloud/spanner_v1/database.py

@@ -103,6 +103,8 @@ def __init__(self, database_id, instance, ddl_statements=(), pool=None):
         self._state = None
         self._create_time = None
         self._restore_info = None
+        self._version_retention_period = None
+        self._earliest_version_time = None
 
         if pool is None:
             pool = BurstyPool()
@@ -200,6 +202,25 @@ def restore_info(self):
         """
         return self._restore_info
 
+    @property
+    def version_retention_period(self):
+        """The period in which Cloud Spanner retains all versions of data
+        for the database.
+
+        :rtype: str
+        :returns: a string representing the duration of the version retention period
+        """
+        return self._version_retention_period
+
+    @property
+    def earliest_version_time(self):
+        """The earliest time at which older versions of the data can be read.
+
+        :rtype: :class:`datetime.datetime`
+        :returns: a datetime object representing the earliest version time
+        """
+        return self._earliest_version_time
+
     @property
     def ddl_statements(self):
         """DDL Statements used to define database schema.
@@ -309,6 +330,10 @@ def reload(self):
         self._state = enums.Database.State(response.state)
         self._create_time = _pb_timestamp_to_datetime(response.create_time)
         self._restore_info = response.restore_info
+        self._version_retention_period = response.version_retention_period
+        self._earliest_version_time = _pb_timestamp_to_datetime(
+            response.earliest_version_time
+        )
 
     def update_ddl(self, ddl_statements, operation_id=""):
         """Update DDL for this database.

tests/system/test_system.py

@@ -330,6 +330,64 @@ def test_create_database(self):
         ]
         self.assertIn(temp_db_id, database_ids)
 
+    @unittest.skipIf(
+        USE_EMULATOR, "PITR-lite features are not supported by the emulator"
+    )
+    def test_create_database_pitr_invalid_retention_period(self):
+        pool = BurstyPool(labels={"testcase": "create_database_pitr"})
+        temp_db_id = "temp_db" + unique_resource_id("_")
+        retention_period = "0d"
+        ddl_statements = [
+            "ALTER DATABASE {}"
+            " SET OPTIONS (version_retention_period = '{}')".format(
+                temp_db_id, retention_period
+            )
+        ]
+        temp_db = Config.INSTANCE.database(
+            temp_db_id, pool=pool, ddl_statements=ddl_statements
+        )
+        with self.assertRaises(exceptions.InvalidArgument):
+            temp_db.create()
+
+    @unittest.skipIf(
+        USE_EMULATOR, "PITR-lite features are not supported by the emulator"
+    )
+    def test_create_database_pitr_success(self):
+        pool = BurstyPool(labels={"testcase": "create_database_pitr"})
+        temp_db_id = "temp_db" + unique_resource_id("_")
+        retention_period = "7d"
+        ddl_statements = [
+            "ALTER DATABASE {}"
+            " SET OPTIONS (version_retention_period = '{}')".format(
+                temp_db_id, retention_period
+            )
+        ]
+        temp_db = Config.INSTANCE.database(
+            temp_db_id, pool=pool, ddl_statements=ddl_statements
+        )
+        operation = temp_db.create()
+        self.to_delete.append(temp_db)
+
+        # We want to make sure the operation completes.
+        operation.result(30)  # raises on failure / timeout.
+
+        database_ids = [
+            database.database_id for database in Config.INSTANCE.list_databases()
+        ]
+        self.assertIn(temp_db_id, database_ids)
+
+        temp_db.reload()
+        self.assertEqual(temp_db.version_retention_period, retention_period)
+
+        with temp_db.snapshot() as snapshot:
+            results = snapshot.execute_sql(
+                "SELECT OPTION_VALUE AS version_retention_period "
+                "FROM INFORMATION_SCHEMA.DATABASE_OPTIONS "
+                "WHERE SCHEMA_NAME = '' AND OPTION_NAME = 'version_retention_period'"
+            )
+            for result in results:
+                self.assertEqual(result[0], retention_period)
+
     def test_table_not_found(self):
         temp_db_id = "temp_db" + unique_resource_id("_")
 
@@ -382,6 +440,62 @@ def test_update_database_ddl_with_operation_id(self):
 
         self.assertEqual(len(temp_db.ddl_statements), len(ddl_statements))
 
+    @unittest.skipIf(
+        USE_EMULATOR, "PITR-lite features are not supported by the emulator"
+    )
+    def test_update_database_ddl_pitr_invalid(self):
+        pool = BurstyPool(labels={"testcase": "update_database_ddl_pitr"})
+        temp_db_id = "temp_db" + unique_resource_id("_")
+        retention_period = "0d"
+        temp_db = Config.INSTANCE.database(temp_db_id, pool=pool)
+        create_op = temp_db.create()
+        self.to_delete.append(temp_db)
+
+        # We want to make sure the operation completes.
+        create_op.result(240)  # raises on failure / timeout.
+
+        self.assertIsNone(temp_db.version_retention_period)
+
+        ddl_statements = DDL_STATEMENTS + [
+            "ALTER DATABASE {}"
+            " SET OPTIONS (version_retention_period = '{}')".format(
+                temp_db_id, retention_period
+            )
+        ]
+        with self.assertRaises(exceptions.InvalidArgument):
+            temp_db.update_ddl(ddl_statements)
+
+    @unittest.skipIf(
+        USE_EMULATOR, "PITR-lite features are not supported by the emulator"
+    )
+    def test_update_database_ddl_pitr_success(self):
+        pool = BurstyPool(labels={"testcase": "update_database_ddl_pitr"})
+        temp_db_id = "temp_db" + unique_resource_id("_")
+        retention_period = "7d"
+        temp_db = Config.INSTANCE.database(temp_db_id, pool=pool)
+        create_op = temp_db.create()
+        self.to_delete.append(temp_db)
+
+        # We want to make sure the operation completes.
+        create_op.result(240)  # raises on failure / timeout.
+
+        self.assertIsNone(temp_db.version_retention_period)
+
+        ddl_statements = DDL_STATEMENTS + [
+            "ALTER DATABASE {}"
+            " SET OPTIONS (version_retention_period = '{}')".format(
+                temp_db_id, retention_period
+            )
+        ]
+        operation = temp_db.update_ddl(ddl_statements)
+
+        # We want to make sure the operation completes.
+        operation.result(240)  # raises on failure / timeout.
+
+        temp_db.reload()
+        self.assertEqual(temp_db.version_retention_period, retention_period)
+        self.assertEqual(len(temp_db.ddl_statements), len(ddl_statements))
+
     def test_db_batch_insert_then_db_snapshot_read(self):
         retry = RetryInstanceState(_has_all_ddl)
         retry(self._db.reload)()

tests/unit/test_database.py

@@ -260,6 +260,20 @@ def test_restore_info(self):
         )
         self.assertEqual(database.restore_info, restore_info)
 
+    def test_version_retention_period(self):
+        instance = _Instance(self.INSTANCE_NAME)
+        pool = _Pool()
+        database = self._make_one(self.DATABASE_ID, instance, pool=pool)
+        version_retention_period = database._version_retention_period = "1d"
+        self.assertEqual(database.version_retention_period, version_retention_period)
+
+    def test_earliest_version_time(self):
+        instance = _Instance(self.INSTANCE_NAME)
+        pool = _Pool()
+        database = self._make_one(self.DATABASE_ID, instance, pool=pool)
+        earliest_version_time = database._earliest_version_time = self._make_timestamp()
+        self.assertEqual(database.earliest_version_time, earliest_version_time)
+
     def test_spanner_api_property_w_scopeless_creds(self):
 
         client = _Client()
@@ -575,6 +589,8 @@ def test_reload_success(self):
             state=2,
             create_time=_datetime_to_pb_timestamp(timestamp),
             restore_info=restore_info,
+            version_retention_period="1d",
+            earliest_version_time=_datetime_to_pb_timestamp(timestamp),
         )
         api.get_database.return_value = db_pb
         instance = _Instance(self.INSTANCE_NAME, client=client)
@@ -585,6 +601,8 @@ def test_reload_success(self):
         self.assertEqual(database._state, enums.Database.State.READY)
         self.assertEqual(database._create_time, timestamp)
         self.assertEqual(database._restore_info, restore_info)
+        self.assertEqual(database._version_retention_period, "1d")
+        self.assertEqual(database._earliest_version_time, timestamp)
         self.assertEqual(database._ddl_statements, tuple(DDL_STATEMENTS))
 
         api.get_database_ddl.assert_called_once_with(