PythonでSnowflakeのユーザー、ロール、付与を管理する

Pythonを使用して、Snowflakeのユーザー、ロール、および付与を管理できます。Snowflake におけるユーザーとその権限の管理については、 ユーザー管理 をご参照ください。

前提条件

このトピックの例では、Snowflakeと接続するコードを追加して Root オブジェクトを作成し、そこからSnowflake Python Snowflake Python APIs を使用することを想定しています。

たとえば、以下のコードでは、構成ファイルで定義された接続パラメーターを使用してSnowflakeへの接続を作成します。

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)
Copy

出来上がった Session オブジェクトを使って、コードは API のタイプとメソッドを使う Root オブジェクトを作成します。詳細については、 Snowflake Python APIs によるSnowflakeへの接続 をご参照ください。

ユーザーの管理

Snowflakeでユーザーを管理することができます。ユーザーは、Snowflakeのアカウントレベルのオブジェクトです。 Snowflake Python APIs は、ユーザーを2つのタイプに分けて表しています。

  • User: ロールの名前などのプロパティを公開します。

  • UserResource: 対応する User オブジェクトを取得し、ユーザーをドロップするために使用できるメソッドを公開します。

ユーザーの作成

UserCollection.create メソッドを呼び出し、作成するデータベースを表す User オブジェクトを渡すと、データベースを作成できます。ユーザーを作成するには、まずユーザー名を指定する User オブジェクトを作成します。

次の例のコードは、 my_user という名前のデータベースを表す User オブジェクトを作成してから、 User オブジェクトを UserCollection.create メソッドに渡してデータベースを作成します。

from snowflake.core.user import User

my_user = User(name="my_user")
root.users.create(my_user)
Copy

ユーザーの詳細の取得

User オブジェクトを返す UserResource.fetch メソッドを呼び出すことで、ユーザーに関する情報を取得できます。

次の例のコードは、 my_user という名前のユーザーの情報を取得します。

my_user = root.users["my_user"].fetch()
print(my_user.to_dict())
Copy

ユーザーをリストする

PagedIter 反復子を返す iter メソッドを使用して、データベースを一覧表示することができます。

次の例のコードは、名前が my で始まるデータベースを一覧表示します。

users = root.users.iter(like="my%")
for user in users:
  print(user.name)
Copy

ユーザーのドロップ

UserResource.drop メソッドを使用してユーザーをドロップできます。

以下の例のコードは、 my_user ユーザーをドロップします。

my_user_res = root.users["my_user"]
my_user_res.drop()
Copy

ロールの管理

Snowflakeではロールを管理することができます。ロールはアカウントレベルのオブジェクトです。 Snowflake Python APIs は、ロールを2つのタイプに分けて表しています。

  • Role: ロールの名前などのプロパティを公開します。

  • RoleResource: 対応する Role オブジェクトを取得し、ロールをドロップするために使用できるメソッドを公開します。

ロールの作成

ロールを作成するには、まずロール名を指定する Role オブジェクトを作成します。

次の例のコードでは、 Role オブジェクトが my_role という名前のデータベースを表します。

from snowflake.core.role import Role

my_role = Role(name="my_role")
root.roles.create(my_role)
Copy

次に、コードは Role オブジェクトを RoleCollection.create メソッドに渡して、データベースロールを作成します。

セッションでロールを使用する

以下の例のコードは、現在のセッションでロール my_role に適用します。

root.session.use_role("my_role")
Copy

ロールをリストする

iter メソッドを使用して、アカウント内のロールをリストすることができます。このメソッドは Role オブジェクトの PagedIter 反復子を返します。

次の例のコードは、アカウント内のすべてのロール名を一覧表示します。

role_list = root.roles.iter()
for role_obj in role_list:
  print(role_obj.name)
Copy

ロールのドロップ

RoleResource.drop メソッドを使用してロールをドロップできます。

以下の例のコードは、 my_role ロールを削除します。

my_role_res = root.roles["my_role"]
my_role_res.drop()
Copy

データベースロールの管理

Snowflakeで データベースロール を管理できます。データベース・ロールはデータベース・レベルのオブジェクトです。 Snowflake Python APIs はデータベースロールを2つのタイプに分けて表しています。

  • DatabaseRole: 名前とコメントなどのデータベース・ロールのプロパティを公開します。

  • DatabaseRoleResource: 対応する DatabaseRole オブジェクトを取得し、データベースロールをドロップするために使用できるメソッドを公開します。

データベースロールの作成

ロールを作成するには、まずロール名を指定する DatabaseRole オブジェクトを作成します。

次の例のコードでは、 DatabaseRole オブジェクトが my_db_role という名前のデータベースを表します。

from snowflake.core.database_role import DatabaseRole

my_db_role = DatabaseRole(
  name="my_db_role",
  comment="sample comment"
)

my_db_role_ref = root.databases['my_db'].database_roles.create(my_db_role)
Copy

次に、コードは DatabaseRole オブジェクトを DatabaseRoleCollection.create メソッドに渡して、データベースロールを作成します。

データベースロールのクローニング

以下の例のコードは、 my_db データベース内の既存の dr1 データベース・ロールのコピーとして、 my_db_2 ターゲット・データベース内に dr2 というデータベース・ロールを作成します。

database_role_ref = root.databases['my_db'].database_roles['dr1'].clone(target_database_role='dr2', target_database='my_db_2')
Copy

データベースのロールをリストする

アカウント内のデータベース・ロールを一覧表示するには、 iter メソッドを使用します。このメソッドは DatabaseRole オブジェクトの PagedIter 反復子を返します。

以下の例のコードは、 my_db データベース内の my_db_role というデータベース・ロールをリストし、結果の数を 1 に制限しています。

db_role_list = root.databases['my_db'].database_roles.iter(limit=1, from_name='my_db_role')
for db_role_obj in db_role_list:
  print(db_role_obj.name)
Copy

データベース・ロールのドロップ

DatabaseRoleResource.drop メソッドを使用してデータベースロールをドロップできます。

以下の例のコードは、 my_db_role データベース・ロールを削除します。

root.databases['my_db'].database_roles['my_db_role'].drop()
Copy

アクセス権限の管理

API を使用して、アカウントロール、データベースロール、またはユーザーに対するセキュアな Snowflake オブジェクトのアクセス権限を管理できます。Snowflake のロール、セキュアなオブジェクト、およびアクセス制御フレームワークの詳細については、 アクセス制御の概要 をご参照ください。

アカウントロールの場合

以下のコード例は、 アカウントロール に対して、 API 権限の付与、権限の取り消し、および権限の一覧表示を行う操作を示しています。

権限を付与する

from snowflake.core.role import Securable

root.roles['my_role'].grant_privileges(
    privileges=["OPERATE"], securable_type="WAREHOUSE", securable=Securable(name='my_wh')
)
Copy

ロールの付与

from snowflake.core.role import Securable

root.roles['my_role'].grant_role(role_type="ROLE", role=Securable(name='my_role_1'))
Copy

すべてに対する権限の付与

from snowflake.core.role import ContainingScope

root.roles['my_role'].grant_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

将来の権限を付与する

from snowflake.core.role import ContainingScope

root.roles['my_role'].grant_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

権限の剥奪

from snowflake.core.role import Securable

root.roles['my_role'].revoke_privileges(
    privileges=["OPERATE"], securable_type="WAREHOUSE", securable=Securable(name='my_wh')
)
Copy

ロールを取り消す

from snowflake.core.role import Securable

root.roles['my_role'].revoke_role(role_type="ROLE", role=Securable(name='my_role_1'))
Copy

オブジェクトに対する権限を取り消しする

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

将来の権限を剥奪する

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

権限の付与オプションを取り消す

from snowflake.core.role import Securable

 root.roles['my_role'].revoke_grant_option_for_privileges(
    privileges=["OPERATE"], securable_type="WAREHOUSE", securable=Securable(name='my_wh')
)
Copy

すべての権限の付与オプションを取り消す

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_grant_option_for_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

将来の権限の付与オプションを取り消す

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_grant_option_for_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

ロールの付与のリスト

root.roles['my_role'].iter_grants_to()
Copy

ロールの付与のリスト

root.roles['my_role'].iter_grants_on()
Copy

ロールの付与のリスト

root.roles['my_role'].iter_grants_of()
Copy

ロールに対する将来の付与のリスト

root.roles['my_role'].iter_future_grants_to()
Copy

ユーザーの場合

以下のコード例は、 API 操作、ロールの付与、ロールの取り消し、ユーザーのロール一覧表示の操作を示しています。

ユーザーにロールを付与する

from snowflake.core.user import Securable

root.users['my_user'].grant_role(role_type="ROLE", role=Securable(name='my_role'))
Copy

ユーザーからロールを取り消す

from snowflake.core.user import Securable

root.users['my_user'].revoke_role(role_type="ROLE", role=Securable(name='my_role'))
Copy

ユーザーに付与されたロールのリスト

root.users['my_user'].iter_grants_to()
Copy

データベースロールの場合

以下のコード例は、 データベースロール に対して、 API 権限の付与、権限の取り消し、および付与の一覧表示を行う操作を示しています。

権限を付与する

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].grant_privileges(
    privileges=["MODIFY"], securable_type="DATABASE", securable=Securable(name='my_db')
)
Copy

ロールの付与

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].grant_role(role_type="DATABASE ROLE", role=Securable(name='my_db_role_1'))
Copy

すべてに対する権限の付与

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].grant_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

将来の権限を付与する

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].grant_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

権限の剥奪

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].revoke_privileges(
    privileges=["MODIFY"], securable_type="DATABASE", securable=Securable(name='my_db')
)
Copy

ロールを取り消す

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].revoke_role(role_type="DATABASE ROLE", role=Securable(name='my_db_role_1'))
Copy

すべての権限を剥奪する

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

将来の権限を剥奪する

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

権限の付与オプションを取り消す

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].revoke_grant_option_for_privileges(
    privileges=["MODIFY"], securable_type="DATABASE", securable=Securable(name='my_db')
)
Copy

すべての権限の付与オプションを取り消す

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_grant_option_for_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

将来の権限の付与オプションを取り消す

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_grant_option_for_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)
Copy

ロールの付与のリスト

root.databases['my_db'].database_roles['my_db_role'].iter_grants_to()
Copy

ロールに対する将来の付与のリスト

root.databases['my_db'].database_roles['my_db_role'].iter_future_grants_to()
Copy

Grant リソースを使用して付与を管理する --- 非推奨

GRANT <権限> 操作を実行して、セキュアな Snowflake オブジェクトのアクセス権限をロールに付与することができます。

権限を付与する

Snowflake オブジェクトに権限を付与するには、まず以下の属性を指定した Grant オブジェクトを作成します。

  • grantee: 権限を付与されるロールまたはユーザー。

  • securable: 権限によって保護されているSnowflakeオブジェクト。

  • privileges: ロールに与えられている権限。

アカウントの CREATE 権限をロールに付与する

次の例のコードは、現在の Snowflake アカウントのロール my_role に権限 create_databasecreate_warehouse を付与する付与操作を表す Grant オブジェクトを作成します。このコードは root.grants.grant メソッドを使用して操作を実行します。

from snowflake.core.grant import Grant
from snowflake.core.grant._grantee import Grantees
from snowflake.core.grant._privileges import Privileges
from snowflake.core.grant._securables import Securables

root.grants.grant(
  Grant(
    grantee=Grantees.role(name='my_role'),
    securable=Securables.current_account,
    privileges=[Privileges.create_database,
                Privileges.create_warehouse],
  )
)
Copy

ロールへのデータベース権限の付与

以下の例のコードでは、 my_db のデータベース上の インポートされた権限my_role のロールに付与しています。

from snowflake.core.grant import Grant
from snowflake.core.grant._grantee import Grantees
from snowflake.core.grant._privileges import Privileges
from snowflake.core.grant._securables import Securables

root.grants.grant(
  Grant(
    grantee=Grantees.role('my_role'),
    securable=Securables.database('my_db'),
    privileges=[Privileges.imported_privileges],
  )
)
Copy

別のロールにロールを付与する

ロールを別のロールに付与すると、ロール間に「親子」関係が作成されます(ロール階層 とも呼ばれます)。

以下の例のコードは、 my_role ユーザー・ロールを ACCOUNTADMIN システム・ロールに付与します。

from snowflake.core.grant import Grant
from snowflake.core.grant._grantee import Grantees
from snowflake.core.grant._securables import Securables

root.grants.grant(
  Grant(
    grantee=Grantees.role('ACCOUNTADMIN'),
    securable=Securables.role('my_role'),
  )
)
Copy