[td] Don’t raise errors when adding dbt files (mage-ai#4191)

* [td] Don’t raise errors when adding dbt files

* fix

* fix

Author: tommydangerous

mage_ai/data_preparation/models/block/dbt/block_sql.py

@@ -186,23 +186,40 @@ def upstream_dbt_blocks(self, read_only=False) -> List['DBTBlockSQL']:
         """
         # Get upstream nodes via dbt list
         with Profiles(self.project_path, self.pipeline.variables) as profiles:
-            args = [
-                'list',
-                # project-dir
-                # /home/src/default_repo/default_platform/default_repo/dbt/demo
-                '--project-dir', self.project_path,
-                '--profiles-dir', str(profiles.profiles_dir),
-                '--select', '+' + Path(self.__get_original_file_path()).stem,
-                '--output', 'json',
-                '--output-keys', 'unique_id original_file_path depends_on',
-                '--resource-type', 'model',
-                '--resource-type', 'snapshot'
-            ]
-            res, _success = DBTCli(args).invoke()
-        if res:
-            nodes_init = [simplejson.loads(node) for node in res]
-        else:
-            return []
+            try:
+                args = [
+                    'list',
+                    # project-dir
+                    # /home/src/default_repo/default_platform/default_repo/dbt/demo
+                    '--project-dir', self.project_path,
+                    '--profiles-dir', str(profiles.profiles_dir),
+                    '--select', '+' + Path(self.__get_original_file_path()).stem,
+                    '--output', 'json',
+                    '--output-keys', 'unique_id original_file_path depends_on',
+                    '--resource-type', 'model',
+                    '--resource-type', 'snapshot'
+                ]
+                res, _success = DBTCli(args).invoke()
+
+                if res:
+                    nodes_init = [simplejson.loads(node) for node in res]
+                else:
+                    return []
+            except Exception as err:
+                print(f'[ERROR] DBTBlockSQL.upstream_dbt_blocks: {err}.')
+                return [
+                    self.build_dbt_block(
+                        block_class=DBTBlock,
+                        block_dict=dict(
+                            block_type=self.type,
+                            configuration=self.configuration,
+                            language=self.language,
+                            name=self.uuid,
+                            pipeline=self.pipeline,
+                            uuid=self.uuid,
+                        ),
+                    )
+                ]
 
         # transform List into dict and remove unnecessary fields
         file_path = self.__get_original_file_path()

mage_ai/data_preparation/models/block/dbt/profiles.py

@@ -135,18 +135,19 @@ def interpolate(self) -> Union[str, os.PathLike]:
 
         # write interpolated profiles.yml
         interpoalted_profiles_full_path = interpolated_profiles_dir / PROFILES_FILE_NAME
-        try:
-            with interpoalted_profiles_full_path.open('w') as f:
-                yaml.safe_dump(self.profiles, f)
-        except Exception as e:
-            raise ProfilesError(
-                f'Failed to write interpolated `{PROFILES_FILE_NAME}`' +
-                f' to `{interpolated_profiles_dir}`: {e}'
-            )
+        if os.path.exists(interpoalted_profiles_full_path):
+            try:
+                with interpoalted_profiles_full_path.open('w') as f:
+                    yaml.safe_dump(self.profiles, f)
+            except Exception as e:
+                raise ProfilesError(
+                    f'Failed to write interpolated `{PROFILES_FILE_NAME}`' +
+                    f' to `{interpolated_profiles_dir}`: {e}'
+                )
 
-        self.__interpolated_profiles_dir = str(interpolated_profiles_dir)
-        self.is_interpolated = True
-        return self.__interpolated_profiles_dir
+            self.__interpolated_profiles_dir = str(interpolated_profiles_dir)
+            self.is_interpolated = True
+            return self.__interpolated_profiles_dir
 
     def __del__(self) -> None:
         self.clean()

mage_ai/data_preparation/models/block/platform/mixins.py

@@ -201,7 +201,7 @@ def build_dbt_block(
         self,
         block_class,
         block_dict,
-        node: Dict,
+        node: Dict = None,
     ):
         block_type = block_dict['block_type']
         configuration = block_dict['configuration'] or {}
@@ -219,7 +219,7 @@ def build_dbt_block(
                 path=remove_base_repo_directory_name(
                     os.path.join(
                         self.project_path,
-                        node['original_file_path'],
+                        (node.get('original_file_path') or '') if node else '',
                     ),
                 ),
             )

mage_ai/tests/data_preparation/models/block/dbt/test_dbt_adapter.py

@@ -1,6 +1,8 @@
+import os
 import shutil
 from decimal import Decimal
 from pathlib import Path
+from unittest.mock import patch
 
 from dbt.include.starter_project import PACKAGE_PATH as starter_project_directory
 
@@ -58,55 +60,105 @@ def test_execute(self):
         Test the Project Interface by reading the original dbt_project.yml
         and return the dictionary.
         """
-        with DBTAdapter(str(self.project_dir)) as dbt_adapter:
-            # create df
-            import pandas as pd
-            df = pd.DataFrame([[1, 'foo'], [2, 'bar']], columns=['id', 'text'])
-            df_dict = df.to_dict(orient='list')
-
-            # create agate table from df
-            from agate import Table
-            table = Table(
-                rows=list(map(list, zip(*[v for _, v in df_dict.items()]))),
-                column_names=df_dict.keys()
-            )
-
-            relation = dbt_adapter.get_relation(
-                database='test',
-                schema='main',
-                identifier='test',
-            )
-            relation_context = dict(this=relation)
-
-            dbt_adapter.execute_macro(
-                'reset_csv_table',
-                context_overide=relation_context,
-                model={'config': {}},
-                old_relation=relation,
-                full_refresh=True,
-                agate_table=table
-            )
-
-            dbt_adapter.execute_macro(
-                'load_csv_rows',
-                context_overide=relation_context,
-                model={'config': {}},
-                agate_table=table
-            )
+        value = 'mage'
 
-            _res, df = dbt_adapter.execute('select * from test.main.test', fetch=True)
+        file_path = os.path.join(
+            self.repo_path,
+            'dbt_test_project',
+            f'.mage_temp_profiles_{value}',
+            'profiles.yml',
+        )
 
-            self.assertEqual(
-                df.to_dict(),
-                {0: {'id': Decimal('1'), 'text': Decimal('2')}, 1: {'id': 'foo', 'text': 'bar'}}
-            )
+        os.makedirs(os.path.dirname(file_path), exist_ok=True)
+        with open(file_path, 'w') as f:
+            f.write('')
+
+        class MockUUID:
+            def __str__(self):
+                return value
+
+            @property
+            def hex(self) -> str:
+                return value
+
+        with patch(
+            'mage_ai.data_preparation.models.block.dbt.profiles.uuid.uuid4',
+            lambda: MockUUID(),
+        ):
+            with DBTAdapter(str(self.project_dir)) as dbt_adapter:
+                # create df
+                import pandas as pd
+                df = pd.DataFrame([[1, 'foo'], [2, 'bar']], columns=['id', 'text'])
+                df_dict = df.to_dict(orient='list')
+
+                # create agate table from df
+                from agate import Table
+                table = Table(
+                    rows=list(map(list, zip(*[v for _, v in df_dict.items()]))),
+                    column_names=df_dict.keys()
+                )
+
+                relation = dbt_adapter.get_relation(
+                    database='test',
+                    schema='main',
+                    identifier='test',
+                )
+                relation_context = dict(this=relation)
+
+                dbt_adapter.execute_macro(
+                    'reset_csv_table',
+                    context_overide=relation_context,
+                    model={'config': {}},
+                    old_relation=relation,
+                    full_refresh=True,
+                    agate_table=table
+                )
+
+                dbt_adapter.execute_macro(
+                    'load_csv_rows',
+                    context_overide=relation_context,
+                    model={'config': {}},
+                    agate_table=table
+                )
+
+                _res, df = dbt_adapter.execute('select * from test.main.test', fetch=True)
+
+                self.assertEqual(
+                    df.to_dict(),
+                    {0: {'id': Decimal('1'), 'text': Decimal('2')}, 1: {'id': 'foo', 'text': 'bar'}}
+                )
 
     def test_credentials(self):
         """
         Test the Project Interface by reading the original dbt_project.yml
         and return the dictionary.
         """
-        with DBTAdapter(str(self.project_dir)) as dbt_adapter:
-            credentials = dbt_adapter.credentials
-        self.assertEqual(credentials.schema, 'main')
-        self.assertEqual(credentials.database, 'test')
+        value = 'mage'
+
+        file_path = os.path.join(
+            self.repo_path,
+            'dbt_test_project',
+            f'.mage_temp_profiles_{value}',
+            'profiles.yml',
+        )
+
+        os.makedirs(os.path.dirname(file_path), exist_ok=True)
+        with open(file_path, 'w') as f:
+            f.write('')
+
+        class MockUUID:
+            def __str__(self):
+                return value
+
+            @property
+            def hex(self) -> str:
+                return value
+
+        with patch(
+            'mage_ai.data_preparation.models.block.dbt.profiles.uuid.uuid4',
+            lambda: MockUUID(),
+        ):
+            with DBTAdapter(str(self.project_dir)) as dbt_adapter:
+                credentials = dbt_adapter.credentials
+            self.assertEqual(credentials.schema, 'main')
+            self.assertEqual(credentials.database, 'test')

mage_ai/tests/data_preparation/models/block/dbt/test_profiles.py

@@ -1,4 +1,6 @@
+import os
 from pathlib import Path
+from unittest.mock import patch
 
 import yaml
 
@@ -58,25 +60,49 @@ def test_interpolate_clean(self):
         - checking the file contents
         - cleaning up the interpolated profiles.yml
         """
-        with Profiles(self.repo_path, self.variables) as profiles:
-            self.assertNotEquals(
-                profiles.profiles_dir,
-                self.repo_path
-            )
+        value = 'mage'
 
-            interpolated_profiles_full_path = Path(profiles.profiles_dir) / 'profiles.yml'
-            with interpolated_profiles_full_path.open('r') as f:
-                interpolated_profiles = yaml.safe_load(f.read())
-            self.assertEqual(
-                self.interpolated_profiles,
-                interpolated_profiles
-            )
+        file_path = os.path.join(
+            self.repo_path,
+            f'.mage_temp_profiles_{value}',
+            'profiles.yml',
+        )
+
+        os.makedirs(os.path.dirname(file_path), exist_ok=True)
+        with open(file_path, 'w') as f:
+            f.write('')
+
+        class MockUUID:
+            def __str__(self):
+                return value
+
+            @property
+            def hex(self) -> str:
+                return value
+
+        with patch(
+            'mage_ai.data_preparation.models.block.dbt.profiles.uuid.uuid4',
+            lambda: MockUUID(),
+        ):
+            with Profiles(self.repo_path, self.variables) as profiles:
+                self.assertNotEquals(
+                    profiles.profiles_dir,
+                    self.repo_path
+                )
+
+                interpolated_profiles_full_path = Path(profiles.profiles_dir) / 'profiles.yml'
+                with interpolated_profiles_full_path.open('r') as f:
+                    interpolated_profiles = yaml.safe_load(f.read())
+                self.assertEqual(
+                    self.interpolated_profiles,
+                    interpolated_profiles
+                )
 
-            self.assertTrue(interpolated_profiles_full_path.exists())
+                self.assertTrue(interpolated_profiles_full_path.exists())
 
-            profiles.clean()
+                profiles.clean()
 
-            self.assertFalse(interpolated_profiles_full_path.exists())
+                self.assertFalse(interpolated_profiles_full_path.exists())
 
     def test_profiles(self):
         profiles = Profiles(self.repo_path, self.variables)