ちょっとポートフォリオでも作ろうかと考えているため、取りあえず初めの一歩として、Qiitaの活動を可視化できるように、Qiitaの情報を収集 + 活用できるAPIをGoogle App Engineで作ってみました。
注意
GAEの手順を可能な限り簡略化しているので、一部 Multi-RegionのCloud Storageが自動生成されて、¥3 / monthほどのコストがかかっています。
対象読者
以下の内容を書いてます。
- Flaskの簡単な書き方
- Qiita APIの使い方
- App Engine へのデプロイ方法
- App Engine のスケジューリング
- PythonからDatastoreへの保存・参照方法
GitHub
概要
以下の仕組みで作成しています。
目次
やりたいこと
自分のポートフォリオサイトとかで、Qiitaの、LGTM数 / ストック数 / View数の集計値出したいなと思ったため、App Engineで
- Qiita APIから、各記事の数を集計
- GCPのFirestoreで管理
- 管理データをAPIで使えるように
を実装します。
Project Architecture
/
|- main.py (Flask App)
|- app.yaml (GAEの設定)
|- cron.yaml (Cloud Schedulerの設定)
┗ requirements.txt (Python依存ライブラリリスト)
Qiita APIから集計
集計クラスの実装
初期化クラス
初期化クラスでは
- Qiita APIのアカウント設定
- 自分のQiita記事のリストを取得 (
_get_post_list()) - Datastore Python Clientの設定
を行っています。
Point
- Qiitaのアクセストークンは、App Engineの環境変数で管理(設定方法は後述)
class QiitaWorker(object):
def __init__(self):
self.access_token = os.environ.get('QIITA_TOKEN')
_username = 'sasayabaku'
self.base_url = "https://qiita.com/api/v2/"
self.items_url = self.base_url + "users/" + _username + "/items?per_page=100"
self.item_url = self.base_url + "items/"
self.headers = {
"content-type": "application/json",
"Authorization": "Bearer " + self.access_token
}
self.total = {
"likes": 0,
"views": 0,
"stocks": 0
}
self._get_post_list()
self.datastore_client = datastore.Client()
def _get_post_list(self):
json = requests.get(self.items_url, headers=self.headers)
self.items_list = [item['id'] for item in json.json()]
Qiita 各記事の情報集計
初期化クラスで取得した、記事リストからQiita APIでLGTM / ストック数 / View数 を取得して、for文で合計値として集計。
class QiitaWorker(object):
<・・中略・・>
def collect(self):
for item in tqdm(self.items_list):
_url = self.item_url + str(item)
# 各記事の情報を取得
res_json = requests.get(_url, headers=self.headers).json()
stock_json = self._get_item_stock(item)
item_new_document = {
"id": item,
"likes": res_json['likes_count'],
"views": res_json['page_views_count'],
"stocks": len(stock_json)
}
# 1つの変数に合計値を追加
self.total['likes'] += item_new_document['likes']
self.total['views'] += item_new_document['views']
self.total['stocks'] += item_new_document['stocks']
def _get_item_stock(self, item):
# ストック数を取得するAPIの呼び出し
_url = self.item_url + str(item) + "/stockers"
stock_json = requests.get(_url, headers=self.headers).json()
return stock_json
Datastoreにアップロード
集計したデータを、DataStoreに保存します。
以下を参考にしました。
-
PythonでDatastoreを扱う
Python3でDatastore を動かしてみた - Qiita
class QiitaWorker(object):
<・・中略・・>
def update_datastore(self):
kind = "Aggregate"
task_key = self.datastore_client.key(kind)
task = datastore.Entity(key=task_key)
task['date'] = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
task['likes'] = self.total['likes']
task['views'] = self.total['views']
task['stocks'] = self.total['stocks']
self.datastore_client.put(task)
API部分の実装
App Engineの関数をCronで回すために、APIとして実装します。先ほどの集計関数を実行するAPIを以下で実装。Web App開発で、localhostからの呼び出しを連発する想定なので、CORS対策も入れています。
app = Flask(__name__)
CORS(app)
@app.route('/')
def worker():
worker = QiitaWorker()
worker.collect()
worker.update_datastore()
return jsonify({'message': 'work success'}), 200
Cloud Storage | App保存先設定
App Engineのコードは、Cloud Storageに保存されます。
特に指定がない場合は、プロジェクト名に準拠した名前のバケットが自動作成されそこに管理されますが、今回は、リージョン指定 + Standard Storage設定にするために、自前で作成したバケットに保存するようにします。
gae-example って名前のバケット名にします
App Engineにデプロイ
作成したFlask Appを App Engineにデプロイ。
App Engineをほとんど触ったことがなかったので、以下を参考にしました.
-
Hello Worldを返すクイックスタート
App Engine スタンダード環境での Python 3 のクイックスタート -
App Engineにデプロイする手順マニュアル
アプリケーションのテストとデプロイ | Python 3 の App Engine スタンダード環境 | Google Cloud -
Qiita APIの使い方
Qiita API v2 の概要(非公式) - Qiita
gcloud app deploy --bucket=gs://qiita-worker
アプリケーション設定(app.yamlの記載)
Qiitaトークンをコードに埋め込みたくなかったので、環境変数で管理したかったため、app.yamlの env_variablesで設定します。
runtime: python38
instance_class: F1
env_variables:
QIITA_TOKEN: 'Qiitaのアクセストークン'
デプロイコマンド
gcloud app deploy --bucket=gs://qiita-worker
Cloud Schedulerで、集計処理を定期実行
ここまでで、Qiitaから集計してDatastoreに保存するアプリケーションが完成しました。
ただ、毎回集計のたびにAPIをコールする必要があるため、Cloud Schedulerで定期実行するように設定します。
記述には以下を参考にしました。
-
(公式)App Engineでcronを設定する方法
cron.yaml を使用したジョブのスケジューリング -
シンプルに説明したBlog
GAEでcronを使って定時実行処理をする - morimo7’s blog
cron.yamlをプロジェクト内に作成します。
今回は、/(ルートエンドポイント)にアクセスするため、url部分は空白で記載してます。
あと、01:00と記載していますが、UTCフォーマットなため、日本時間では9時間+の10:00に実行するスケジュールになってます。(タイムゾーンは、リージョンに依存するのか？)
cron:
- description: "daily qiita summary log"
url: ""
schedule: every day 01:00
App Engineにデプロイ
cron.yamlは、アプリケーションとは別にデプロイする必要があります。
同様にデプロイします。
gcloud app deploy cron.yaml --bucket=gs://gae-example
Cloud Schedulerの確認
これで、スケジューリングの設定が完了したので、Cloud Schedulerをチェックすると、APP ENGINEのジョブに入ってるので、今すぐ実行で動作確認をします。
その後、DataStoreを確認して、APIで集計した値が格納されているのを確認します。
この処理が、cron.yamlで設定したスケジュールで定期実行されます。
Datastoreの値を返すAPI実装
ここまでで、自動で毎日Qiitaの情報を集計できるようになったため、次はこの情報をポートフォリオ等のアプリケーションで活用できるようにAPIを実装します。具体的には以下の処理を行っています。
- Datastoreから、対象のデータを取得
- JavaScriptで扱えるように、DateをISO形式に変換
- Datastoreは、順不同なため、日付ごとにソートをかける
@app.route('/get_work')
def get_work():
datastore_client = datastore.Client()
query = datastore_client.query(kind="Aggregate")
datastore_response = query.fetch()
data_list = list(datastore_response)
data = list([])
for item in data_list:
_json = {
"date": item['date'].isoformat(),
"likes": int(item['likes']),
"stocks": int(item['stocks']),
"views": int(item['views'])
}
data.append(_json)
data = sorted(data, key=lambda x: x['date'])
return jsonify({"data": data}), 200
すると以下のレスポンスが取得できます。
{
"data": [
{
"date": "2021-02-15T07:31:52.400326+00:00",
"likes": 633,
"stocks": 325,
"views": 564128
},
{
"date": "2021-02-16T01:00:34.956754+00:00",
"likes": 633,
"stocks": 325,
"views": 564364
},
{
"date": "2021-02-17T01:01:24.152121+00:00",
"likes": 634,
"stocks": 325,
"views": 564797
},
]
}
再度 App Engine にデプロイ
gcloud app deploy --bucket=gs://gae-example
https://<Project名>.an.r.appspot.com/get_workに、リクエストしてデータが取得できるようになりました。
まとめ
App Engine + Cloud Schedulerで、Qiitaの活動を集計して、Datastoreに保存、それらを取得するAPIを実装しました。
今後どう活用するか考えます!