Skip to content

Commit

Permalink
update changie, add type sigs
Browse files Browse the repository at this point in the history
  • Loading branch information
aranke committed Jun 18, 2024
1 parent 5aacd01 commit 126ba32
Show file tree
Hide file tree
Showing 2 changed files with 12 additions and 13 deletions.
2 changes: 1 addition & 1 deletion .changes/unreleased/Fixes-20240516-224134.yaml
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
kind: Fixes
body: Implement get_relation_metadata macro
body: Get catalog metadata for a single relation in the most optimized way using the get_catalog_for_single_relation macro and capability
time: 2024-05-16T22:41:34.256095+01:00
custom:
Author: aranke
Expand Down
23 changes: 11 additions & 12 deletions dbt/adapters/snowflake/impl.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,9 @@
from dataclasses import dataclass
from typing import Mapping, Any, Optional, List, Union, Dict, FrozenSet, Tuple, TYPE_CHECKING

from dbt.adapters.base import BaseRelation
from dbt.adapters.base.impl import AdapterConfig, ConstraintSupport
from dbt.adapters.base.meta import available
from dbt.adapters.capability import CapabilityDict, CapabilitySupport, Support, Capability
from dbt.adapters.snowflake import SnowflakeColumn
from dbt.adapters.snowflake import SnowflakeConnectionManager
from dbt.adapters.snowflake import SnowflakeRelation
from dbt.adapters.sql import SQLAdapter
from dbt.adapters.sql.impl import (
LIST_SCHEMAS_MACRO_NAME,
Expand All @@ -24,6 +20,10 @@
from dbt_common.exceptions import CompilationError, DbtDatabaseError, DbtRuntimeError
from dbt_common.utils import filter_null_values

from dbt.adapters.snowflake import SnowflakeColumn
from dbt.adapters.snowflake import SnowflakeConnectionManager
from dbt.adapters.snowflake import SnowflakeRelation

if TYPE_CHECKING:
import agate

Expand Down Expand Up @@ -140,22 +140,21 @@ def get_columns_in_relation(self, relation):
else:
raise

def _show_object_metadata(self, relation):
def _show_object_metadata(self, relation: SnowflakeRelation) -> Optional[dict]:
try:
kwargs = {"relation": relation}
results = self.execute_macro(SHOW_OBJECT_METADATA_MACRO_NAME, kwargs=kwargs)
except DbtDatabaseError as exc:
# If we don't have permissions to the object, return empty
if "Object does not exist" in str(exc):
return None
else:
# If we don't get any results, return empty

if len(results) == 0:
return None

return results
except DbtDatabaseError:
return None

def get_catalog_for_single_relation(self, relation: BaseRelation) -> Optional[CatalogTable]:
def get_catalog_for_single_relation(
self, relation: SnowflakeRelation
) -> Optional[CatalogTable]:
object_metadata = self._show_object_metadata(relation)

if not object_metadata:
Expand Down

0 comments on commit 126ba32

Please sign in to comment.