feat: add encryption_info to Database

google/cloud/spanner_v1/database.py

@@ -142,6 +142,7 @@ def __init__(
         self._restore_info = None
         self._version_retention_period = None
         self._earliest_version_time = None
+        self._encryption_info = None
         self.log_commit_stats = False
         self._logger = logger
         self._encryption_config = encryption_config
@@ -269,6 +270,14 @@ def encryption_config(self):
         """
         return self._encryption_config
 
+    @property
+    def encryption_info(self):
+        """Encryption info for this database.
+        :rtype: a list of :class:`~google.cloud.spanner_admin_instance_v1.types.EncryptionInfo`
+        :returns: a list of objects representing encryption info for this database
+        """
+        return self._encryption_info
+
     @property
     def ddl_statements(self):
         """DDL Statements used to define database schema.
@@ -403,6 +412,7 @@ def reload(self):
         self._version_retention_period = response.version_retention_period
         self._earliest_version_time = response.earliest_version_time
         self._encryption_config = response.encryption_config
+        self._encryption_info = response.encryption_info
 
     def update_ddl(self, ddl_statements, operation_id=""):
         """Update DDL for this database.

tests/unit/test_database.py

@@ -318,6 +318,17 @@ def test_encryption_config(self):
         )
         self.assertEqual(database.encryption_config, encryption_config)
 
+    def test_encryption_info(self):
+        from google.cloud.spanner_admin_database_v1 import EncryptionInfo
+
+        instance = _Instance(self.INSTANCE_NAME)
+        pool = _Pool()
+        database = self._make_one(self.DATABASE_ID, instance, pool=pool)
+        encryption_info = database._encryption_info = [
+            mock.create_autospec(EncryptionInfo, instance=True)
+        ]
+        self.assertEqual(database.encryption_info, encryption_info)
+
     def test_spanner_api_property_w_scopeless_creds(self):
 
         client = _Client()
@@ -682,6 +693,7 @@ def test_reload_not_found(self):
     def test_reload_success(self):
         from google.cloud.spanner_admin_database_v1 import Database
         from google.cloud.spanner_admin_database_v1 import EncryptionConfig
+        from google.cloud.spanner_admin_database_v1 import EncryptionInfo
         from google.cloud.spanner_admin_database_v1 import GetDatabaseDdlResponse
         from google.cloud.spanner_admin_database_v1 import RestoreInfo
         from google.cloud._helpers import _datetime_to_pb_timestamp
@@ -693,6 +705,12 @@ def test_reload_success(self):
         client = _Client()
         ddl_pb = GetDatabaseDdlResponse(statements=DDL_STATEMENTS)
         encryption_config = EncryptionConfig(kms_key_name="kms_key")
+        encryption_info = [
+            EncryptionInfo(
+                encryption_type=EncryptionInfo.Type.CUSTOMER_MANAGED_ENCRYPTION,
+                kms_key_version="kms_key_version",
+            )
+        ]
         api = client.database_admin_api = self._make_database_admin_api()
         api.get_database_ddl.return_value = ddl_pb
         db_pb = Database(
@@ -702,6 +720,7 @@ def test_reload_success(self):
             version_retention_period="1d",
             earliest_version_time=_datetime_to_pb_timestamp(timestamp),
             encryption_config=encryption_config,
+            encryption_info=encryption_info,
         )
         api.get_database.return_value = db_pb
         instance = _Instance(self.INSTANCE_NAME, client=client)
@@ -716,6 +735,7 @@ def test_reload_success(self):
         self.assertEqual(database._earliest_version_time, timestamp)
         self.assertEqual(database._ddl_statements, tuple(DDL_STATEMENTS))
         self.assertEqual(database._encryption_config, encryption_config)
+        self.assertEqual(database._encryption_info, encryption_info)
 
         api.get_database_ddl.assert_called_once_with(
             database=self.DATABASE_NAME,