in client/python/cli/command/privileges.py [0:0]
def execute(self, api: PolarisDefaultApi) -> None:
if self.privileges_subcommand == Subcommands.LIST:
for grant in api.list_grants_for_catalog_role(self.catalog_name, self.catalog_role_name).grants:
print(grant.to_json())
else:
grant = None
if self.privileges_subcommand == Subcommands.CATALOG:
grant = CatalogGrant(
type=Subcommands.CATALOG,
privilege=CatalogPrivilege(self.privilege)
)
elif self.privileges_subcommand == Subcommands.NAMESPACE:
grant = NamespaceGrant(
type=Subcommands.NAMESPACE,
namespace=self.namespace,
privilege=NamespacePrivilege(self.privilege)
)
elif self.privileges_subcommand == Subcommands.TABLE:
grant = TableGrant(
type=Subcommands.TABLE,
namespace=self.namespace,
table_name=self.table,
privilege=TablePrivilege(self.privilege)
)
elif self.privileges_subcommand == Subcommands.VIEW:
grant = ViewGrant(
type=Subcommands.VIEW,
namespace=self.namespace,
view_name=self.view,
privilege=ViewPrivilege(self.privilege)
)
if not grant:
raise Exception(f'{self.privileges_subcommand} is not supported in the CLI')
elif self.action == Actions.GRANT:
request = AddGrantRequest(
grant=grant
)
api.add_grant_to_catalog_role(self.catalog_name, self.catalog_role_name, request)
elif self.action == Actions.REVOKE:
request = RevokeGrantRequest(
grant=grant
)
api.revoke_grant_from_catalog_role(self.catalog_name, self.catalog_role_name, self.cascade, request)
else:
raise Exception(f'{self.action} is not supported in the CLI')