GCP無料枠で遊ぶ② ~Airflow (on Compute Engine), Cloud Functions~

import requests
import json
import datetime
from google.cloud import storage

# function1... 明日の天気をAPIから取得してCloud Storageに保存
def function1(request):
    url = "http://weather.livedoor.com/forecast/webservice/json/v1"
    payload = {"city": 130010} # 東京
    res = requests.get(url, params=payload)
    res_json = json.loads(res.text.replace("\n", "")) # "\n"がエラーを起こすので置換
    tomorrow = datetime.datetime.now() + datetime.timedelta(days=1)
    forecast = [x for x in res_json["forecasts"] if x["date"] == tomorrow.strftime("%Y-%m-%d")][0]
    client = storage.Client()
    bucket = client.get_bucket("xxxxx") # 自分のbucketを作成して置き換えてください
    blob = bucket.blob("forecast.json")
    blob.upload_from_string(json.dumps(forecast))

# function2... Cloud Storageから明日の天気を取得し、LINE Notifyで通知
def send_message(msg):
    url = "https://notify-api.line.me/api/notify"
    token = "xxxxx" # 自分のトークンに置き換えてください
    payload = {"message": msg}
    headers = {"Authorization": "Bearer {}".format(token)}
    requests.post(url, data=payload, headers=headers)

def function2(requests):
    client = storage.Client()
    bucket = client.get_bucket("xxxxx") # 自分のbucketを作成して置き換えてください
    blob = bucket.blob("forecast.json")
    forecast = json.loads(blob.download_as_string())
    send_message(forecast["telop"])

requests==2.22.0
google-cloud-storage==1.26.0

gcloud functions deploy qiita_function1 --entry-point function1 --runtime python37 --trigger-http --ingress-settings internal-only --allow-unauthenticated
gcloud functions deploy qiita_function2 --entry-point function2 --runtime python37 --trigger-http --ingress-settings internal-only --allow-unauthenticated

sudo apt update
sudo apt -y install python3.7 python3-pip
pip3 install --upgrade apache-airflow requests

airflow initdb

# DAGが認識されたときに停止状態にしない（Trueのままだと明示的にairflow unpauseコマンドが必要）
dags_are_paused_at_creation = False

# 開始日が過去のDAGを実行するときに、過去分を実行しない
catchup_by_default = False

# DAGの例を表示しない
load_examples = False

from datetime import timedelta
# The DAG object; we'll need this to instantiate a DAG
from airflow import DAG
# Operators; we need this to operate!
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago
# These args will get passed on to each operator
# You can override them on a per-task basis during operator initialization
import requests
import os
from datetime import timedelta

def exec_functions(url):
    payload = {}
    res = requests.post(url, data=payload)
    if res.status_code//100 != 2:
        raise Exception("response status code is not in 200 - 299")

common_args = {
    'owner': os.environ.get("USER", "unknown"),
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'qiita_sample_v0.0',
    default_args=common_args,
    description='sample dag',
    start_date=days_ago(1),
    schedule_interval="00 09 *  *  *", # 毎日9時に実行（日本時間で18時）
)

task1 = PythonOperator(
    task_id='qiita_function1',
    python_callable=exec_functions,
    #provide_context=True,
    op_kwargs={
        "url": "xxxxx", # 1つめのCloud FunctionsのURL
    },
    dag=dag,
)

task2 = PythonOperator(
    task_id='qiita_function2',
    python_callable=exec_functions,
    #provide_context=True,
    op_kwargs={
        "url": "xxxxx", # 2つめのCloud FunctionsのURL
    },
    dag=dag,
)

task1 >> task2 # タスクの依存関係を指定

airflow scheduler -D