Implemented graceful warning when encounter identifier which does not…

… conform to SnowDDL standards

CHANGELOG.md

@@ -1,8 +1,12 @@
 # Changelog
 
+## [0.28.3] - 2024-06-08
+
+- Implemented graceful warning when encounter identifier which does not conform to SnowDDL standards while processing existing role grants. Previously it caused SnowDDL to stop with hard error.
+
 ## [0.28.2] - 2024-05-28
 
-- Relax view parsing regexp in `VIEW` converter.
+- Relaxed view parsing regexp in `VIEW` converter.
 
 ## [0.28.1] - 2024-05-21
 

snowddl/app/base.py

@@ -469,6 +469,7 @@ def execute(self):
 
  self.engine.connection.close()
  self.output_engine_stats()
+ self.output_engine_warnings()
 
  if self.args.get("show_sql"):
  self.output_executed_ddl()
@@ -539,6 +540,12 @@ def output_engine_stats(self):
  f"Executed {len(self.engine.executed_ddl)} DDL queries, Suggested {len(self.engine.suggested_ddl)} DDL queries"
  )
 
+ def output_engine_warnings(self):
+ for object_type, object_names in self.engine.intention_cache.invalid_name_warning.items():
+ for name in object_names:
+ self.logger.warning(f"Detected {object_type.name} with name [{name}] "
+ f"which does not conform to SnowDDL standards, please rename or drop it manually")
+
  def output_suggested_ddl(self):
  if self.engine.suggested_ddl:
  print("--- Suggested DDL ---\n")

snowddl/app/singledb.py

@@ -284,6 +284,7 @@ def execute(self):
 
  self.engine.connection.close()
  self.output_engine_stats()
+ self.output_engine_warnings()
 
  if self.args.get("show_sql"):
  self.output_executed_ddl()

snowddl/blueprint/ident_builder.py

@@ -46,7 +46,7 @@ def build_role_ident(env_prefix, *args: Union[AccountObjectIdent, str]) -> Accou
  )
 
 
-def build_grant_name_ident_snowflake(grant_name, object_type: ObjectType):
+def build_grant_name_ident_snowflake(object_type: ObjectType, grant_name: str):
  env_prefix = ""
 
  parts = [p.strip('"') for p in grant_name.split(".")]
@@ -81,7 +81,7 @@ def build_grant_name_ident_snowflake(grant_name, object_type: ObjectType):
  raise ValueError(f"Unexpected grant name format [{grant_name}] in Snowflake for object type [{object_type}]")
 
 
-def build_future_grant_name_ident_snowflake(grant_name):
+def build_future_grant_name_ident_snowflake(object_type: ObjectType, grant_name: str):
  env_prefix = ""
 
  parts = [p.strip('"') for p in grant_name.split(".")]

snowddl/cache/intention_cache.py

@@ -22,12 +22,17 @@ def __init__(self, engine: "SnowDDLEngine"):
  self.drop_intention: Dict[ObjectType, Set[str]] = defaultdict(set)
  self.replace_intention: Dict[ObjectType, Set[str]] = defaultdict(set)
 
+ self.invalid_name_warning: Dict[ObjectType, Set[str]] = defaultdict(set)
+
  def add_drop_intention(self, object_type: ObjectType, object_full_name: str):
  self.drop_intention[object_type].add(object_full_name)
 
  def add_replace_intention(self, object_type: ObjectType, object_full_name: str):
  self.replace_intention[object_type].add(object_full_name)
 
+ def add_invalid_name_warning(self, object_type: ObjectType, object_full_name: str):
+ self.invalid_name_warning[object_type].add(object_full_name)
+
  def check_drop_intention(self, object_type: ObjectType, object_full_name: str):
  return object_full_name in self.drop_intention[object_type]
 

snowddl/resolver/abc_role_resolver.py

@@ -77,11 +77,17 @@ def get_existing_role_grants(self, role_name):
  if object_type == ObjectType.ACCOUNT:
  account_grants.append(AccountGrant(privilege=r["privilege"]))
  else:
+ try:
+ grant_name = build_grant_name_ident_snowflake(object_type, r["name"])
+ except ValueError:
+ self.engine.intention_cache.add_invalid_name_warning(object_type, r["name"])
+ continue
+
  grants.append(
  Grant(
  privilege=r["privilege"],
  on=object_type,
- name=build_grant_name_ident_snowflake(r["name"], object_type),
+ name=grant_name,
  )
  )
 
@@ -100,14 +106,18 @@ def get_existing_role_grants(self, role_name):
  # Skip future grants on unknown object types
  continue
 
- name = build_future_grant_name_ident_snowflake(r["name"])
+ try:
+ grant_name = build_future_grant_name_ident_snowflake(object_type, r["name"])
+ except ValueError:
+ self.engine.intention_cache.add_invalid_name_warning(object_type, r["name"])
+ continue
 
  future_grants.append(
  FutureGrant(
  privilege=r["privilege"],
  on_future=object_type,
- in_parent=ObjectType.SCHEMA if isinstance(name, SchemaIdent) else ObjectType.DATABASE,
- name=name,
+ in_parent=ObjectType.SCHEMA if isinstance(grant_name, SchemaIdent) else ObjectType.DATABASE,
+ name=grant_name,
  )
  )
 

snowddl/resolver/inbound_share_role.py

@@ -29,7 +29,7 @@ def get_existing_role_grants(self, role_name):
  Grant(
  privilege="IMPORTED PRIVILEGES",
  on=ObjectType.DATABASE,
- name=build_grant_name_ident_snowflake(r["name"], ObjectType.DATABASE),
+ name=build_grant_name_ident_snowflake(ObjectType.DATABASE, r["name"]),
  )
  )
 

snowddl/resolver/outbound_share.py

@@ -164,7 +164,7 @@ def get_existing_share_grants(self, share_name):
  Grant(
  privilege=r["privilege"],
  on=ObjectType[r["granted_on"]],
- name=build_grant_name_ident_snowflake(r["name"], ObjectType[r["granted_on"]]),
+ name=build_grant_name_ident_snowflake(ObjectType[r["granted_on"]], r["name"]),
  )
  )
 

snowddl/resolver/user_role.py

@@ -23,11 +23,19 @@ def get_existing_role_grants(self, role_name):
  if r["granted_on"] != "ROLE":
  continue
 
+ object_type = ObjectType.ROLE
+
+ try:
+ grant_name = build_grant_name_ident_snowflake(object_type, r["name"])
+ except ValueError:
+ self.engine.intention_cache.add_invalid_name_warning(object_type, r["name"])
+ continue
+
  grants.append(
  Grant(
  privilege=r["privilege"],
  on=ObjectType[r["granted_on"]],
- name=build_grant_name_ident_snowflake(r["name"], ObjectType[r["granted_on"]]),
+ name=grant_name,
  )
  )
 

snowddl/version.py

@@ -1 +1 @@
-__version__ = "0.28.2"
+__version__ = "0.28.3"