GH-1908 Preserve source database description when creating shared Glue databases

- Add optional description parameter to GlueClient.create_database() and _create_glue_database()
- Retrieve and preserve source database description in LFShareManager
- Update check_if_exists_and_create_shared_database_in_target() to pass source description
- Add comprehensive test coverage for various description scenarios
- Maintain backward compatibility with fallback to default description

Fixes #1908

Author: surya-madhav

backend/dataall/modules/s3_datasets_shares/aws/glue_client.py

@@ -15,27 +15,28 @@ def __init__(self, account_id, region, database):
         self._account_id = account_id
         self._region = region
 
-    def create_database(self, location):
+    def create_database(self, location, description=None):
         try:
             log.info(f'Creating database {self._database} in account {self._account_id}...')
             existing_database = self.get_glue_database()
             if existing_database:
                 glue_database_created = True
             else:
-                self._create_glue_database(location)
+                self._create_glue_database(location, description=description)
                 glue_database_created = True
             log.info(f'Successfully created database {self._database} in account {self._account_id}')
             return glue_database_created
         except ClientError as e:
             log.error(f'Failed to create database {self._database} in account {self._account_id} due to {e}')
             raise e
 
-    def _create_glue_database(self, location):
+    def _create_glue_database(self, location, description=None):
         database = self._database
         try:
+            db_description = description or f'dataall database {database}'
             db_input = {
                 'Name': database,
-                'Description': 'dataall database {} '.format(database),
+                'Description': db_description,
                 'CreateTableDefaultPermissions': [],
             }
             if location:

backend/dataall/modules/s3_datasets_shares/services/share_managers/lf_share_manager.py

@@ -266,10 +266,19 @@ def check_if_exists_and_create_shared_database_in_target(self) -> dict:
         """
         Checks if shared database exists in target account
         Creates the shared database if it does not exist
+        Preserves source database description if available
         :return: boto3 glue create_database
         """
+        source_database = self.glue_client_in_source.get_glue_database()
+        source_description = None
+        if source_database:
+            source_description = source_database.get('Database', {}).get('Description')
 
-        database = self.glue_client_in_target.create_database(location=f's3://{self.dataset.S3BucketName}')
+        # Create database in target with source description
+        database = self.glue_client_in_target.create_database(
+            location=f's3://{self.dataset.S3BucketName}',
+            description=source_description,
+        )
         return database
 
     def grant_pivot_role_all_database_permissions_to_shared_database(self) -> True:

tests/modules/s3_datasets_shares/tasks/test_lf_share_manager.py

@@ -298,14 +298,90 @@ def test_grant_pivot_role_all_database_permissions_to_source_database(
     )
 
 
-def test_check_if_exists_and_create_shared_database_in_target(manager_with_mocked_clients, dataset1: S3Dataset):
+def test_check_if_exists_and_create_shared_database_in_target_with_source_description(
+    manager_with_mocked_clients, dataset1: S3Dataset
+):
+    """Test that source database description is retrieved and passed to create_database"""
+    manager, lf_client, glue_client, mock_glue_client = manager_with_mocked_clients
+    glue_client.create_database.return_value = True
+    glue_client.get_glue_database.return_value = {
+        'Database': {
+            'Name': dataset1.GlueDatabaseName,
+            'Description': 'Production customer data - PII encrypted',
+        }
+    }
+    # When
+    manager.check_if_exists_and_create_shared_database_in_target()
+    # Then
+    glue_client.get_glue_database.assert_called_once()
+    glue_client.create_database.assert_called_once()
+    glue_client.create_database.assert_called_with(
+        location=f's3://{dataset1.S3BucketName}',
+        description='Production customer data - PII encrypted',
+    )
+
+
+def test_check_if_exists_and_create_shared_database_in_target_without_source_description(
+    manager_with_mocked_clients, dataset1: S3Dataset
+):
+    """Test that None description is passed when source database has no description"""
+    manager, lf_client, glue_client, mock_glue_client = manager_with_mocked_clients
+    glue_client.create_database.return_value = True
+    glue_client.get_glue_database.return_value = {
+        'Database': {
+            'Name': dataset1.GlueDatabaseName,
+        }
+    }
+    # When
+    manager.check_if_exists_and_create_shared_database_in_target()
+    # Then
+    glue_client.get_glue_database.assert_called_once()
+    glue_client.create_database.assert_called_once()
+    glue_client.create_database.assert_called_with(
+        location=f's3://{dataset1.S3BucketName}',
+        description=None,
+    )
+
+
+def test_check_if_exists_and_create_shared_database_in_target_source_db_not_found(
+    manager_with_mocked_clients, dataset1: S3Dataset
+):
+    """Test that None description is passed when source database doesn't exist"""
+    manager, lf_client, glue_client, mock_glue_client = manager_with_mocked_clients
+    glue_client.create_database.return_value = True
+    glue_client.get_glue_database.return_value = False
+    # When
+    manager.check_if_exists_and_create_shared_database_in_target()
+    # Then
+    glue_client.get_glue_database.assert_called_once()
+    glue_client.create_database.assert_called_once()
+    glue_client.create_database.assert_called_with(
+        location=f's3://{dataset1.S3BucketName}',
+        description=None,
+    )
+
+
+def test_check_if_exists_and_create_shared_database_in_target_empty_description(
+    manager_with_mocked_clients, dataset1: S3Dataset
+):
+    """Test that empty string description is passed through (fallback handled in GlueClient)"""
     manager, lf_client, glue_client, mock_glue_client = manager_with_mocked_clients
     glue_client.create_database.return_value = True
+    glue_client.get_glue_database.return_value = {
+        'Database': {
+            'Name': dataset1.GlueDatabaseName,
+            'Description': '',
+        }
+    }
     # When
     manager.check_if_exists_and_create_shared_database_in_target()
     # Then
+    glue_client.get_glue_database.assert_called_once()
     glue_client.create_database.assert_called_once()
-    glue_client.create_database.assert_called_with(location=f's3://{dataset1.S3BucketName}')
+    glue_client.create_database.assert_called_with(
+        location=f's3://{dataset1.S3BucketName}',
+        description='',
+    )
 
 
 def test_grant_pivot_role_all_database_permissions_to_shared_database(