闇の Slack 魔術に対抗する Python 防衛術

この記事は、Slack Advent CalendarJX 通信社 Advent Calendarの最終日です。

メリークリスマス! 素敵なクリスマスをお過ごしでしょうか。取締役の小笠原(@yamitzky)です。

突然ですが、みなさん、ダークモードは好きですか? ダークモードは昨今のソフトウェアのトレンドで、Slack のデスクトップ版も今年の 9 月にダークモードに対応しました。

slackhq.com

しかし Slack をダークモードに設定すると、透過背景・黒文字なカスタム絵文字が見づらいという問題がありました。Slack が仕事のワークフローの中心にある JX 通信社にとって、これは死活問題です。

f:id:yamitzky:20191225003235p:plain

そこで、Slack の闇の魔術(ダークモード)に Python で対抗し、これらの絵文字が見えるようにしたいと思います!

方針

基本方針は簡単です。黒背景に黒文字だと見づらいのが原因なので、背景を白でベタ塗りした、ダークモードフレンドリーな絵文字を作ります。

f:id:yamitzky:20191225003312p:plain

この問題を分解し、次のような処理を Python で行うことを考えます。

  • Slack の API を使い、絵文字の一覧を取得
  • 取得した絵文字のうち、透過背景・黒文字なものを抽出
  • 白背景を合成した絵文字を生成

本当であれば、Python 経由で絵文字を更新したいところですが、公式の API ドキュメントに記載はなく、xoxs から始まるトークンでしか実行できないようです(参考issue)。そのため、本稿では触れません。

絵文字一覧の取得

Slack には emoji.list という API があるので、これを通信ライブラリの requests で実行します。

import requests
# TOKEN には Slack API のトークンを入れる
res = requests.get('https://slack.com/api/emoji.list', headers={'Authorization': f'Bearer {TOKEN}'})
emojis = res.json()['emoji']

emojis 変数は、次のような "絵文字名": "画像URL" という形式の dict 型です。また、結果には alias も含んでいます。

{
    ...
    "saiko": "https://emoji.slack-edge.com/XXXXXXX/saiko/0123456789abcdef.png",
    "shussha": "alias:syussha",  # alias のパターン
    ...
}

絵文字の URL から、画像データ(バイナリ)を取得します。

img_bytes = requests.get(url).content

闇絵文字判定

続いては、取得した画像データが、「黒文字・透過」の絵文字かどうかを判定します。

今回は、Python での画像処理に PillowNumPy を利用します。Pillow は PIL=Python Image Library のフォークで、画像処理用のライブラリです。NumPy は行列計算などに使われるライブラリです。これらライブラリを組み合わせ、取得した絵文字のバイナリ(png)を配列(高さ×幅×RGBA)に展開して操作していきます。

from io import BytesIO
from PIL import Image
import numpy as np
img = Image.open(BytesIO(img_bytes))
img_arr = np.asarray(img)  # img_arr.shape == (128, 128, 4)

透過画像かどうかの判定は簡単です。img_arr は RGBA=(赤, 緑, 青, 透明度) の 4 チャンネルのため、Aの最小値が 0 であれば透明です。

transparent = len(img_arr.shape) == 3 and img_arr.shape[2] == 4 and img_arr[:, :, 3].min() == 0

次に、暗い文字かどうかを判定します。この判定はいろいろ調整した結果、次のように判定することにしました。

gray = img_arr[:, :, :3].max(axis=2).astype(float)  # 簡易的なグレースケールとして、RGBの最大値を取ります
gray *= (img_arr[:, :, 3] / 255)  # 透過度を反映したグレー値を計算します
dark = np.percentile(gray[img_arr[:, :, 3] > 0], 80) <= 85  # 80%パーセンタイルが、85=255/3 の明るさ以下の場合、暗いと判定

シンプルなロジックですが、比較的精度良く闇絵文字を判定できています。

f:id:yamitzky:20191225003455p:plain

白背景の合成

最後に、闇絵文字と判定されたものに白背景を付与します。

white = Image.fromarray(np.ones(img_arr.shape, dtype=img_arr.dtype) * 255)  # 真っ白な画像を作る
white.paste(img, (0, 0), img.split()[3])  # 白背景に、元の画像を貼り付ける
white.save(f'result/{name}.png')

完成!

こうして、無事に闇の魔術に対抗することができました。安心して Slack をダークモードにして使えそうです。

f:id:yamitzky:20191225003505p:plain

完成した Python スクリプトは Gist に添付しましたので、ぜひ参考にしてください。

gist.github.com

K8sデビューしてから4日で小さな本番プロダクトをリリースした話

JX通信社Advent Calendar 2019」24日目の記事です.

昨日は@kimihiro_nさんによる, 「Immutable Python ~ NamedTuple で書く副作用のないプログラム」の話でした.

こんにちは. JX通信社でシニア・エンジニア&データ基盤エンジニアおよび野球*1をしています, @shinyorke(しんよーく)と申します.

JX通信社では, データ駆動での意思決定および施策実施をより円滑に進めるため, データ基盤の構築・運用を進めています.

その過程でいろんなFrameworkやサービスを試して...の一つの成果として以前こんな話をしました.*2

tech.jxpress.net

このエントリーではもうちょっとプラットフォーム・基盤に寄り添ったテーマで,

同僚氏に「K8sいいぞ!」と布教活動を受けた結果, 4日で本番運用までこぎつけた話

をしたいと思います.

TL;DR

  • はじめて使うものは「急がば回れ」で基礎から学ぶことが良さそう
  • kubectlが楽すぎてGCRとかの認証がうっかり気が付かなくてハマることもあるので気をつけよう.
  • 欲張らず小さい構成で試しながらやるのオススメ, いきなりガッツリ作らなくてもいい.

あらすじ - 私のK8sスキル

かなりざっくりいうと,

  • 前提としてDockerは使いこなしてる. 何があっても独力で解決できるレベル.
  • クラウドインフラとかの知識・経験も同様.
  • なお, K8sは過去2回挑戦して2回放置挫折.

といったレベルです(ちなみに本職はデータ基盤エンジニアメインのなんでも屋ですが, インフラは専門ってわけでもない*3).

なお, Dockerは自分のブログで何度かチャレンジしてそこそこヒットする記事も出してるのですが,

  • K8sはHello world的に動かしたりしたしその場では動いてるの確認してる
  • だがしかし, 用語とか混乱するしわからん
  • というのと, ローカルだったらdocker-compose, STG/PRDならシンプルにAWS ECSとかGAEで良いのでは説*4

という思いもあり, 正直K8sにチャレンジするモチベはあまりありませんでした.*5

そう, 社内で布教されるまでは.

K8sデビューしてから4日で本番リリースするまで

というわけでこっからいろんな記録(Slackから社内ブログまで)をサルベージして4日でやった記録を述べます.

やりたかったこと・やったこと

その前にやりたかったことを絵にしました.

※ブログ用に書き起こした絵図です(ので色々端折ってます)

f:id:shinyorke:20191208181114p:plain
やったことの絵

やりたいことは一言で言うと「ETL*6の"L"にあたるバッチプロセスを作りたい」でした.

  • GCS(Google Cloud Storage), 上図の「App Logs」に溜まったログファイル(hourlyでダンプされる, json.tgz)をBigQuery「Datasets」にimport
  • GCS -> BigQuery ImportをPythonのGCS + BigQueryのスクリプトとして開発*7*8
  • 上記の「オレオレBigQueryスクリプト」をhourly実行. 上図の「Backend」の部分.

と思いつき, 絵にしたら実にシンプルだったので早速やることにしました.

なお, 環境はお絵かきの通りGCPと決まっており, GCS(App Logs)とBigQuery(Datasets)はすでに存在しており, 残すBackendを

  • 普通のDocker Containerとして動かす前提で開発. hourly実行はcronとかで上手くやる
  • ちょっと調べたらK8sで行けそう?別件で作ってたWorkflowもCloud Composer*9だしせっかくだったら揃えたい

という2択になり, チームのSlackチャンネルでつぶやいた所,

K8s布教マン「それK8sのJobで行けるのでは?」

というレスが来たので, 速攻でK8sでやる場合の手順などを検証することにしました.

1日目 - K8sに慣れ親しむ

というわけで初日はK8sに入門(通算3回目)をすることにしました.

題材は, おすすめされたこちらの書籍を使いました(会社の書棚にあった).

gihyo.jp

こちらを読みながら色々お試ししていたのですが, こちらの絵を見てすべてを解釈しました.

f:id:shinyorke:20191223235459p:plain
【図】K8sと実際の環境の差(参考書籍を元に描いた)

その時自分が思わず自分のSlackに一言

K8sの仕組みがあんまわかってなかったもやもやが午前中で吹き飛んだ

ローカル環境(Macbook Pro上のDocker Desktopでクラスタ作ってやった)上での移植および, 通常のJobを用いた単体での動作確認が初日でアッサリ終わりました.

2日目・3日目

2日目・3日目は,

  • Backendの動作環境である, GKE上でクラスタを立ち上げる
  • その前に, GCRにApplicationのImageをPushする
  • STG確認して問題なかったら本番!

に注力しました.

今回は新規プロダクト・新規開発かつ, 「ログをBigQueryに放り込むバッチ」という小さいプロダクトだったのでSTG上手く行ったら即本番は既定路線でした.

が, ここで一つトラブルが.

ワイ「GCRからpullできないンゴ...」

dockerコマンドを用いてpush/pullは通るのですが, kubectl上でコマンドを叩いてもSTG(GKE)でpullに失敗してJobが起動しない...orz

自分でも調べつつ, 周りの手も借りたのですが中々いい感じに行かなかったのですが見つけたこちらのサイトに助けられました.

blog.container-solutions.com

要約すると,

  • GCR用のSecretを定義しないとあかんよ
  • 使うべきSecret指定してね!

っていうことでした.

念の為書くとこんな感じです.

GCR用のSecret作成

$ kubectl create namespace local-dev    # namespace作る
# 鍵を保管するSecretを作る
$ kubectl create secret docker-registry gcr-key \
 --docker-server=https://gcr.io \
 --docker-username=_json_key \
 --docker-email=hoge@example.com \
--docker-password="$(cat ../credentinal-hoge.json)"  # 予めcredential.jsonを作っておく

Secretの指定.

# imagePullSecretsに鍵とそのパスを指定
$ kubectl get serviceaccount default -o yaml                   
apiVersion: v1
imagePullSecrets:
- name: gcr-key
kind: ServiceAccount
metadata:
  creationTimestamp: "2019-11-12T01:12:19Z"
  name: default
  namespace: local-dev
  resourceVersion: "38601"
  selfLink: /api/v1/namespaces/local-dev/serviceaccounts/default
  uid: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
secrets:
- name: default-token-hogehoge
- name: gcr-key

Patchを当てて使えるようになったらおしまい.

$ kubectl patch serviceaccount default \    
-p '{"imagePullSecrets": [{"name": "gcr-key"}]}'
serviceaccount/default patched
$ kubectl config set-context --namespace=local-dev docker-desktop

これで無事にクラスタもでき, STG確認もうまくいきました.

また, バッチ化に伴い今までJobで動かしていたものをCronJobに移行しました(これは瞬殺で終わった).

ここでハマった知見・ノウハウを社内ブログを書いたのを含め, ここまで3日使いました.

4日目 - 本番リリース

こうしてすべての課題がクリアし, K8sデビューしてから丸4日で本番にリリースができました.

リリース自体はSecretおよびConfigの向け先を変更(環境変数を変えた), 祈るようにモニタリングしてたぐらいなので楽でした.

運用から一ヶ月ちょっと, サービスのログを1時間に一度, 割と多い量を転送していますが,

  • (社内で広く使われている)Sentryの警告が鳴らない限り放置でOK*10
  • 予防線として, Slackでの通知も入れている(流量が極端に多い・少ないときにに備えて)
  • バッチの再実行・時刻変更およびbugfixのリリースなどのときもkubectlでチョチョイと終わる

などかなり快適な感じになりました.

今後やりたいこと&結び

というわけで, 3度目の挑戦にして成功したK8s入門*11および, 4日で実戦投入の話はこれにておしまいです.

今回の勝因は色々ありますが,

  • 急がば回れ, で基礎から攻めた結果の爆速キャッチアップとリリースにつながった.のと, K8s推進派のたっち(TatchNicolas)さんが技術的にも気持ち的にも色々サポートしてくれたのに感謝.
  • 自分自身が構想していたサービスが小さくかつシンプルであり, 導入に関して障壁が少なかった*12
  • 既存プロダクトのリプレースではなく, ゼロイチでの新規立ち上げだったのでやり方についてある程度自由度があった*13

といった所が大きかったです.

とはいえいくつか課題・やりたいこともあり,

  • CronJobで動かしてるが全体最適って意味ではCloud Composer(Airflow)のWorkflow管理にしたい. K8sクラスタをAirflowにKickさせる感じ.
  • 今回のK8sバッチがオレオレBatchとして風化しないように似たようなバックエンドをK8sに寄せるのと, ちゃんとCIもやりたい...*14
  • 何よりも, マイクロサービスが多い弊社でK8sが最適な場面は多そうなのでそこにContributeしていく(もしくは周りがContributeするような楽しい感じにする)

というのが今後のアクションかなと思っていますし, そんなK8sの良さを伝えてくれたりサポートしてくれた たっち(TatchNicolas)さんの用にいい感じにK8sを使いこなせるようになりたいなと思いました, この記事とかあの記事とか.

JX通信社Advent Calendar 2019」のクローザーはつよつよエンジニア統括な@YAMITZKYさんです.

皆さん良いメリクリを.

*1:元々野球のデータを扱うお仕事をしていました&Pythonを使った野球データサイエンスは18日の記事(類似性スコア)でやりました. 余談ですがプレーする方は全くもって苦手です.

*2:PySpark MLlibはきっといつかやります(小声)

*3:ですがベアメタルからサーバー一台作って納品程度の仕事は過去にしてました.

*4:過去に経験した技術スタックって意味で

*5:前にいたチームにもK8s普及マンがいてついてやっていい感じになったこともあるけど結局仕事で使えずぐぬぬってのがありました.

*6:Extract Transform Loadの略で, 要するに「データを取得して加工して吐き出す」バッチのこと. 今回はEとTの部分は別に作っていました.

*7:やってることは公式のBigQueryおよびGCSクライアントを繋いでPython3.7でいい感じに開発したスクリプトです. 実装は割愛.

*8:今思えばembulkでも良かった説.

*9:データ基盤のWorkflowとして活用, これは作った人にブログ書いてほしいかもなお気持ち.

*10:Pythonスクリプトの内部にimportして使ってます. Sentryが何者かを知りたい方は公式サイトを御覧ください.Pythonクライアントはこれ.

*11:もう入門することはないやろってぐらい普段着的に使おうなってお気持ちはあります.

*12:これは私見ですが, K8s = 大規模構成なサービスに合ってる, と誤解してましたが小さくやるのはまあまあアリだと思いました. とはいえ使い所難しいのもあるので技術選定でK8s選ぶ時は気をつけたほうが良さげ.

*13:本題からちょっと外れますが最悪K8s使わずにGAEとかで動かしても良かったわけで.っていう逃げ道も考えてました.要するに選択肢・余白は重要です.

*14:リリース頻度が多くない, ローカルからshell叩いておしまいとはいえCIしときたいなと.

Immutable Python ~ NamedTuple で書く副作用のないプログラム

f:id:nsmr_jx:20191220091445j:plain

この記事はJX通信社 Advent Calendar 2019 の23日めの記事です。
昨日は Yosk さんの 名刺作成をデザイナーの業務から外して、効率化させた話 でした。

こんにちは、サーバーサイドエンジニアの @kimihiro_n です。 今回は Python の Immutable を最大限活用してみる話を書いてみます。


【2019/12/23 訂正 1 】 : NamedTupleを使う際、同値であればオブジェクト自体も同一であると書いてましたがこちらは誤りでした。 個々の id を調べてみると別々のオブジェクトが割り当てられていたため、記事の表現を一部修正しました。

【2019/12/24 訂正 2 】: コメントにて、Java の String オブジェクトも不変であるというご指摘をいただきました。 不変であることとオブジェクトが同一であることの性質を混同してしまっていたため、記事を修正させていただきました。


Immutable ってなに

Immutable (イミュータブル) は「不変」であることを指す英単語です。 日常では出てくることはまずないですが、プログラミングの用語としては時々耳にする名前かもしれません。

プログラミングにおいて「不変」、つまり変わらないというのはどういうことかというと、一度生成したら値を変えることができない もののことを指します。 一方生成後も変更が可能なオブジェクトのことを Mutable (ミュータブル) といいます。

Python でいうと list は可変ですが、str や tuple は不変のオブジェクトになっています。 比較しやすいので list と tuple で比べてみてみましょう。

list

>>> l = [1, 2, 3, 4, 5]  # list

>>> l[1]  # index を指定して読み取り
2
>>> l[:3]  # 範囲を指定して取り出し
[1, 2, 3]
>>> len(l)  # 長さを取得
3

tuple

>>> t = (1, 2, 3, 4, 5)  # tuple

>>> t[1]  # index を指定して読み取り
2
>>> t[:3]  # 範囲を指定して取り出し
[1, 2, 3]
>>> len(t)  # 長さを取得
3

list と tuple、どちらもシーケンス型とよばれるタイプの組み込み型なので、同じような読み出し操作を行うことができます。list や tuple の中身を変更していないので当然といえば当然ですね。

ではこれら2つの変更操作をそれぞれみてみましょう。

list

# 
>>> l[0] = 6   # index を指定して書き換え
>>> l
[6, 2, 3, 4, 5]
>>> l.append(7)  # 末尾に追加
>>> l
[6, 2, 3, 4, 5, 7]
>>> l = [8, 9, 10]  # 変数の使い回し
>>> l
[8, 9, 10]

tuple

>>> t[0] = 6  # index を指定して書き換え
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: 'tuple' object does not support item assignment

>>> t.append(6)  # 末尾に追加
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'tuple' object has no attribute 'append'

>>> t = [8, 9, 10]  # 変数の使い回し
>>> t
[8, 9, 10]

今度は tuple 側がエラーになってしまいました。特定の index に代入しようとすると TypeError が出てます。また末尾に追加する append というメソッドは tuple では用意されていません。

このように 一度生成したら変更する手段がない のが Immutable なオブジェクトの特性となります。 なお3つめの例の変数の使い回しについては成功してます。不変なのはあくまで tuple の実体に対してだけなので、実体を指し示す変数については制約がかかりません。

なぜ Immutable をつかうのか

一見不便に見える Immutable ですが、Immutableだからこその利点がいくつかあります。

同値比較ができる

Immutable であるということは、同じ値を持っていれば同じもののようにふるまうことを示すことができます。 中身が変化しないのでいつどこで参照しても同一のものとして扱うことができます。

代表的な例でいうと文字列ですね。Python では文字列も Immutable なオブジェクトとなっています。 なので

>>> "hoge" == "hoge"
True

というのが常に成立します。

Python を使っているユーザーにとっては当たり前と思うかもしれませんが、Java のような文字列が可変として扱われる言語だと、こういった等式による比較が常に成立するとは限りません。文字列に同値であるかを判定するメソッド equals が用意されており、こちらを使う必要があります。 (Java においても文字列は不変オブジェクトとのことでした。同値であってもオブジェクトが異なるため等式による比較が常に成立するわけではないようです。)

Python でもクラスは可変なオブジェクトになります。 なので

>>> class Hoge:
...     def __init__(self):
...         self.fuga = 'fuga'
...
>>> h1 = Hoge()
>>> h2 = Hoge()
>>> h1 == h2
False

のようにインスタンスが同じ値を持っていたとしても、等式による比較は失敗してしまいます。

ところが後述する NamedTuple というものを使うと、値が一致していれば等しいと見なすオブジェクトを作成することが可能です。

>>> from typing import NamedTuple
>>> class Hoge(NamedTuple):
...     fuga: str = 'fuga'
...
>>> h1 = Hoge()
>>> h2 = Hoge()
>>> h1 == h2
True

NamedTuple は tuple の一種で、クラス初期化のタイミングでしか値をセットすることが出来ません。

Immutable なオブジェクトの場合、中身の値が一致していればオブジェクトも同値であることを保証できます。 (__eq__ というメソッドをオーバーライドすることで普通のクラスでも同値による比較が可能になりますが、本筋ではないので割愛します)

変化を気にしなくていい

Immutable 最大のメリットとなるのがこの性質です。一度生成したオブジェクトの変化を気にしなくていいので安全なプログラミングを書くことが可能になります。

def print_last(l: list):
    """ Listの一番後ろを表示する関数 """
    last = l.pop()
    print(last)

l = [1, 2, 3]
print_last(l)  # => 3

print(l)  # => [1, 2] 
# l 自体が変化してしまっている

(いい例が思いつかなくて微妙なサンプルになってしまってますが、)プログラムを書いていて上記みたいなパターンではまったことはないでしょうか。 print_last という画面表示を行うための関数を作っていますが、この関数は値を読み取るだけではなく、渡した配列を変化させてしまっています。 そのため呼び出し前と呼び出し後で予期せず配列の中身が変わってしまっているわけですね。 こういった関数で多かれ少なかれ苦い経験をしたことはあるのではないでしょうか。

受け取った引数を変更する関数は破壊的メソッドと呼ばれており、取り扱う際は注意して使う必要があります。 (この例でいうと pop というリスト用のメソッドが破壊的な操作になっています。)

こういった副作用の問題は リストが変更可能であるからこそ起こっている ともいえます。 変更不可能な Immutable オブジェクトを引数に利用していれば、呼び出し元を変化させるような手段が用意されていないため安全に利用することができます。

def print_last(t : tuple):
    # t.pop() のような破壊的メソッドはそもそも存在しない
    t = t[-1]   # また、仮に t に再代入しても呼び出し元は変化しないので安全  (これはリストでも同じ)
    print(t)

t = (1, 2, 3)
print_last(t)  # => 3

print(t)  # => (1, 2, 3) 
# t は変化していない

ハッシュ化が可能

同値比較ができるということに起因するのですが、Immutable なオブジェクトであればハッシュ化して扱うことができます。 ハッシュ化とは何か、説明するよりコード見てもらった方が早いかもしれません。

>>> l = [1, 2, 3]  # list
>>> t = (1, 2, 3)  # tuple

>>> d = {}

>>> d[l] = "これは list です"
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: unhashable type: 'list'

>>> d[t] = "これは tuple です"
>>> d
{(1, 2, 3): 'tuple'}

見慣れてないと不思議に思うかもしれませんが、Immutable であるタプルは辞書(dict)のキーとして利用することが出来ます。 可変である List のほうは dict のキーとして使おうとすると TypeError が発生してますね。 定義後に値が変わらないので、内部的にオブジェクトのハッシュ値を割り当てれば、その値から一意のオブジェクトが復元できるため、こういったことが可能になります。 ちょうど git のコミットハッシュから一意の変更が取り出せるのも似た理屈です。

これで何がうれしいかというと、オブジェクトのまま重複チェックなどが利用できるようになります。 クラスのインスタンスのまま重複チェックをしようとすると、重複確認用のキーを生成して比較することになりますが、Immutable なオブジェクトであれば同値のものをそのまま重複と見なして扱うことが出来ます。

Hello Immutable Python

Immutable について紹介したのでより具体的に Python で扱っていく方法を紹介していきます。

Python で使える Immutable 要素

Python で主に使える Immutable な要素は以下になります

  • 真偽値(bool)
  • 数値(int, float, Decimal)
  • 文字列(str)
  • タプル(tuple, NamedTuple)

あとは bytes だったり、Frozen(=変更不可な)属性を持った特別なオブジェクトfrozenset, Dataclass(frozen=True) なんかもあります。

これらを活用していくことで Immutable で副作用のないプログラムを書いていくことが可能です。 なお、tuple や NamedTuple を使う場合でも中に入れるオブジェクトを Mutable にしてしまうと、完全な Immutable とはいえないので注意が必要です。

具体例をみていきましょう。

listを捨ててtupleへ

list は mutable なので immutable な tuple へと置き換えます。

mutable = [1, 2, 3]
immutable = (1, 2, 3)

シンタックスも似てるので簡単ですね。immutable = tuple(mutable) のように tuple 関数を利用して変換することも可能です。 ちなみに要素が一つしかない tuple は (1) ではなく (1, ) と書かなくてはならない罠があります。

要素追加

immutable = immutable + (4, 5)

immutable なタプルには要素を追加するためのメソッドがありません。 なので、要素を追加したいときには新しいタプルを生成してあげる必要があります。 繰り返しになりますが、再代入は Immutable の制約ではないので問題ありません。

Typeヒント

from typing import Tuple

def hoge(t: Tuple[int, ...]):
     ...

Tuple の型ヒントを書いてあげるときはこんな感じになります。List の場合 List[int] と書けば整数のリストであることを表現できますが、Tuple の場合は , ... という表記で同じ型が続くことを明記してあげる必要があります。

dict を捨てて NamedTuple へ

dict もmutable なのでこちらも immutable にしたいですね。frozendict のようなものは標準パッケージに存在しないので、Immutable に辞書を扱うには別のアプローチをとる必要があります。

個人的におすすめしたいのが NamedTuple です。

NamedTuple とは

Python には NamedTuple と呼ばれる、順序でなく名前でアクセスできる tuple が用意されています。 これだけだと使い勝手そんなによくないのですが、Python 3.6.1 から typing モジュールにクラス風にかける NamedTuple が追加されました。 こちらを使うと簡単に構造化した Tuple を生成することが出来ます。 また型ヒントもしっかり活用することができるのでより安全なプログラムを意識して作れるようになります。

from typing import NamedTuple

class Hoge(NamedTuple):
    fuga: str
    piyo: int

h1 = Hoge(fuga='fuga', piyo=1)
h2 = Hoge(fuga='fugafuga', piyo=2)

見た目は普通のクラスに近いですね。実際、クラスと同じようにメソッドを生やしたりできます。 NamedTuple を継承してつくることと、init を使わずクラス変数のように型付きのフィールドを列挙してあげれば作成が出来ます。

>>> h1.fuga  # 値の取得
'fuga'

>>> h1.fuga = 3  # 値の変更
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: cant set attribute

>>> h1._replace(fuga=3)  # 値を変更した新しい tuple をつくる
Hoge(fuga=3, piyo=1)

tuple の要素にドット付きでアクセスすることができますが、値を変更しようとすると Immutable なので怒られます。一部の要素を変えた tuple を手に入れたい時は _replace というメソッドが定義されています。こちらを呼ぶとと元の tuple を破壊せず新しい tuple を生成出来ます。

from typing import Tuple, NamedTuple

class User(NamedTuple):
    name: str

class Group(NamedTuple):
    users: Tuple[User, ...]

u1 = User(name='taro')
u2 = User(name='hanako')
g = Group(users=(u1, u2))
# => Group(users=(User(name='taro'), User(name='hanako')))

構造化されたデータもこの通り。

動的にキーが変わるようなものは dict でないと対応が厳しいですが、API のレスポンスなど事前に形式が決まっているものであれば NamedTuple を活用して immutable なコードを書くことが出来ます。

実践 Immutable

Immutable に書くためのパーツがそろったので、実際に Immutable でどんなコードが書けるのかを紹介したいと思います。

本当はもっと Immutable が生かせるお題見つけたかったのですが、全然思いつかなかったので Qiita の新着アイテム API を叩いて、タイトルの長い順にソートして出力する というようなことをやってみます。

Qiita の API についてはこちら。 このエンドポイントは認証なしで叩くことができます。

NamedTupleを定義する

from typing import NamedTuple

class User(NamedTuple):
    id: str

    @classmethod
    def from_dict(cls, d: dict) -> 'User':
        return cls(id=d['id'])


class Item(NamedTuple):
    title: str
    url: str
    user: User
    created_at: str

    @classmethod
    def from_dict(cls, d: dict) -> 'Item':
        return cls(
            title=d['title'],
            url=d['url'],
            user=User.from_dict(d['user']),
            created_at=d['created_at']
        )

Qiita のレスポンスに従って、使う要素をまず NamedTuple で構造化します。 扱いやすいように from_dict() というクラスメソッドを各 NamedTuple に定義してあげて、dict から NamedTuple に変換できるようにしています。 from_dict 自体は NamedTuple の機能でもなんでもないので、好きなやり方で作ってもらって大丈夫です。

あれ、dict つかうの?と思うかもしれませんが、Python の場合、json.loads() の結果が dict で返ってくるので、NamedTuple に変換するまでは mutable な dict で受け渡しする必要があります。

API を叩いて Immutable なオブジェクトを手に入れる

from typing import Item

import requests

QIITA_API_ENDPOINT = 'https://qiita.com/api/v2/items'


def fetch_qiita_items(page: int = 1) -> Tuple[Item, ...]:
    res = requests.get(QIITA_API_ENDPOINT, {'page': page})
    res.raise_for_status()
    return tuple(
        Item.from_dict(item) for item in res.json()
    )

requests という外部ライブラリを使って JSON をとってきます。 とってきた JSON から先ほどの Item.from_dict() に1つづつ渡してオブジェクト化していきます。 最後に Item の配列(厳密には generator)を tuple に変換してあげれば Immutable な Qiita 記事一覧の完成です。

文字数順にソートして出力する

ここまでくればあとは煮るなり焼くなりするだけですね。 PyCharm などの IDE を使っていれば型ヒントによる補完機能が効くのでいちいち JSON の構造を覚えてなくてもサクサクコーディングできるようになります。

def sort_by_title_length(items: Tuple[Item]) -> Tuple[Item, ...]:
    """ タイトルの長さでソートした Item の配列を返す """
    return tuple(
        sorted(items, key=lambda i: len(i.title), reverse=True)
    )


def format_for_display(item: Item) -> str:
    """ 出力用に整形する """
    return f'''{item.title}
        URL: {item.url}
        Author: {item.user.id}
        CreatedAt: {item.created_at}'''


def main():
    items = fetch_qiita_items()
    items = sort_by_title_length(items)
    for item in items[:5]:
        print(format_for_display(item))

sort_by_title_length は tuple を並び替える関数ですが、引数自身を並び替えるのではなく、並び替え済みの新しい tuple を生成して返しているので副作用がなく安全です。 なお、Python の sorted を通すと Generator 型になってしまうので明示的に tuple に変換しています。

これでプログラムが完成したので実行してみましょう。

f:id:nsmr_jx:20191218190627p:plain

ちゃんと表示されてますね。お疲れさまでした。

全体のソースはこちらに置いておきます。

Immutable を意識して書く Python はいかがでしたか。 値の変更が発生しないことで、バグが起こりにくい安全なシステムを作ることが出来ます。

おまけ

実際 Immutable ってどうなの

イントロダクションの記事なので意図して Immutable 化して使ってみましたが、実際のコーディングでここまで頑張って Immutable 化する必要はないと思ってます。 実際 Python の関数通すと簡単に dict とか generator に戻ってしまうので、都度 Immutable 化するのは結構手間です。 関数内で可変な引数をいじらない くらいのルールに留めておくのが現実的なラインではないでしょうか。 Immutable なオブジェクトを利用すると言語レベルで制約をかけれますが、可変なオブジェクトでも扱いを気をつけるだけで安全性が変わってきます。

どちらかというと NamedTuple で構造化したり、関数に型ヒントを記述することで得られるメリットのほうが大きいです。 最近のエディタは賢いので構造化されたデータや型ヒントを活用して安全かつ楽にコーディングが出来るようになります。 型ヒント書きましょう!

NamedTuple VS DataClass

好みの問題で NamedTuple を紹介しましたが、Python3.7 からは Dataclass というちょっと便利なクラスを使うことが出来ます。オプションで frozen=True とすれば変更不可になるのでこちらも Immutable なオブジェクトとして扱うことができます。

便利なメソッドが用意されていたり、若干読み取り速度が速かったりするらしいので Python3.7 以降を使うのであれば積極的に使うのがよさそうです。

ただ Immutable として使うには frozen=True が冗長だったりするので自分は NamedTuple 派です。Mutable として値を書き換えて使うときは Dataclass を使わないてはないでしょう。

from_dict みたいにいちいち変換するの面倒くさい

そんなこともあろうかと、dict から NamedTuple にマッピングするライブラリを作ってみました。 階層化した NamedTuple も一発で変換できます。 (プロダクションに投入した実績はないのでご利用の際は自己責任でお願いします。)

本文中のスクリプト

説明で使ったスクリプトの ipynb 張っておきます。


明日のアドベントカレンダー@shinyorke さんの K8s デビューまわりのお話です。

情報共有を効率化 - Slackの分報チャンネルにあるリンクを拾うSlack Botを作ってみた

f:id:jxpress:20191221002017j:plain

こちらはJX通信社 Advent Calendar 2019 の21日目の記事です。どうも、何でも屋のkainです。

きっかけ

昨日の朝、弊社メンバーからこんな投稿がありました。

メンバーの投稿

弊社の分報チャンネルでは、今やってることや困ってることなど以外に、それぞれが読んだリンクを雑に投稿しています。中には有意義な情報も多く、それぞれの分報チャンネルでキュレーションメディアを運営している、といった様子です。

一応、チームや分野ごとの記事共有用のチャンネルもありますが、分報チャンネルに投稿した後に、いちいち投稿するのが手間というのもあり、利用機会は限られています。なので、メンバーはSlackのチャンネルを徘徊して、情報収集を行う必要がありました

そこで、「ここを見ればおk」とったチャンネル(=情報収集チャンネル)作れないかと思い、アドベントカレンダーのネタに困っていたのでサクッと作ってみました。

どうやるか

Slack Appには、Slack内で発生したイベント情報を購読するため機能(Event Subscription)があります。開発者は事前に設定したRequest URLに対して、Slack上で特定の操作(メッセージ投稿、リアクションなど)が行われるたび、イベント情報が送られてきます。

今回欲しいのは、「特定のチャンネル(分報チャンネル)でリンクが含まれたメッセージが投稿された」というイベント情報です。そこで、Event Subscription機能のうち、Bot Eventの message.channels (Botユーザーが参加しているチャンネルの投稿)を購読し、こちら側でリンクが含まれているかどうかを選別 → 情報収集チャンネルへ投稿するようにします。

なので、運用は「チャンネルにBotユーザーを招待」すれば、勝手にリンクを拾ってくれて、情報共有チャンネルに投稿してくれます。

作ってみる

まずは、Slack Appをポチポチして作ります。こちらから、「Create App」をクリックし「アプリ名」を決め「ワークスペース」を選択して完了です。そしたら、Basic Information設定画面にあるApp CredentialsのSigning Secretと、OAuth & Permissions設定画面のBot User OAuth Access Tokenをメモしておいてください。

次に、投稿を受け取る処理を書いていきます。弊社はPythonユーザーが多いのでサクッと検証する際にはPythonを使うことが多いです。下記スクリプトを利用する際は、 {{ 情報収集チャンネル名 }} を適宜書き換えてください。

# app.py
import os

from flask import Flask, request, jsonify
from slack import WebClient
from slackeventsapi import SlackEventAdapter

SLACK_BOT_TOKEN = os.environ['SLACK_BOT_TOKEN']  # メモした
SLACK_SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"]
SLACK_CHANNEL = "{{ 情報収集チャンネル名 }}"

app = Flask(__name__)
slack_client = WebClient(token=SLACK_BOT_TOKEN)
slack_events_adapter = SlackEventAdapter(SLACK_SIGNING_SECRET, "/", app)


@slack_events_adapter.on('message')
def handle_message_channels_event(data):
    event = data['event']

    # リンクを抽出し投稿
    for block in event.get('blocks') or []:
        for section in block['elements']:
            for i in section['elements']:
                if i['type'] != 'link':
                    continue

                slack_client.chat_postMessage(
                    channel=SLACK_CHANNEL, text=i['url'], unfurl_links=True
                )


if __name__ == "__main__":
    app.run(debug=True, host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))

Slackの公式ライブラリのpython-slack-events-apiは、処理をイベント種類ごとに書くことができて大変便利です。

動かしてみる

こちらのスクリプトを動かしてみます。

$ pip install flask slackclient slackeventsapi
$ export SLACK_BOT_TOKEN={{ Bot User OAuth Access Token }}
$ export SLACK_SIGNING_SECRET={{ App CredentialsのSigning Secret }}
$ FLASK_APP=app.py FLASK_DEBUG=1 flask run

このままではSlackからのイベントを受け取れないので、ngrokを使ってインターネット空間にプロキシします。

$ ngrok http 5000

表示されたForwarding URLのうちhttpsのものをメモしておき、Slack AppのEvent Subscriptions設定画面のRequest URL欄に入力します。

これで準備完了です。Botユーザーを分報チャンネルに招待してから、リンクを投稿してみます。

f:id:jxpress:20191221003227j:plain

すると、情報収集チャンネルでもリンクが共有されます。

最後に

メンバーの投稿を見たのがたまたま東京から九州方面に向かう飛行機の搭乗前で、目的地に着くまでの間に完成しました。Slack Appはとても簡単にパワフルな機能を実装できるので、ぜひ業務に活用してみてください。

Python Clickユニットテスト・レシピ集 - CLIではじめるテスト駆動開発(その1)

JX通信社Advent Calendar 2019」20日目の記事です。 こんにちは。2019年9月からJX通信社のエンジニアとなった鈴木(泰)です。好きな食べ物はオムライスです。

本日は、Python Clickユニットテスト・レシピ集 - CLIではじめるテスト駆動開発(その1)と題して、CLIのユニットテストのスニペットを書いてみたいと思います! "その1"とした理由は、アドカレに間に合わな(小声)・・・じゃなかった・・・この記事だけで全ての備忘録を列挙すると長くなりすぎてしまい、記事が読み難くなると判断したからです。

今後も引き続き少しづつ備忘録を紹介していければと思います。

はじめに

私はCLIをよく書きます。その理由は、バックエンドシステムの運用業務に携わっていることにあります。運用業務では様々な場面でCLIを作成します。私の場合、運用業務における手作業を自動化するため、バッチ処理を書くため、というのが主です。

にも関わらず、私はCLIのユニットテストを、これまであまり書いてきませんでした。これについてはとても反省しています。ごめんなさい!

書かなかった理由(a.k.a 言い訳)は多々あります。

  • 書き捨てスクリプトだから。
  • 外部コンポネントと密結合しているので、ユニットテストが書き難い。
  • テストを書く時間がない。忙しい。
  • テストを書くことは、オーバーワーク。
  • etc ...

ざっくりとひとまとめにすると、要は「書いている時間がない」のです。

書いている時間がない場合どうするか?私がよく使う手はコピー・ペーストです。つまり、CLIを書くときに頻出するパターンを想定した、コピーペースト用のスニペットを用意しておけば良いのです。今回は、CLIを書くときに頻出するパターン毎に、コピーペースト用のスニペットを書いてみることにしました。

私の知らないもっと良い方法を知っている方がいましたら、コメント等でご教示いただけると幸いです!

目次

対象とするCLI

対象とするのは、外部コンポネントとの結合がないCLIのユニットテストです。

外部コンポネントとの結合とは、ファイルシステムやネットワーク通信を介してシステム繋がりのことです。例えば、WebAPIから何らかのデータを取得し、結果をローカルファイルに保存するCLIは、外部コンポネントとの結合があるCLIです。なぜなら、WebAPIとローカルファイルシステムという、2つのコンポネントとつながっているからです。

click

clickというライブラリを用いてCLIを書きます。clickを用いる理由は

といったことがあります。

諸注意

  • 本記事で掲載するソースコードは、MacOS 10.15.2、Python3.7、こちらのclickおよびその依存パッケージのバージョンで動作確認をしております。
  • 私がよく使うスニペットを前提に書いています。私があまり使わないなぁ・・・と思ったら、そのスニペットは書いていません。(・・・が、こんなスニペットも使える!とか、これは必要だろ!とかあればコメントお願いします!)
  • 本記事で使ったソースコードはこちらにあります。

スニペット一覧

外部コンポネントとの結合がない場合

標準出力、標準エラー出力、終了ステータスコードの検証

まずは、必ず検証すべき3つの値(標準出力、標準エラー出力、終了ステータスコード)からです。

CLIのソースコード 全体

import click

@click.command()
def cli():
    click.echo('こんにちは')
    click.echo('世界!', err=True)  # 標準エラー出力
    exit(100)

テストコード 全体

from click.testing import CliRunner
from cli_stdout import cli

def test_cli_output():
    result = CliRunner().invoke(cli)
    assert result.output == 'こんにちは\n世界!\n'

def test_cli_output_separatly():
    # 標準出力と標準エラー出力を分離して出力をテストしたい場合、
    # `CliRunner(mix_stderr=False)`とする。
    result = CliRunner(mix_stderr=False).invoke(cli)
    assert result.stdout == 'こんにちは\n'
    assert result.stderr == '世界!\n'

def test_cli_exit_code():
    result = CliRunner().invoke(cli)
    assert result.exit_code == 100
  • clickを使用する場合click.echo関数を用いて出力することが慣習ですが、print関数で出力した場合でも、上のテストは動きます。

色付き出力の検証

clickではANSI Color codeを用いて、出力される文字列に色を付与できます。

f:id:taisuzuk:20191219114904p:plain

CLIのソースコード 全体

import click

@click.command()
def cli():
    click.echo('こんにちは')
    click.echo(click.style('JX', fg='green'), nl=False)
    click.echo(click.style('通信社!', fg='red'))

テストコード 全体

from click.testing import CliRunner
from cli_color_output import cli

def test_cli_output():
    result = CliRunner().invoke(cli)
    assert result.output == 'こんにちは\nJX通信社!\n'
  • ユニットテストで検証するときは、ANSI Color codeを無視できます。
  • ANSI Color codeを含めた出力の検証をしたい場合、CliRunner(color=True)とすればできます。

例外の検証

例外を投げるCLIの検証です。

CLIのソースコード 全体

import click

@click.command()
def cli():
    raise Exception('Hello world!')

テストコード 全体

from click.testing import CliRunner
from cli_exception import cli

def test_cli_exception():
    result = CliRunner().invoke(cli)
    assert 'Hello world!' == str(result.exception)
    assert Exception == type(result.exception)
  • 例外がない場合、result.exceptionNoneです。

コマンドライン引数やオプションの検証

clickでは、コマンドライン引数やオプション指定を間違えていた場合、デフォルトでは終了ステータス2(より厳密には、staticな変数click.exceptions.UsageError.exit_codeの値)で終了します。

私としては、CLIの引数の検証(例えば、このオプションが必須で・・・この引数は文字列で・・・といったような仕様の検証)は不要であり、終了ステータスだけ検証すれば良いと思います。なぜなら、CLIの引数周りの処理はclickライブラリが担う責務であり、利用者である私たちが検証すべきことではないからです。

CLIのソースコード 全体

import click

@click.command()
@click.argument('src', nargs=-1)
@click.argument('dst', nargs=1)
def cli(src, dst):
    click.echo(src)
    click.echo(dst)

テストコード 全体

from click.testing import CliRunner
from cli_option import cli

def test_cli_option_usage_exception():
    result = CliRunner().invoke(cli, args=[])
    assert 2 == result.exit_code

サブコマンドのテストの書き方

サブコマンドのテストは次のように書きます。

CLIのソースコード 全体

import click

@click.group()
def cli():
    pass

@cli.command()
def sub1():
    click.echo('sub command1')

@cli.command()
def sub2():
    click.echo('sub command2')

テストコード 全体

from click.testing import CliRunner
from cli_sub import

def test_cli_sub1():
    result = CliRunner().invoke(cli, args=['sub1'])
    assert 'sub command1\n' == result.output

def test_cli_sub2():
    result = CliRunner().invoke(cli, args=['sub2'])
    assert 'sub command2\n' == result.output

環境変数の検証

clickでは、CliRunnerのenvという引数に、テストで使用する環境変数を指定できます。

CLIのソースコード 全体

import click

@click.command()
@click.option('--name', type=str, envvar='NAME')
@click.option('--age', type=int)
def cli(name: str, age: int):
    click.echo('{} {}'.format(name, age))

テストコード 全体

from click.testing import CliRunner
from cli_env import 

def test_cli():
    result = CliRunner(env={'NAME': 'hoge'}).invoke(cli, args=['--age=1'])
    assert 'hoge 1\n' == result.output

標準入力

clickには、標準入力からデータを読むためのget_text_stream関数があるのですが、この関数を使った場合のユニットテストの書き方はわかりませんでした。。。じゃあこの記事に掲載するなよ!と思われるかもしれませんが、もしかしたら誰か良い方法を知っているかもしれないということで・・・記事に掲載することとしました。

CLIのソースコード 全体

import click

@click.command()
def cli():
    body: str = click.get_text_stream('stdin', encoding='utf-8').read()
    click.echo(body)

テストコード 全体

# どう書いたら良いかわかりませんでした・・・。

標準入力2

標準入力を読み込むために、clickのget_text_stream関数ではなくinput関数を使用した場合であれば、次のようにしてユニットテストを書くことができます。

CLIのソースコード 全体

import click

@click.command()
def cli():
    body: str = input()
    click.echo(body)

テストコード 全体

from click.testing import CliRunner
from unittest.mock import patch
from cli_stdin2 import cli

def test_cli():
    with patch('builtins.input', return_value='hoge'):
        result = CliRunner().invoke(cli, args=[])
        assert 'hoge\n' == result.output

所感

今回は、外部コンポネントとの結合がない場合のみなので、あまり迷うことはありませんでした。その理由はclick.testing.CliRunnerが用意されていることにあると思います。これのおかげで「こういう場合はこう書けば良い」という方針が明確になっています。

次回は、外部コンポネントとの結合がある場合のスニペットについて書く予定です。特に、ファイルシステムに読み書きする、HTTP通信する、MySQLに接続する、ロガーを通してアプリケーションログを残す、については頻出パターンなので、スニペットを用意したいと考えています。外部コンポネントとの結合部をどうモックするか?スニペットし易い簡潔なモックをどう書くか?というところが鍵となりそうだと考えています。

それでは次回もご期待ください! ありがとうございました。