呼び出し側の権利を使用するSnowparkコンテナーサービスを作成する¶

概要¶

このチュートリアルでは、ユーザーに代わって SQL クエリを実行する際に、発信者権限機能を使用するウェブサービス UI を構築します。

リクエストで提供されたクエリを実行するサービス（query_service）を作成します。既定の設定では、アプリケーションコンテナーはサービスのオーナーロールを使用してサービスユーザーとしてSnowflakeに接続します。しかし、このアプリケーションは、エンドユーザーとしてサービスエンドポイントに接続し、そのユーザーに付与された権限を使用するために、発信者権限機能を使用しています。

テスト時には、ウェブブラウザからサービスを使用します。これは、発信者権限機能が、ネットワーク・イングレスを使用してサービスにアクセスする場合にのみサポートされるためです。サービス関数を使用してサービスにアクセスする場合、発信者権限機能は使用できません。

このサービスでは以下のことを行います。

• パブリックエンドポイントを1つ公開します。

• ユーザーがエンドポイントにログインすると、サービスはクエリを提供するためにウェブ UI を提供します。サービスはSnowflakeでクエリを実行し、結果を返します。このチュートリアルでは、以下の SQL コマンドを実行します。

  SELECT CURRENT_USER(), CURRENT_ROLE();
  

  Copy

  このコマンドは、現在ログインしているユーザーの名前と現在アクティブなロールを返します。これらは両方とも、呼び出し元の権限が使用されているかどうかに依存します。

  • 発信者権限を使用する場合、サービスは発信ユーザーとユーザーの既定のロールとしてSnowflakeに接続します。コマンドはユーザー名と既定のロールを返します。

  • 発信者権限を使用しない場合、既定の動作が実行され、サービスはサービスユーザーおよびサービスの所有者ロールとしてSnowflakeに接続します。したがって、コマンドはサービスユーザー名を返します： SF$SERVICE$unique-id, TEST_ROLE.

このチュートリアルには2つのパートがあります。

パート1: サービスを作成してテストする。 このチュートリアルで提供されるコードをダウンロードし、ステップバイステップの手順に従います。

1. このチュートリアルのサービスコードをダウンロードします。

2. Snowpark Container Services用のDockerイメージをビルドし、アカウントのリポジトリにイメージをアップロードします。

3. サービスを作成します。

4. ネットワークイングレスを使用してサービスと通信し、サービスが公開するパブリックエンドポイントに接続します。ウェブブラウザを使ってパブリック・エンドポイントにログインし、 SELECT CURRENT_USER(); コマンドを実行します。コンテナーがログインユーザーとしてコマンドを実行したことを確認するために、コマンド出力を確認します。

パート2：サービスを理解する。このセクションでは、サービスコードの概要を説明し、アプリケーションコードが呼び出し元の権利をどのように使用するかを強調します。

準備¶

Common Setup に従って、前提条件を構成し、このドキュメントで提供されるすべての Snowpark コンテナーチュートリアルに必要な Snowflake リソースを作成します。

サービスコードのダウンロード¶

クエリサービスを作成するためのコード（Pythonアプリケーション）がプロバイダーとして提供されます。

1. ダウンロード: SnowparkContainerServices -Tutorials.zip </samples/spcs/SnowparkContainerServices-Tutorials.zip> をダウンロードします。

2. ファイルを解凍します。チュートリアルごとに1つのディレクトリが含まれています。 Tutorial-6-callers-rights ディレクトリには以下のファイルがあります。

   • Dockerfile

   • main.py

   • templates/basic_ui.html

イメージの作成とアップロード¶

Snowpark Container Servicesがサポートするlinux/amd64プラットフォーム用のイメージをビルドし、アカウントのイメージリポジトリにイメージをアップロードします（共通セットアップ を参照）。

イメージをビルドしてアップロードする前に、リポジトリに関する情報（リポジトリ URL とレジストリのホスト名）が必要です。詳細については、 レジストリおよびリポジトリ をご参照ください。

リポジトリに関する情報を取得する

1. リポジトリ URL を取得するには、 SHOW IMAGE REPOSITORIES SQL コマンドを実行します。

   SHOW IMAGE REPOSITORIES;
   

   Copy

   • 出力の repository_url 列は、 URL を提供します。以下に例を示します。

     <orgname>-<acctname>.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository
     

   • リポジトリ URL のホスト名はレジストリのホスト名です。以下に例を示します。

     <orgname>-<acctname>.registry.snowflakecomputing.com
     

イメージをビルドし、リポジトリにアップロードする

1. ターミナルウィンドウを開き、解凍したファイルのあるディレクトリに移動します。

2. Dockerイメージをビルドするには、Docker CLI を使用して以下の docker build コマンドを実行します。コマンドは、イメージの構築に使用するファイルの PATH として、現在の作業ディレクトリ (.) を指定することに注意してください。

   docker build --rm --platform linux/amd64 -t <repository_url>/<image_name> .
   

   Copy

   • image_name には、 query_service:latest を使用します。

   例

   docker build --rm --platform linux/amd64 -t myorg-myacct.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository/query_service:latest .
   

   Copy

3. Snowflakeアカウントのリポジトリにイメージをアップロードします。Dockerがあなたの代わりにリポジトリにイメージをアップロードするには、まずSnowflakeでDockerを認証する必要があります。

   1. SnowflakeレジストリでDockerを認証するには、以下のコマンドを実行します。

      docker login <registry_hostname> -u <username>
      

      Copy

      • username には、Snowflakeのユーザー名を指定します。Dockerは、パスワードの入力を求めるプロンプトを表示します。

   2. イメージをアップロードするには、以下のコマンドを実行します。

      docker push <repository_url>/<image_name>
      

      Copy

      例

      docker push myorg-myacct.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository/query_service:latest
      

      Copy

サービスの作成¶

このセクションでは、サービス（query_service）を作成します。

1. コンピューティングプールの準備ができ、サービスを作成するのに適切なコンテキストにいることを確認します。

   1. 以前は 共通セットアップ ステップでコンテキストを設定しました。このステップの SQL ステートメントが正しいコンテキストにあることを確認するために、以下を実行してください。

   USE ROLE test_role;
   USE DATABASE tutorial_db;
   USE SCHEMA data_schema;
   USE WAREHOUSE tutorial_warehouse;
   

   Copy

   1. 共通セットアップ で作成したコンピュートプールが準備できていることを確認するために、 DESCRIBE COMPUTE POOL を実行し、 state が ACTIVE または IDLE であることを確認します。 state が STARTING の場合は、 state が ACTIVE か IDLE に変わるまで待つ必要があります。

   DESCRIBE COMPUTE POOL tutorial_compute_pool;
   

   Copy

2. サービスを作成するには、 test_role を使用して以下のコマンドを実行します。

   CREATE SERVICE query_service
     IN COMPUTE POOL tutorial_compute_pool
     FROM SPECIFICATION $$
       spec:
         containers:
         - name: main
           image: /tutorial_db/data_schema/tutorial_repository/query_service:latest
           env:
             SERVER_PORT: 8000
           readinessProbe:
             port: 8000
             path: /healthcheck
         endpoints:
         - name: execute
           port: 8000
           public: true
       capabilities:
         securityContext:
           executeAsCaller: true
       serviceRoles:
       - name: ui_usage
         endpoints:
         - execute
   $$;
   

   Copy

   注釈

   その名前のサービスがすでに存在する場合は、 DROP SERVICE コマンドを使用して以前に作成したサービスを削除してから、このサービスを作成します。

3. 以下の SQL コマンドを実行して、先ほど作成したサービスの詳細情報を取得します。詳細については、 Snowpark Container Services: サービスの操作 をご参照ください。

   • アカウント内のサービスをリストするには、 SHOW SERVICES コマンドを実行します。

     SHOW SERVICES;
     

     Copy

   • サービスのステータスを取得するには、 SHOW SERVICE CONTAINERS IN SERVICE コマンドを実行します。

     SHOW SERVICE CONTAINERS IN SERVICE query_service;
     

     Copy

   • サービスに関する情報を取得するには、 DESCRIBE SERVICE コマンドを実行します。

     DESCRIBE SERVICE query_service;
     

     Copy

サービスの使用¶

このセクションでは、サービス用に構成された 呼び出し元権限 が機能することを確認します。ブラウザからパブリックエンドポイントにログインし、クエリを実行し、サービスが作成したSnowflakeセッションがサービスユーザーとしてではなく、呼び出しユーザーとして動作することを確認します。

まず、このセクションの SQL ステートメントのコンテキストをセットアップするために、以下を実行します。

USE ROLE test_role;
USE DATABASE tutorial_db;
USE SCHEMA data_schema;
USE WAREHOUSE tutorial_warehouse;

サービスはパブリックエンドポイント（CREATE SERVICE コマンドで提供されるインライン指定を参照）を公開します。そのため、まずウェブブラウザを使ってエンドポイントにログインし、サービスがインターネットに公開しているウェブ UI を使ってサービスエンドポイントにクエリリクエストを送信します。

1. サービスが公開しているパブリックエンドポイントの URL を検索します。

   SHOW ENDPOINTS IN SERVICE query_service;
   

   Copy

   応答の ingress_url 列は URL を提供します。

   例

   p6bye-myorg-myacct.snowflakecomputing.app
   

2. /ui をエンドポイント URL に追加し、ウェブブラウザーに貼り付けます。これにより、サービスは ui() 関数を実行します（main.py を参照）。

   エンドポイント URL に初めてアクセスするときは、Snowflakeにログインするようにリクエストされることに注意してください。

3. サービスの作成に使用したユーザーと同じユーザーを使用します。ログインに成功すると、サービスは以下のウェブ UI のように表示されます。

   

   テキストボックスに以下のコマンドを入力し、Enterキーを押すと結果が表示されます。

   SELECT CURRENT_USER(), CURRENT_ROLE()DONE;
   

   Copy

   サービス仕様に executeAsCaller 機能を含めたため、リクエストが届くと、Snowflake はリクエストに Sf-Context-Current-User-Token ヘッダーを挿入し、リクエストをサービスエンドポイントに転送します。

   説明のため、このチュートリアルのサービスコードでは、呼び出し側とサービスユーザー側の両方でクエリを実行します。

   

   • 呼び出し元（イングレスユーザー）に代わってクエリを実行する： この場合、コードは Snowflake が提供するユーザートークンを使用して、Snowflake と接続するためのログイントークンを構築します。したがって、このサービスは呼び出し元の権限を利用します。Snowflakeは呼び出し元の代わりにクエリを実行し、クエリ結果に呼び出し元の名前とアクティブロール名を表示します。例:

     ['TESTUSER, PUBLIC']
     

   • サービスユーザーの代わりにクエリを実行する： この場合、コードは Snowflake と接続するためのログイントークンを構築する際に、リクエストで Snowflake が提供するユーザートークンを使用しません。そのため、サービスは呼び出し元の権限を利用せず、Snowflakeがサービスユーザーに代わってクエリを実行することになります。クエリ結果にはサービスユーザー名（サービス名と同じ）とアクティブなロールが表示されます。

     ['QUERY_SERVICE, TEST_ROLE']
     

サービスが呼び出し元に代わってクエリ(SELECT CURRENT_USER(), CURRENT_ROLE();)を実行する場合、Snowflakeはこの単純なクエリを実行するためにユーザーのウェアハウスを必要としません。そのため、このサービスには 呼び出し元グラント は必要ありませんでした。次のセクションでは、サービスが呼び出しユーザーに代わって自明でないクエリを実行します。このクエリの実行には、サービスに 呼び出し元グラント をする必要があります。

呼び出し元グラントサービスを利用する¶

このセクションで、サービスは呼び出し元（サービスのイングレスエンドポイントにログインしたユーザー）に代わって以下のクエリを実行します。

SELECT * FROM ingress_user_db.ingress_user_schema.ingress_user_table;

このサービスにはテーブルへのアクセス許可がなく、デフォルト・ウェアハウスでクエリを実行する権限もありません。サービスが呼び出し元に代わってこのクエリを実行できるようにするには、必要な 呼び出し元グラント をサービスに付与します。

このシナリオを示すために、新しいロール（ingress_user_role）と、新しいロールにはアクセスできるがサービスの所有者ロール（test_role）にはアクセスできないテーブル（ingress_user_table）を作成します。そのため、サービスがサービス認証情報を使用してクエリを実行しようとすると、Snowflakeはエラーを返します。しかし、サービスがユーザーに代わってクエリを実行すると、Snowflakeはクエリを実行して結果を返します。

クリーンアップ¶

作成した請求可能リソースを削除してください。詳細については、 チュートリアル3 のステップ5をご参照ください。

サービスコードの見直し¶

このセクションでは、以下のトピックを取り上げます。

• チュートリアルのコードを調べる: クエリ・サービスを実装するコード・ファイルを確認してください。

チュートリアルのコードを調べる¶

ステップ1でダウンロードしたzipファイルには、以下のファイルが含まれています。

• Dockerfile

• main.py

• templates/basic_ui.html

サービスを作成する際にもサービス仕様を使用します。次のセクションでは、これらのコードコンポーネントがどのように連携してサービスを作成するかを説明します。

main.pyファイル¶

この Python ファイルには、リクエストのクエリを実行してクエリ結果を返す最小限の HTTP サーバを実装するコードが含まれています。このコードは、エコーリクエストを送信するためのウェブインターフェイス (UI) を提供します。

from flask import Flask
from flask import request
from flask import render_template
import logging
import os
import sys

from snowflake.snowpark import Session
from snowflake.snowpark.exceptions import *

# Environment variables below will be automatically populated by Snowflake.
SNOWFLAKE_ACCOUNT = os.getenv("SNOWFLAKE_ACCOUNT")
SNOWFLAKE_HOST = os.getenv("SNOWFLAKE_HOST")
SNOWFLAKE_DATABASE = os.getenv("SNOWFLAKE_DATABASE")
SNOWFLAKE_SCHEMA = os.getenv("SNOWFLAKE_SCHEMA")

# Custom environment variables
SNOWFLAKE_USER = os.getenv("SNOWFLAKE_USER")
SNOWFLAKE_PASSWORD = os.getenv("SNOWFLAKE_PASSWORD")
SNOWFLAKE_ROLE = os.getenv("SNOWFLAKE_ROLE")
SNOWFLAKE_WAREHOUSE = os.getenv("SNOWFLAKE_WAREHOUSE")

SERVICE_HOST = os.getenv("SERVER_HOST", "0.0.0.0")
SERVER_PORT = os.getenv("SERVER_PORT", 8080)


def get_logger(logger_name):
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler(sys.stdout)
    handler.setLevel(logging.DEBUG)
    handler.setFormatter(
        logging.Formatter("%(name)s [%(asctime)s] [%(levelname)s] %(message)s")
    )
    logger.addHandler(handler)
    return logger


def get_login_token():
    """
    Read the login token supplied automatically by Snowflake. These tokens
    are short lived and should always be read right before creating any new connection.
    """
    with open("/snowflake/session/token", "r") as f:
        return f.read()


def get_connection_params(ingress_user_token=None):
    """
    Construct Snowflake connection params from environment variables.
    """
    if os.path.exists("/snowflake/session/token"):
        if ingress_user_token:
            logger.info("Creating a session on behalf of user.")
            token = get_login_token() + "." + ingress_user_token
        else:
            logger.info("Creating a session as service user.")
            token = get_login_token()

        return {
            "account": SNOWFLAKE_ACCOUNT,
            "host": SNOWFLAKE_HOST,
            "authenticator": "oauth",
            "token": token,
            "warehouse": SNOWFLAKE_WAREHOUSE,
            "database": SNOWFLAKE_DATABASE,
            "schema": SNOWFLAKE_SCHEMA,
        }
    else:
        return {
            "account": SNOWFLAKE_ACCOUNT,
            "host": SNOWFLAKE_HOST,
            "user": SNOWFLAKE_USER,
            "password": SNOWFLAKE_PASSWORD,
            "role": SNOWFLAKE_ROLE,
            "warehouse": SNOWFLAKE_WAREHOUSE,
            "database": SNOWFLAKE_DATABASE,
            "schema": SNOWFLAKE_SCHEMA,
        }


logger = get_logger("query-service")
app = Flask(__name__)


@app.get("/healthcheck")
def readiness_probe():
    return "I'm ready!"


@app.route("/ui", methods=["GET", "POST"])
def ui():
    """
    Main handler for providing a web UI.
    """
    if request.method == "POST":
        # get ingress user token
        ingress_user = request.headers.get("Sf-Context-Current-User")
        ingress_user_token = request.headers.get("Sf-Context-Current-User-Token")

        if ingress_user:
            logger.info(f"Received a request from user {ingress_user}")

        # getting input in HTML form
        query = request.form.get("query")
        if query:
            logger.info(f"Received a request for query: {query}.")
            query_result_ingress_user = (
                run_query(query, ingress_user_token)
                if ingress_user_token
                else "Token is missing. Can't execute as ingress user."
            )
            query_result_service_user = run_query(query)
            return render_template(
                "basic_ui.html",
                query_input=query,
                query_result_ingress_user=query_result_ingress_user,
                query_result_service_user=query_result_service_user,
            )
    return render_template("basic_ui.html")


@app.route("/query", methods=["GET"])
def query():
    """
    Main handler for providing programmatic access.
    """
    # get ingress user token
    query = request.args.get("query")
    logger.info(f"Received query request: {query}.")
    if query:
        ingress_user = request.headers.get("Sf-Context-Current-User")
        ingress_user_token = request.headers.get("Sf-Context-Current-User-Token")

        if ingress_user:
            logger.info(f"Received a request from user {ingress_user}")

        res = run_query(query, ingress_user_token)
        return str(res)
    return "DONE"


def run_query(query, ingress_user_token=None):
    # start a Snowflake session as the ingress user
    try:
        with Session.builder.configs(
            get_connection_params(ingress_user_token)
        ).create() as session:
            logger.info(
                f"Snowflake connection established (id={session.session_id}). Now executing query: {query}."
            )
            try:
                res = session.sql(query).collect()
                logger.info(f"Query execution done: {query}.")
                return (
                    "[Empty Result]"
                    if len(res) == 0
                    else [", ".join(row) for row in res]
                )
            except Exception as e:
                return "Encountered an error when executing query: " + str(e)
    except Exception as e:
        return "Encountered an error when connecting to Snowflake: " + str(e)

if __name__ == '__main__':
  app.run(host=SERVICE_HOST, port=SERVER_PORT)

Copy

コードで、

• ui 関数は以下のウェブフォームを表示し、ウェブフォームから送信されたクエリリクエストを処理します。

  

  この関数は、 @app.route() デコレーターを使用して、 /ui に対するリクエストをこの関数で処理するように指定します。

  @app.route("/ui", methods=["GET", "POST"])
  def ui():
  

  Copy

  クエリサービスは execute のエンドポイントを公開し (サービス作成時に提供したインラインサービス仕様を参照)、ウェブサービスとの通信を可能にします。ブラウザーで/uiを付加したパブリックエンドポイントの URL をロードすると、ブラウザーはこのパスに対する HTTP GET リクエストを送信し、サーバーはリクエストをこの関数にルーティングします。この関数は実行され、ユーザーがクエリを入力するためのシンプルな HTML フォームを返します。

  ユーザーがクエリを入力してフォームを送信すると、ブラウザはこのパスに対して HTTP POST リクエストを送信します。サービス仕様には executeAsCaller 機能が含まれているため、Snowflake は受信リクエストに Sf-Context-Current-User-Token ヘッダを追加し、この同じ関数にリクエストを転送します（呼び出し元権限を使用した Snowflake への接続 をご参照ください）。

  このコードは run_query 関数を2回実行します。

  • イングレス・ユーザーとして。この場合のログイントークンは、 OAuth トークンとイングレスユーザートークンの両方を連結したものです。

    token = get_login_token() + "." + ingress_user_token
    

    Copy

  • サービスユーザーとして。この場合のログイントークンは、 OAuth トークンのみです。

    token = get_login_token()
    

    Copy

• readiness_probe 関数は、 @app.get() デコレーターを使用して、 /healthcheck に対するリクエストがこの関数で処理されることを指定します。

  @app.get("/healthcheck")
  def readiness_probe():
  

  Copy

  この関数により、Snowflakeはサービスの準備状況を確認することができます。コンテナーが起動すると、Snowflakeはアプリケーションが動作していることと、サービスがリクエストに対応する準備ができていることを確認します。Snowflakeは、このパスで HTTP GET リクエストを送信し（healthプローブ、readinessプローブとして）、正常なコンテナーだけがトラフィックを提供できるようにします。この関数は何でもできます。

• get_logger 関数はログのセットアップに役立ちます。

Dockerfile¶

このファイルには、Dockerを使用してイメージを構築するためのすべてのコマンドが含まれています。

ARG BASE_IMAGE=python:3.10-slim-buster
FROM $BASE_IMAGE
COPY main.py ./
COPY templates/ ./templates/
RUN pip install --upgrade pip && pip install flask snowflake-snowpark-python
CMD ["python", "main.py"]

Copy

Dockerfileには、FlaskライブラリをDockerコンテナーにインストールする手順が含まれています。 main.py のコードは、 HTTP のリクエストを処理するためにFlaskライブラリに依存しています。

/template/basic_ui.html¶

クエリサービスは echoendpoint のエンドポイントを公開し（サービス仕様を参照）、ウェブサービス上での通信を可能にしています。ブラウザで /ui を付加したパブリックエンドポイント URL をロードすると、クエリサービスはこのフォームを表示します。

フォームにクエリを入力してフォームを送信すると、サービスが HTTP レスポンスで結果を返します。

<!DOCTYPE html>
<html lang="en">
  <head>
    <title>Welcome to the query service!</title>
  </head>
  <body>
    <h1>Welcome to the query service!</h1>
    <form action="{{ url_for("ui") }}" method="post">
      <label for="query">query:<label><br>
      <input type="text" id="query" name="query" size="50"><br>
    </form>
    <h2>Query:</h2>
    {{ query_input }}
    <h2>Result (executed on behalf of ingress user):</h2>
    {{ query_result_ingress_user }}
    <h2>Result (executed as service user):</h2>
    {{ query_result_service_user }}
  </body>
</html>

Copy

サービス仕様¶

Snowflakeは、この仕様で提供された情報を使用して、サービスを構成および実行します。

spec:
  containers:
  - name: main
    image: /tutorial_db/data_schema/tutorial_repository/query_service:latest
    env:
      SERVER_PORT: 8000
    readinessProbe:
      port: 8000
      path: /healthcheck
  endpoints:
  - name: execute
    port: 8000
    public: true
capabilities:
  securityContext:
    executeAsCaller: true
serviceRoles:
- name: ui_usage
  endpoints:
  - execute

Copy

サービス仕様では、 spec、 capabilities、 serviceRoles がトップレベルのフィールドです。

• spec には仕様の詳細が記載されています（サービス仕様リファレンス をご参照ください）。このサービスは、パブリックウェブからサービスへのイングレスアクセスを可能にする1つのパブリックエンドポイント(execute)を公開していることに注意してください。

• capabilities executeAsCaller 機能を指定します。これはSnowflakeに、アプリケーションが 呼び出し元権限 を使用する予定であることを伝えます。

• serviceRoles は、 USAGE 権限を付与するサービスロール (ui_usage) とエンドポイント名 (execute) を指定します。

• readinessProbe フィールドは、Snowflakeがreadinessプローブに HTTP GET リクエストを送信するために使用できる port と path を識別して、サービスがトラフィックを処理する準備ができていることを確認します。

  サービスコード（echo_python.py）は、以下のように準備確認を実装します。

  @app.get("/healthcheck")
  def readiness_probe():
  

  Copy

  そのため、仕様ファイルにはそれに応じて container.readinessProbe フィールドが含まれます。

サービス仕様の詳細については、 サービス仕様リファレンス をご参照ください。

次の内容¶

このチュートリアルを完了したため、 サービスの操作 に戻って他のトピックを調べることができます。