グロースエクスパートナーズグループのリレーブログ企画10日目担当の森下です。
前回の記事は「Spring Boot + Kafka：1つのトピックで複数Consumerによる順序処理パターン」でした。
まだご覧になっていない方は、ぜひそちらもチェックしてみてください！
はじめに
Snowflakeの新機能は、ほとんど毎日のように更新されていきます。これを毎日手動でチェックするのは中々に面倒な作業です。RSS があれば Slack に流して終わりですが、探しても購読できるフィードは見当たりませんでした。
代わりに目を付けたのが Snowflake の External Access Integration （外部アクセス統合）です。Snowflake の中から外部の HTTP を叩けるなら、ドキュメントを取りに行くのも Slack へ投げるのも中で完結します。通知の仕組みをわざわざ外に建てる必要はありません。
その発想で、更新頻度が高い英語版のリリースノートを取得し、Snowflake のAI機能である Cortex で日本語に訳し、Slack へ流すところまでを組みました。使うのは External Access Integration、Secret、Network Rule、Cortex AI 関数、Task の 5 つで、いずれも標準機能です。AWS Lambda のような実行環境は不要です。
▼ 実際に届く通知はこんな感じです。
ほかの方法との比較
作り始める前に、もっと簡単な方法がないかを順に検討しました。比べたのは、Snowflake の外に実行環境を増やさずに済むか、日本語訳まで同じ仕組みで完結するか、新着かどうかの判定を自前で持つ必要があるか、の 3 点です。
| 方法 | 追加インフラ | 日本語化 | 新着の判定 |
|---|---|---|---|
| Slack 公式 RSS アプリ | 不要 | できない | アプリ任せ |
| クラウド関数（AWS Lambda など） | 必要 | 自由に組める | 自前 |
通知統合（TYPE = WEBHOOK） |
不要 | 別の仕組みが要る | 自前 |
| 外部アクセス統合 + プロシージャ | 不要 | Cortex で完結 | 自前 |
一番楽なのは RSS アプリで、フィードの URL を貼れば終わりです。ただし購読できるフィードが見当たらず、この時点で候補から外れました。日本語化ができない点も、英語版が先に更新されるという今回の課題には応えられません。
クラウド関数を使う手もあります。処理は好きに書ける代わりに、実行基盤とデプロイの面倒をあらたに抱えることになります。翻訳を外部 API に任せるなら、認証情報の置き場所もひとつ増えます。今回はそこまでの自由度は必要ないと判断しました。
Snowflake の通知統合（TYPE = WEBHOOK）は、SQL だけで Slack に投げられるのが魅力です。ただし今回は一覧の解析と Cortex の呼び出しが避けられず、結局プロシージャを書くことになります。それならプロシージャから直接 Webhook を叩いても手間は変わりません。
残ったのが外部アクセス統合とプロシージャの組み合わせです。Snowflake の中で完結し、Webhook URL は Secret に収まり、権限は既存の RBAC にそのまま乗ります。翻訳も Cortex で済むので、外部の翻訳サービスを別途契約せずに済みました。
構成
処理の流れは次のようになっています。
| # | やること | 使う機能 |
|---|---|---|
| 1 | Task がプロシージャを呼び出す | Task |
| 2 | リリースノートの一覧ページを取ってくる | External Access Integration |
| 3 | ログテーブルと突き合わせて新着だけ拾う | 通常のテーブル |
| 4 | 記事の本文ページを取ってくる | External Access Integration |
| 5 | 英語のまま要約する | SNOWFLAKE.CORTEX.SUMMARIZE |
| 6 | タイトルと要約を日本語に訳す | SNOWFLAKE.CORTEX.COMPLETE |
| 7 | 種別ごとに色を付けて Slack に投稿する | Secret + Incoming Webhook |
| 8 | URL と公開日を記録する | 通常のテーブル |
4 から 8 は記事ごとに繰り返します。1 回の実行で送るのは最大 10 件までにしています。
用意するもの
| 対象 | 必要なもの |
|---|---|
| Snowflake ロール |
ACCOUNTADMIN、または外部アクセス統合を作成できる権限を渡されたロール |
| Snowflake 権限 | 作業先スキーマへの CREATE NETWORK RULE と CREATE SECRET
|
| Cortex | 使いたいモデルが自分のリージョンで利用できること |
| Slack | ワークスペースにアプリをインストールできること |
外部アクセス統合はアカウントレベルのオブジェクトなので、作成にはアカウントに対する CREATE INTEGRATION 権限が必要です。既定でこの権限を持っているのは ACCOUNTADMIN だけです。
Only the ACCOUNTADMIN role has this privilege by default. The privilege can be granted to additional roles as needed.
Cortex AI 関数の権限は、明示的に付与しなくても使える場合がほとんどです。
この SNOWFLAKE.CORTEX_USER データベースロールは、顧客にSnowflake Cortexの機能へのアクセス権を付与するために使用されます。デフォルトでは、このロールは PUBLIC ロールに付与されます。
引用元: https://docs.snowflake.com/ja/sql-reference/snowflake-db-roles#snowflake-cortex-user-database-role
アカウントによっては取り消されていることもあるので、動かなかったら疑ってみてください。
実装
0. 作業場所を用意する
USE ROLE SYSADMIN;
-- 作業用のデータベースとスキーマ
CREATE DATABASE IF NOT EXISTS OPS_DB;
CREATE SCHEMA IF NOT EXISTS OPS_DB.RELEASE_NOTES;
-- 作業用のウェアハウス
CREATE WAREHOUSE IF NOT EXISTS OPS_XS_WH
WAREHOUSE_SIZE = XSMALL
AUTO_SUSPEND = 60
AUTO_RESUME = TRUE
INITIALLY_SUSPENDED = TRUE;
USE DATABASE OPS_DB;
USE SCHEMA RELEASE_NOTES;
USE WAREHOUSE OPS_XS_WH;
以降のオブジェクトはすべてこのスキーマに置きます。データベースとウェアハウスの名前は環境に合わせて読み替えてください。
1. Slack の Webhook を用意する
Incoming Webhook には古いタイプと新しいタイプがあり、古い「レガシーカスタムインテグレーション」版は非推奨になっています。
If you previously created any incoming webhooks using legacy custom integrations, you should switch to using the same functionality with a Slack app instead.
ここで使うのはアプリ経由の現行方式です。手順は次のとおりです。
| 手順 | 操作 |
|---|---|
| 1 |
https://api.slack.com/apps で「Create New App」を選ぶ |
| 2 | 左メニューの「Incoming Webhooks」をオンにする |
| 3 | 「Add New Webhook to Workspace」で投稿先チャンネルを選んで認可する |
| 4 | 発行された URL を控えておく |
プライベートチャンネルに投稿したい場合は、先に自分がそのチャンネルに入っておいてください。
発行された URL はそれ自体が秘密情報です。スクリーンショットやコードを外に出すときは必ずマスクしてください。
Your webhook URL contains a secret. Don't share it online, including via public version control repositories. Slack actively searches out and revokes leaked secrets.
引用元: https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks/#create_a_webhook
2. Network Rule と Secret を作る
-- 外部への通信先を明示的に許可する
CREATE OR REPLACE NETWORK RULE RELEASE_NOTES_EGRESS_RULE
MODE = EGRESS
TYPE = HOST_PORT
VALUE_LIST = ('docs.snowflake.com', 'hooks.slack.com');
-- Webhook URL を Secret に預ける
CREATE OR REPLACE SECRET SLACK_WEBHOOK_SECRET
TYPE = GENERIC_STRING
SECRET_STRING = 'https://hooks.slack.com/services/XXXXXXXXX/XXXXXXXXXXX/XXXXXXXXXXXXXXXXXXXXXXXX';
GENERIC_STRING は、単純な文字列をひとつ預けておくための型です。
TYPE = GENERIC_STRING
機密性の高い文字列値を保存するためのシークレットを指定します。
引用元: https://docs.snowflake.com/ja/sql-reference/sql/create-secret#generic-string-parameters
Secret のふるまいをまとめると次のようになります。
| 項目 | どうなるか |
|---|---|
| 値を見る |
SHOW SECRETS や DESCRIBE SECRET では見えない |
| 値を取り出す | UDF やプロシージャのハンドラーコードからだけ |
3. External Access Integration を作る
ここからはアカウントレベルのオブジェクトを作るので、ロールを切り替えます。
USE ROLE ACCOUNTADMIN;
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION RELEASE_NOTES_EAI
ALLOWED_NETWORK_RULES = (RELEASE_NOTES_EGRESS_RULE)
ALLOWED_AUTHENTICATION_SECRETS = (SLACK_WEBHOOK_SECRET)
ENABLED = TRUE;
この記事では、ここから先も ACCOUNTADMIN のまま作っています。運用に乗せるなら、統合を作るロールと実際に動かすロールは分けたほうが安全です。その場合は次の権限を渡します。
GRANT USAGE ON INTEGRATION RELEASE_NOTES_EAI TO ROLE <実行ロール>;
GRANT READ ON SECRET OPS_DB.RELEASE_NOTES.SLACK_WEBHOOK_SECRET TO ROLE <実行ロール>;
GRANT USAGE ON DATABASE OPS_DB TO ROLE <実行ロール>;
GRANT USAGE ON SCHEMA OPS_DB.RELEASE_NOTES TO ROLE <実行ロール>;
統合を作るロール自身にも、参照するシークレットへの USAGE が要ります。実行ロールに渡す READ とは別物なので、ACCOUNTADMIN 以外で作る場合は忘れないでください。
統合によって参照されるすべてのシークレットに必要です。
このとき引っかかりやすいのが PUBLIC ロールです。Notebook 向けのドキュメントには、統合への USAGE を PUBLIC に付けても効かないと明記されています。ロールを明示して渡しておくのが無難です。
ノートブックの作成に使用されるロールは、 EAI で USAGE を持っている必要があります。PUBLIC ロールに USAGE を付与しても機能しません。
4. ログテーブルを作る
CREATE TABLE IF NOT EXISTS RELEASE_NOTES_SENT_LOG (
article_url STRING NOT NULL,
article_date DATE,
sent_at TIMESTAMP_NTZ DEFAULT CURRENT_TIMESTAMP(),
PRIMARY KEY (article_url)
);
| カラム | 役割 |
|---|---|
article_url |
重複を判定するためのキー |
article_date |
差分を絞り込むための公開日 |
sent_at |
いつ通知したかの記録 |
ここで知っておきたいのが、標準テーブルでは主キーが効かないことです。
標準テーブルでは、Snowflake は PRIMARY KEY 制約を強制しませんが、 ハイブリッドテーブル では強制されます。
引用元: https://docs.snowflake.com/ja/sql-reference/sql/show-primary-keys#usage-notes
同じ URL をもう一度入れてもエラーにはなりません。重複を防ぐのはアプリ側の役目なので、次のプロシージャでは Python 側の集合とログテーブルとの照合を二重にかけています。
5. ストアドプロシージャを作る
関数ごとの役割は次のとおりです。
| 関数 | 役割 |
|---|---|
build_http_session |
リトライ付きの HTTP セッションを用意する |
extract_date_from_url |
URL パスから公開日を取り出す |
extract_date_from_text |
ページ上のラベルから公開日を取り出す |
fetch_release_notes_index |
一覧ページから URL とタイトルと日付を集める |
filter_new_articles |
ログと突き合わせて新着だけに絞る |
fetch_article_content |
記事の本文を取ってくる |
summarize_content |
本文を要約して日本語に訳す |
translate_title |
タイトルを日本語に訳す |
classify_release_type |
リリース種別を判定して色とアイコンを決める |
post_to_slack |
attachments 形式で Slack に投げる |
process_article |
1 記事分の処理をまとめて回す |
main |
全体の流れを組み立てて実行結果を返す |
主な定数は次のように決めました。
| 定数 | 値 | ねらい |
|---|---|---|
MAX_NOTIFICATIONS |
10 | 1 回の実行で送る上限 |
MAX_ARTICLE_AGE_DAYS |
30 | 古い記事をはじく閾値 |
MAX_CONTENT_LENGTH |
3000 | 要約に渡す本文の最大長 |
HTTP_TIMEOUT |
30 | 外部通信のタイムアウト秒数 |
プロシージャの全文（クリックで展開）
CREATE OR REPLACE PROCEDURE NOTIFY_SNOWFLAKE_RELEASE_NOTES()
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.11'
PACKAGES = ('snowflake-snowpark-python','requests','beautifulsoup4')
HANDLER = 'main'
EXTERNAL_ACCESS_INTEGRATIONS = (RELEASE_NOTES_INTEGRATION)
SECRETS = ('slack_webhook'=SLACK_WEBHOOK)
EXECUTE AS CALLER
AS
$$
import _snowflake
import requests
import json
import re
import logging
from datetime import date, timedelta
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
logger = logging.getLogger(__name__)
# =============================================================================
# 定数
# =============================================================================
RELEASE_NOTES_URL = 'https://docs.snowflake.com/en/release-notes/all-release-notes?change_type=all'
BASE_URL = 'https://docs.snowflake.com'
LOG_TABLE = 'SNOWFLAKE_LEARNING_DB.RELEASE_NOTES.release_notes_sent_log'
MAX_CONTENT_LENGTH = 3000
MAX_NOTIFICATIONS = 10
MAX_ARTICLE_AGE_DAYS = 30
HTTP_TIMEOUT = 30
TRANSLATION_MODEL = 'claude-sonnet-5'
DEFAULT_ICON = '📝'
DEFAULT_COLOR = '#29B5E8'
MONTH_MAP = {
'JAN': '01', 'JANUARY': '01', 'FEB': '02', 'FEBRUARY': '02',
'MAR': '03', 'MARCH': '03', 'APR': '04', 'APRIL': '04',
'MAY': '05', 'JUN': '06', 'JUNE': '06', 'JUL': '07', 'JULY': '07',
'AUG': '08', 'AUGUST': '08', 'SEP': '09', 'SEPTEMBER': '09',
'OCT': '10', 'OCTOBER': '10', 'NOV': '11', 'NOVEMBER': '11',
'DEC': '12', 'DECEMBER': '12'
}
# =============================================================================
# HTTPセッション(リトライ付き)
# =============================================================================
def build_http_session():
"""一時的な障害に強いHTTPセッションを構築する。
5xx系のレスポンスや接続エラーが発生した場合に最大3回までリトライする。
"""
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=1.0,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=('GET', 'POST'),
raise_on_status=False,
)
adapter = HTTPAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
# =============================================================================
# 日付抽出
# =============================================================================
def extract_date_from_url(href):
"""URLパスから YYYY-MM-DD 形式の日付を抽出する
例: /release-notes/2026/other-features/2026-05-12 → 2026-05-12
"""
m = re.search(r'/release-notes/\d{4}/\w+/(\d{4}-\d{2}-\d{2})', href)
return m.group(1) if m else None
def extract_date_from_text(element):
"""記事リンクの直前にある .text-card-label から公開日を抽出する
ドキュメント順でリンク要素の直前にあるラベルを使うことで、
別のカードのラベルを誤って取得するのを防ぐ。
範囲表記 (例: "MAY 8 - MAY 14, 2026") の場合は開始日を返す。
単一日付 (例: "May 22, 2026") の場合はその日付を返す。
"""
label = element.find_previous(class_='text-card-label')
if not label:
return None
text = label.get_text(strip=True)
year_match = re.search(r'(\d{4})', text)
if not year_match:
return None
month_day = re.search(r'([A-Za-z]+)\s+(\d{1,2})', text)
if not month_day:
return None
month_num = MONTH_MAP.get(month_day.group(1).upper())
if not month_num:
return None
year = year_match.group(1)
return f"{year}-{month_num}-{int(month_day.group(2)):02d}"
# =============================================================================
# 記事一覧の取得
# =============================================================================
def fetch_release_notes_index(http):
"""リリースノート一覧ページから記事を収集する"""
res = http.get(RELEASE_NOTES_URL, timeout=HTTP_TIMEOUT)
res.raise_for_status()
soup = BeautifulSoup(res.text, 'html.parser')
seen = set()
articles = []
for a in soup.select('a[href*="/release-notes/"]'):
href = a.get('href', '')
title = a.get_text(strip=True)
if not href or not title or title == 'Read more' or not href.startswith('/en/'):
continue
article_date = extract_date_from_url(href) or extract_date_from_text(a)
if not article_date:
continue
full_url = f'{BASE_URL}{href}'.split('#')[0].split('?')[0].rstrip('/')
if full_url not in seen:
seen.add(full_url)
articles.append({'url': full_url, 'title_en': title, 'date': article_date})
return articles
# =============================================================================
# 新着フィルタリング
# =============================================================================
def filter_new_articles(session, articles):
"""最終送信日より新しい、かつ、未登録の記事のみ返す
また、MAX_ARTICLE_AGE_DAYSより古い記事は安全策として除外する。
日付抽出に失敗した場合や初回実行時に過去の記事が大量に送信されるのを防ぐ。
"""
if not articles:
return []
# 古すぎる記事を除外するための閾値日付 (YYYY-MM-DD 文字列)
age_threshold = (date.today() - timedelta(days=MAX_ARTICLE_AGE_DAYS)).isoformat()
# MAX(article_date)を文字列として取得することで、ドライバの型変換に依存しない
result = session.sql(
f"SELECT TO_CHAR(MAX(article_date), 'YYYY-MM-DD') AS last_date FROM {LOG_TABLE}"
).collect()
last_date = result[0]['LAST_DATE'] # NULL もしくは 'YYYY-MM-DD' 文字列
# プロシージャ実行後に前日分の記事が投稿される場合があるため、
# 最終送信日の1日前以降の記事も対象とする
if last_date is None:
cutoff_date = None
url_rows = session.sql(f"SELECT article_url FROM {LOG_TABLE}").collect()
else:
cutoff_date = (date.fromisoformat(last_date) - timedelta(days=1)).isoformat()
url_rows = session.sql(
f"SELECT article_url FROM {LOG_TABLE} WHERE article_date >= TO_DATE(?)",
params=[cutoff_date]
).collect()
sent_urls = {row['ARTICLE_URL'] for row in url_rows}
filtered = []
for a in articles:
if a['url'] in sent_urls:
continue
if cutoff_date is not None and a['date'] < cutoff_date:
continue
# 閾値より古い記事は除外する (日付抽出ミスや初回実行時の保険)
if a['date'] < age_threshold:
continue
filtered.append(a)
return filtered
# =============================================================================
# 記事本文の取得・要約・翻訳
# =============================================================================
def fetch_article_content(http, url):
"""記事ページの本文テキストを取得する"""
try:
res = http.get(url, timeout=HTTP_TIMEOUT)
res.raise_for_status()
soup = BeautifulSoup(res.text, 'html.parser')
main_content = (
soup.select_one('article')
or soup.select_one('.document')
or soup.select_one('main')
)
return (main_content or soup).get_text(strip=True)[:MAX_CONTENT_LENGTH]
except Exception as e:
logger.warning(f"Failed to fetch article content from {url}: {e}")
return ''
def summarize_content(session, content):
"""英語の記事本文をCortexで要約し、日本語に翻訳して返す"""
if not content:
return ''
# fetch_article_content で既にMAX_CONTENT_LENGTHでトリム済み
result = session.sql(
"SELECT SNOWFLAKE.CORTEX.SUMMARIZE(?) AS summary",
params=[content]
).collect()
summary_en = result[0]['SUMMARY']
prompt = (
"Snowflakeリリースノートの英語要約を自然な日本語に翻訳。"
"製品名・機能名・関数名・技術用語はそのまま。3文以内。訳のみ出力。\n\n"
f"{summary_en}"
)
result = session.sql(
f"SELECT SNOWFLAKE.CORTEX.COMPLETE('{TRANSLATION_MODEL}', ?) AS translated",
params=[prompt]
).collect()
return result[0]['TRANSLATED'].strip()
def translate_title(session, text_en):
"""記事タイトルを日本語に翻訳する"""
prompt = (
"Snowflakeリリースノートのタイトルを日本語訳。"
"製品名・機能名・関数名・技術用語はそのまま。"
"(Generally Available)(Preview)等の提供状況表記は原文のまま。訳のみ出力。\n\n"
f"{text_en}"
)
result = session.sql(
f"SELECT SNOWFLAKE.CORTEX.COMPLETE('{TRANSLATION_MODEL}', ?) AS translated",
params=[prompt]
).collect()
return result[0]['TRANSLATED'].strip()
# =============================================================================
# リリース種別の分類
# =============================================================================
RELEASE_TYPE_CONFIG = {
'Connector': {'icon': '🔌', 'color': '#8B5CF6'},
'Client/Driver/Library':{'icon': '📦', 'color': '#F59E0B'},
'Openflow': {'icon': '🌊', 'color': '#14B8A6'},
'AIM-Virtualization': {'icon': '🖥️', 'color': '#6366F1'},
'Server Release': {'icon': '🚀', 'color': '#EF4444'},
'Feature Update': {'icon': '✨', 'color': '#29B5E8'},
}
def classify_release_type(url, title):
"""URLとタイトルからリリース種別を判定する"""
lower_url = url.lower()
lower_title = title.lower()
if '/connectors/' in lower_url or 'connector' in lower_title:
return 'Connector'
if '/clients-drivers/' in lower_url:
return 'Client/Driver/Library'
if '/openflow' in lower_url or 'openflow' in lower_title:
return 'Openflow'
if '/aim-virtualization/' in lower_url or 'aim-virtualization' in lower_title:
return 'AIM-Virtualization'
if re.search(r'/\d+_\d+', lower_url):
return 'Server Release'
return 'Feature Update'
# =============================================================================
# Slack通知
# =============================================================================
def post_to_slack(http, webhook_url, title, url, summary, release_type):
"""Incoming Webhookでattachments付きメッセージを送信する"""
config = RELEASE_TYPE_CONFIG.get(
release_type, {'icon': DEFAULT_ICON, 'color': DEFAULT_COLOR}
)
payload = {
'attachments': [
{
'color': config['color'],
'fallback': f"{config['icon']} {release_type}: {title}",
'author_name': f"{config['icon']} {release_type}",
'title': title,
'title_link': url,
'text': summary,
'footer': 'Snowflake Release Notes',
'footer_icon': 'https://www.snowflake.com/wp-content/themes/flavor/assets/img/favicon.ico',
}
],
}
res = http.post(
webhook_url,
data=json.dumps(payload),
headers={'Content-Type': 'application/json'},
timeout=HTTP_TIMEOUT,
)
res.raise_for_status()
# =============================================================================
# 1記事ごとの処理
# =============================================================================
def process_article(session, http, webhook_url, article):
"""1件の記事を翻訳・要約してSlackへ通知し、ログを記録する"""
title_ja = translate_title(session, article['title_en'])
content = fetch_article_content(http, article['url'])
summary = summarize_content(session, content)
release_type = classify_release_type(article['url'], article['title_en'])
post_to_slack(
http,
webhook_url,
title=f"「{title_ja}」",
url=article['url'],
summary=summary,
release_type=release_type,
)
session.sql(
f"INSERT INTO {LOG_TABLE} (article_url, article_date) VALUES (?, ?)",
params=[article['url'], article['date']],
).collect()
# =============================================================================
# メイン処理
# =============================================================================
def main(session):
webhook_url = _snowflake.get_generic_secret_string('slack_webhook')
http = build_http_session()
articles = fetch_release_notes_index(http)
new_articles = filter_new_articles(session, articles)
if not new_articles:
return 'No new release notes.'
# 1. 日付の降順で並べ替え、先頭から最新MAX_NOTIFICATIONS件を取得する
latest_articles = sorted(
new_articles, key=lambda x: x['date'], reverse=True
)[:MAX_NOTIFICATIONS]
# 2. 取得した記事を古い順に並べ直して送信対象とする
new_articles = sorted(latest_articles, key=lambda x: x['date'])
success_count = 0
failure_count = 0
for article in new_articles:
try:
process_article(session, http, webhook_url, article)
success_count += 1
except Exception as e:
failure_count += 1
logger.error(
f"Failed to process article {article.get('url')}: {e}",
exc_info=True,
)
continue
return (
f'Sent {success_count} notifications. '
f'Failed {failure_count}. '
f'Total candidates: {len(new_articles)}.'
)
$$;
TRANSLATION_MODEL は、自分のアカウントで使えるモデル名に置き換えてください。公式ドキュメントの例では次のような書き方をしています。
SELECT SNOWFLAKE.CORTEX.COMPLETE( 'claude-sonnet-4-6 ', [ { 'role': 'user', 'content': 'how does a snowflake get its unique pattern?' } ], { 'temperature': 0.7, 'max_tokens': 10 } );
作り終わったら、まず手で呼んで動きを見ておくと安心です。
CALL NOTIFY_RELEASE_NOTES();
6. Task で動かす
CREATE OR REPLACE TASK NOTIFY_RELEASE_NOTES_TASK
WAREHOUSE = OPS_XS_WH
SCHEDULE = 'USING CRON 0 12 * * * Asia/Tokyo'
AS
CALL NOTIFY_RELEASE_NOTES();
ALTER TASK NOTIFY_RELEASE_NOTES_TASK RESUME;
RESUME を忘れるといつまでも動かないので気をつけてください。
タスクが作成されると、一時停止状態で開始されます。タスクにスケジュールに従わせたり、イベントを継続的に検出させたりするには、ALTER TASK ... RESUME を使用します。
引用元: https://docs.snowflake.com/ja/user-guide/tasks-intro#define-schedules-or-triggers
今回は ACCOUNTADMIN で作っているのでそのまま動きますが、別のロールに持たせる場合は権限が 2 つ必要です。
EXECUTE TASK コマンドがタスク実行をトリガーすると、Snowflakeは、タスクに対する OWNERSHIP 権限を持つロールが、タスクに割り当てられたウェアハウスに対する USAGE 権限、およびグローバル EXECUTE TASK 権限を持っていることを確認します。それ以外の場合は、エラーが生成されます。
引用元: https://docs.snowflake.com/ja/sql-reference/sql/execute-task#usage-notes
EXECUTE TASK を渡せるのは ACCOUNTADMIN だけです。運用を考えると、専用のロールにまとめておくほうが扱いやすくなります。
アカウント管理者（ ACCOUNTADMIN ロールを持つユーザー）のみが、ロールに EXECUTE TASK 権限を付与できます。使いやすくするために、カスタムロール（例： TASKADMIN）を作成し、このロールに EXECUTE TASK 権限を割り当てることをお勧めします。
引用元: https://docs.snowflake.com/ja/sql-reference/sql/alter-task#usage-notes
WAREHOUSE を省略すれば、代わりにサーバーレスコンピューティングが使われます。ただし、その場合はアカウントレベルの権限をもう一つ用意する必要がありますし、動作確認でプロシージャを手動で呼ぶときには結局ウェアハウスが必要になります。そのため、この記事ではウェアハウスを指定しています。
EXECUTE MANAGED TASK / アカウント / 実行でサーバーレスコンピューティングリソースに依存するタスクにのみ必要です。
引用元: https://docs.snowflake.com/ja/sql-reference/sql/create-task#access-control-requirements
実装で考慮した点
実装を進めるなかで判断が要った点と、その対応をまとめます。
| 考慮した点 | 対応 |
|---|---|
| 公開日が URL から取れない記事がある | まず URL パスを見て、だめならページ上のラベルから拾う |
| 別のカードのラベルを拾ってしまう |
find_previous で直前のラベルに限定する |
| 本文の位置がページによって違う |
article / .document / main の順に探し、どれも無ければページ全体を使う |
| 初回に過去記事がどっと流れる | 30 日より古い記事をはじき、1 回の送信を 10 件までにする |
| 前日分の記事を取りこぼす | 基準日を 1 日戻した範囲で URL を照合する |
| 1 件失敗すると全部止まる | 記事ごとに例外を受け止めて、成功と失敗の件数を返す |
| 外部サイトが一時的に不安定 | 429 と 5xx に対して最大 3 回リトライする |
前日分の取りこぼしについては、少し補足します。プロシージャが動いたあとに前日付の記事が追加されることもあるので、「最終送信日より後」だけで絞ると漏れてしまいます。日付でざっくり絞ってから URL でしっかり重複を弾く、という二段構えにしているのはそのためです。
送信順にもひと工夫あります。日付の新しい順に 10 件取ったあと、投稿する前に古い順へ並べ替えています。こうするとチャンネル上では時系列に並んでくれます。
動作確認をやり直したいときは、ログテーブルを空にすれば同じ記事をもう一度流せます。ただし本番のチャンネルに向けたまま実行すると通知が二重に飛ぶので、テスト用の Webhook に差し替えてから試すのが安全です。
この構成の利点
| 観点 | 中身 |
|---|---|
| インフラ | 実行基盤を建てる必要も、デプロイの面倒を見る必要もない |
| セキュリティ | Webhook URL が Secret に収まり、通信先もホスト単位で絞られる |
| 監査 | 外部アクセス履歴、Task の実行履歴、Cortex の使用量を同じ場所で追える |
| 権限 | 既存の RBAC にそのまま乗る |
外部アクセスの追跡については、公式ドキュメントにも記載があります。
管理者は EXTERNAL_ACCESS_HISTORY ビューを使うことで、外部ネットワークロケーションへのリクエストをモニターできます。
残る課題
いちばんの弱点は、HTML を直接読んでいることです。公式に購読できる仕組みがない以上、ページの作りが変われば動かなくなります。取得は 1 日 1 回に抑えていますが、いつか壊れるものだと思って付き合うのがよさそうです。
通知には必ず原文へのリンクを付けています。訳文はあくまで目を通すきっかけで、正確なところは原文で確かめてもらう前提の作りです。
コスト
| 要素 | どのくらいか |
|---|---|
| ウェアハウス | 1 日 1 回の短い処理なので最小サイズで足りる |
| Cortex 呼び出し | 1 記事あたり要約 1 回と翻訳 2 回、1 実行で最大 10 記事 |
| データ転送 | テキストだけなので気にしなくてよい |
記事の本文まで丸ごと訳すとトークン数がぐっと増えるので、その場合は事前に試算しておくと安心です。
おわりに
公式の購読手段がないところから始めて、Snowflake の中で完結する通知の仕組みを作ってみました。External Access Integration、Secret、Network Rule、Cortex AI 関数、Task、Python ストアドプロシージャと一通り触ることになるので、これらの機能を覚える題材としてもちょうどよかったです。
既存の機能を組み合わせるだけで済んだのは収穫でした。同じことで困っている方の参考になればうれしいです。
参考リンク
Snowflake
Slack
| タイトル | URL |
|---|---|
| Slack Incoming Webhooks | https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks |
| Slack レガシーカスタムインテグレーション | https://docs.slack.dev/legacy/legacy-custom-integrations/legacy-custom-integrations-incoming-webhooks |