SNOW-1706990 Extract non action-related fields in ActionContext to ne…

…w WorkspaceContext (#1652)

Entity classes current have lots of static methods, which were added to support the v1 to v2 migration, but they have way too many parameters, so we want to allow calling these methods on entity instances. 

Right now, the only instance methods are `EntityAction` implementations, which represent the top-level actions that can be performed by these entities, but the static methods we want to convert to instance methods don't need to be `EntityAction` implementations (since they don't represent top-level operations). However, it would be useful for these methods to have access to utilities in the `ActionContext`, like `ctx.default_role` for example. Since these utilities are not really part of the action, let's extract them from `ActionContext` into a new `WorkspaceContext` that gets set as a field of the entity when instantiated by the `WorkspaceManager`.

src/snowflake/cli/_plugins/nativeapp/entities/application.py

@@ -41,7 +41,7 @@
  SameAccountInstallMethod,
 )
 from snowflake.cli._plugins.nativeapp.utils import needs_confirmation
-from snowflake.cli._plugins.workspace.action_context import ActionContext
+from snowflake.cli._plugins.workspace.context import ActionContext
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.console.abc import AbstractConsole
 from snowflake.cli.api.entities.common import EntityBase, get_sql_executor
@@ -118,7 +118,7 @@ class ApplicationEntity(EntityBase[ApplicationEntityModel]):
 
  def action_deploy(
  self,
- ctx: ActionContext,
+ action_ctx: ActionContext,
  from_release_directive: bool,
  prune: bool,
  recursive: bool,
@@ -133,26 +133,29 @@ def action_deploy(
  **kwargs,
  ):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  app_name = model.fqn.identifier
  debug_mode = model.debug
  if model.meta:
- app_role = model.meta.role or ctx.default_role
- app_warehouse = model.meta.warehouse or ctx.default_warehouse
+ app_role = model.meta.role or workspace_ctx.default_role
+ app_warehouse = model.meta.warehouse or workspace_ctx.default_warehouse
  post_deploy_hooks = model.meta.post_deploy
  else:
- app_role = ctx.default_role
- app_warehouse = ctx.default_warehouse
+ app_role = workspace_ctx.default_role
+ app_warehouse = workspace_ctx.default_warehouse
  post_deploy_hooks = None
 
- package_entity: ApplicationPackageEntity = ctx.get_entity(model.from_.target)
+ package_entity: ApplicationPackageEntity = action_ctx.get_entity(
+ model.from_.target
+ )
  package_model: ApplicationPackageEntityModel = (
  package_entity._entity_model # noqa: SLF001
  )
  package_name = package_model.fqn.identifier
  if package_model.meta and package_model.meta.role:
  package_role = package_model.meta.role
  else:
- package_role = ctx.default_role
+ package_role = workspace_ctx.default_role
 
  if not stage_fqn:
  stage_fqn = f"{package_name}.{package_model.stage}"
@@ -169,7 +172,7 @@ def action_deploy(
 
  def deploy_package():
  package_entity.action_deploy(
- ctx=ctx,
+ action_ctx=action_ctx,
  prune=True,
  recursive=True,
  paths=[],
@@ -180,8 +183,8 @@ def deploy_package():
  )
 
  self.deploy(
- console=ctx.console,
- project_root=ctx.project_root,
+ console=workspace_ctx.console,
+ project_root=workspace_ctx.project_root,
  app_name=app_name,
  app_role=app_role,
  app_warehouse=app_warehouse,
@@ -202,21 +205,22 @@ def deploy_package():
 
  def action_drop(
  self,
- ctx: ActionContext,
+ action_ctx: ActionContext,
  interactive: bool,
  force_drop: bool = False,
  cascade: Optional[bool] = None,
  *args,
  **kwargs,
  ):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  app_name = model.fqn.identifier
  if model.meta and model.meta.role:
  app_role = model.meta.role
  else:
- app_role = ctx.default_role
+ app_role = workspace_ctx.default_role
  self.drop(
- console=ctx.console,
+ console=workspace_ctx.console,
  app_name=app_name,
  app_role=app_role,
  auto_yes=force_drop,

src/snowflake/cli/_plugins/nativeapp/entities/application_package.py

@@ -42,7 +42,7 @@
 from snowflake.cli._plugins.nativeapp.utils import needs_confirmation
 from snowflake.cli._plugins.stage.diff import DiffResult
 from snowflake.cli._plugins.stage.manager import StageManager
-from snowflake.cli._plugins.workspace.action_context import ActionContext
+from snowflake.cli._plugins.workspace.context import ActionContext
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.console.abc import AbstractConsole
 from snowflake.cli.api.entities.common import EntityBase, get_sql_executor
@@ -150,10 +150,11 @@ class ApplicationPackageEntity(EntityBase[ApplicationPackageEntityModel]):
  A Native App application package.
  """
 
- def action_bundle(self, ctx: ActionContext, *args, **kwargs):
+ def action_bundle(self, action_ctx: ActionContext, *args, **kwargs):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  return self.bundle(
- project_root=ctx.project_root,
+ project_root=workspace_ctx.project_root,
  deploy_root=Path(model.deploy_root),
  bundle_root=Path(model.bundle_root),
  generated_root=Path(model.generated_root),
@@ -163,7 +164,7 @@ def action_bundle(self, ctx: ActionContext, *args, **kwargs):
 
  def action_deploy(
  self,
- ctx: ActionContext,
+ action_ctx: ActionContext,
  prune: bool,
  recursive: bool,
  paths: List[Path],
@@ -175,6 +176,7 @@ def action_deploy(
  **kwargs,
  ):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  package_name = model.fqn.identifier
 
  if force:
@@ -185,15 +187,15 @@ def action_deploy(
  policy = DenyAlwaysPolicy()
 
  return self.deploy(
- console=ctx.console,
- project_root=ctx.project_root,
+ console=workspace_ctx.console,
+ project_root=workspace_ctx.project_root,
  deploy_root=Path(model.deploy_root),
  bundle_root=Path(model.bundle_root),
  generated_root=Path(model.generated_root),
  artifacts=model.artifacts,
  bundle_map=None,
  package_name=package_name,
- package_role=(model.meta and model.meta.role) or ctx.default_role,
+ package_role=(model.meta and model.meta.role) or workspace_ctx.default_role,
  package_distribution=model.distribution,
  prune=prune,
  recursive=recursive,
@@ -202,32 +204,34 @@ def action_deploy(
  validate=validate,
  stage_fqn=stage_fqn or f"{package_name}.{model.stage}",
  package_warehouse=(
- (model.meta and model.meta.warehouse) or ctx.default_warehouse
+ (model.meta and model.meta.warehouse) or workspace_ctx.default_warehouse
  ),
  post_deploy_hooks=model.meta and model.meta.post_deploy,
  package_scripts=[], # Package scripts are not supported in PDFv2
  policy=policy,
  )
 
- def action_drop(self, ctx: ActionContext, force_drop: bool, *args, **kwargs):
+ def action_drop(self, action_ctx: ActionContext, force_drop: bool, *args, **kwargs):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  package_name = model.fqn.identifier
  if model.meta and model.meta.role:
  package_role = model.meta.role
  else:
- package_role = ctx.default_role
+ package_role = workspace_ctx.default_role
 
  self.drop(
- console=ctx.console,
+ console=workspace_ctx.console,
  package_name=package_name,
  package_role=package_role,
  force_drop=force_drop,
  )
 
  def action_validate(
- self, ctx: ActionContext, interactive: bool, force: bool, *args, **kwargs
+ self, action_ctx: ActionContext, interactive: bool, force: bool, *args, **kwargs
  ):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  package_name = model.fqn.identifier
  if force:
  policy = AllowAlwaysPolicy()
@@ -237,42 +241,43 @@ def action_validate(
  policy = DenyAlwaysPolicy()
 
  self.validate_setup_script(
- console=ctx.console,
- project_root=ctx.project_root,
+ console=workspace_ctx.console,
+ project_root=workspace_ctx.project_root,
  deploy_root=Path(model.deploy_root),
  bundle_root=Path(model.bundle_root),
  generated_root=Path(model.generated_root),
  artifacts=model.artifacts,
  package_name=package_name,
- package_role=(model.meta and model.meta.role) or ctx.default_role,
+ package_role=(model.meta and model.meta.role) or workspace_ctx.default_role,
  package_distribution=model.distribution,
  prune=True,
  recursive=True,
  paths=[],
  stage_fqn=f"{package_name}.{model.stage}",
  package_warehouse=(
- (model.meta and model.meta.warehouse) or ctx.default_warehouse
+ (model.meta and model.meta.warehouse) or workspace_ctx.default_warehouse
  ),
  post_deploy_hooks=model.meta and model.meta.post_deploy,
  package_scripts=[], # Package scripts are not supported in PDFv2
  policy=policy,
  use_scratch_stage=True,
  scratch_stage_fqn=f"{package_name}.{model.scratch_stage}",
  )
- ctx.console.message("Setup script is valid")
+ workspace_ctx.console.message("Setup script is valid")
 
  def action_version_list(
- self, ctx: ActionContext, *args, **kwargs
+ self, action_ctx: ActionContext, *args, **kwargs
  ) -> SnowflakeCursor:
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  return self.version_list(
  package_name=model.fqn.identifier,
- package_role=(model.meta and model.meta.role) or ctx.default_role,
+ package_role=(model.meta and model.meta.role) or workspace_ctx.default_role,
  )
 
  def action_version_create(
  self,
- ctx: ActionContext,
+ action_ctx: ActionContext,
  version: Optional[str],
  patch: Optional[int],
  skip_git_check: bool,
@@ -282,16 +287,17 @@ def action_version_create(
  **kwargs,
  ):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  package_name = model.fqn.identifier
  return self.version_create(
- console=ctx.console,
- project_root=ctx.project_root,
+ console=workspace_ctx.console,
+ project_root=workspace_ctx.project_root,
  deploy_root=Path(model.deploy_root),
  bundle_root=Path(model.bundle_root),
  generated_root=Path(model.generated_root),
  artifacts=model.artifacts,
  package_name=package_name,
- package_role=(model.meta and model.meta.role) or ctx.default_role,
+ package_role=(model.meta and model.meta.role) or workspace_ctx.default_role,
  package_distribution=model.distribution,
  prune=True,
  recursive=True,
@@ -300,7 +306,7 @@ def action_version_create(
  validate=True,
  stage_fqn=f"{package_name}.{model.stage}",
  package_warehouse=(
- (model.meta and model.meta.warehouse) or ctx.default_warehouse
+ (model.meta and model.meta.warehouse) or workspace_ctx.default_warehouse
  ),
  post_deploy_hooks=model.meta and model.meta.post_deploy,
  package_scripts=[], # Package scripts are not supported in PDFv2
@@ -313,24 +319,25 @@ def action_version_create(
 
  def action_version_drop(
  self,
- ctx: ActionContext,
+ action_ctx: ActionContext,
  version: Optional[str],
  interactive: bool,
  force: bool,
  *args,
  **kwargs,
  ):
  model = self._entity_model
+ workspace_ctx = self._workspace_ctx
  package_name = model.fqn.identifier
  return self.version_drop(
- console=ctx.console,
- project_root=ctx.project_root,
+ console=workspace_ctx.console,
+ project_root=workspace_ctx.project_root,
  deploy_root=Path(model.deploy_root),
  bundle_root=Path(model.bundle_root),
  generated_root=Path(model.generated_root),
  artifacts=model.artifacts,
  package_name=package_name,
- package_role=(model.meta and model.meta.role) or ctx.default_role,
+ package_role=(model.meta and model.meta.role) or workspace_ctx.default_role,
  package_distribution=model.distribution,
  version=version,
  force=force,

src/snowflake/cli/_plugins/workspace/action_context.py → src/snowflake/cli/_plugins/workspace/context.py

@@ -7,16 +7,17 @@
 
 
 @dataclass
-class ActionContext:
+class WorkspaceContext:
  """
- An object that is passed to each action when called by WorkspaceManager
+ An object that is passed to each entity when instantiated by WorkspaceManager
+ to allow access to the CLI context without requiring the entities to use
+ get_cli_context().
  """
 
  console: AbstractConsole
  project_root: Path
  get_default_role: Callable[[], str]
  get_default_warehouse: Callable[[], str | None]
- get_entity: Callable
 
  @cached_property
  def default_role(self) -> str:
@@ -25,3 +26,13 @@ def default_role(self) -> str:
  @cached_property
  def default_warehouse(self) -> str | None:
  return self.get_default_warehouse()
+
+
+@dataclass
+class ActionContext:
+ """
+ An object that is passed to each action when called by WorkspaceManager
+ to provide access to metadata about the entity and project being acted upon.
+ """
+
+ get_entity: Callable

src/snowflake/cli/_plugins/workspace/manager.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from typing import Dict
 
-from snowflake.cli._plugins.workspace.action_context import ActionContext
+from snowflake.cli._plugins.workspace.context import ActionContext, WorkspaceContext
 from snowflake.cli.api.cli_global_context import get_cli_context
 from snowflake.cli.api.console import cli_console as cc
 from snowflake.cli.api.entities.common import EntityActions, get_sql_executor
@@ -43,7 +43,13 @@ def get_entity(self, entity_id: str):
  raise ValueError(f"No such entity ID: {entity_id}")
  entity_model_cls = entity_model.__class__
  entity_cls = v2_entity_model_to_entity_map[entity_model_cls]
- self._entities_cache[entity_id] = entity_cls(entity_model)
+ workspace_ctx = WorkspaceContext(
+ console=cc,
+ project_root=self.project_root,
+ get_default_role=_get_default_role,
+ get_default_warehouse=_get_default_warehouse,
+ )
+ self._entities_cache[entity_id] = entity_cls(entity_model, workspace_ctx)
  return self._entities_cache[entity_id]
 
  def perform_action(self, entity_id: str, action: EntityActions, *args, **kwargs):
@@ -53,16 +59,13 @@ def perform_action(self, entity_id: str, action: EntityActions, *args, **kwargs)
  entity = self.get_entity(entity_id)
  if entity.supports(action):
  action_ctx = ActionContext(
- console=cc,
- project_root=self.project_root(),
- get_default_role=_get_default_role,
- get_default_warehouse=_get_default_warehouse,
  get_entity=self.get_entity,
  )
  return entity.perform(action, action_ctx, *args, **kwargs)
  else:
  raise ValueError(f'This entity type does not support "{action.value}"')
 
+ @property
  def project_root(self) -> Path:
  return self._project_root