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に接続する、ロガーを通してアプリケーションログを残す、については頻出パターンなので、スニペットを用意したいと考えています。外部コンポネントとの結合部をどうモックするか?スニペットし易い簡潔なモックをどう書くか?というところが鍵となりそうだと考えています。

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

DartでCLIツールを作ろう

この記事はJX通信社アドベントカレンダーの19日目です。

sakebookです。最近はServer Side Kotlinをやってますが、Flutterも少し触ってます。

去年はKotlin で CLI のネタを書いたので今年はそれのDart版を書こうと思います。

tech.jxpress.net

全体の流れ

  • Dartとプロジェクトのセットアップ
  • CLIでの動作確認
  • GitHub Actionsで配布

Dartのインストール

Flutterを使っていればbundleでインストールされていますが、standalone版が必要(後述)なのでHomebrewでいれます。

$ brew tap dart-lang/dart
$ brew install dart

筆者の環境は次の通りです

$ dart --version
Dart VM version: 2.7.0 (Fri Dec 6 16:26:51 2019 +0100) on "macos_x64"

CLIツールのテンプレート作成

Stagehandでプロジェクトテンプレートを作成します(関係ないですが画像が好み)。

まずはStagehandを有効化します。

$ pub global activate stagehand

Stagehandにはいくつかのテンプレートが用意されていますが、今回はCLIツールを作りたいので console-full を選択します。

$ mkdir dart-cli-sample
$ cd dart-cli-sample
$ stagehand console-full

作成したテンプレートは次のような構成になっています。pub packageに則した構成になっています。

.
├── CHANGELOG.md
├── README.md
├── analysis_options.yaml
├── bin
│   └── main.dart
├── lib
│   └── dart_cli_sample.dart
├── pubspec.yaml
└── test
    └── dart_cli_sample_test.dart

dartファイルを見ていきます。

  • bin/main.dart
import 'package:dart_cli_sample/dart_cli_sample.dart' as dart_cli_sample;

void main(List<String> arguments) {
  print('Hello world: ${dart_cli_sample.calculate()}!');
}
  • lib/dart_cli_sample.dart
int calculate() {
  return 6 * 7;
}

mainはbinにあり、実装はlibに置くような構成になっています。

テストも作成されます。

  • test/dart_cli_sample_test.dart
import 'package:dart_cli_sample/dart_cli_sample.dart';
import 'package:test/test.dart';

void main() {
  test('calculate', () {
    expect(calculate(), 42);
  });
}

動かしてみる

初回は依存ライブラリのDLが必要です。

$ pub get
$ dart bin/main.dart
Hello world: 42!

この状態だと、Dartコードを実行しただけです。

Dartコードを変換してネイティブコードにします。

ネイティブコードの作成

dart2native コマンドを実行して作成します。このコマンドはFlutterにbundleされているDartには含まれていません。

$ dart2native bin/main.dart -o main
Generated: /YOUR_PATH/dart-cli-sample/main

作成した main を実行してみます。

$ ./main 
Hello world: 42!

無事実行できました。

制約

どこでも実行可能なものができたと思いきや、現状はホストOS用のネイティブコードしかコンパイルされません。なので、macOS, Windows, Linuxとそれぞれでコンパイルしないとダメです。

現状の制約は辛いですが、その辛さを和らげる方法があります。

GitHub Actions

GitHub Actionsでは、マトリクスビルドをサポートしています。これを利用して、OSごとに実行してそれぞれに対応したネイティブコードを作成します。

コードの修正

各OSで動かしていることがわかるように、コードを変更します。

  • bin/main.dart
import 'dart:io';

import 'package:dart_cli_sample/dart_cli_sample.dart' as dart_cli_sample;

void main(List<String> arguments) {
  stdout.writeln('Hello ${dart_cli_sample.system()}!');
  exitCode = 0;
}
  • lib/dart_cli_sample.dart
import 'dart:io';

String system() {
  return Platform.operatingSystem;
}

動作環境のOS名を返すコードです。

Actionを定義

.github/workflows/ にyamlファイルを置きます。

先に完成したものを貼っておきます。

name: Cross compile
on: [push]

jobs:
  build:
    name: Compile
    runs-on: ${{ matrix.os }}
    strategy:
      fail-fast: false
      # https://help.github.com/ja/actions/automating-your-workflow-with-github-actions/virtual-environments-for-github-hosted-runners#supported-runners-and-hardware-resources
      matrix:
        os: [windows-latest, ubuntu-latest, macos-latest]
        include:
          - os: windows-latest
            file-name: windows.exe
          - os: ubuntu-latest
            file-name: ubuntu
          - os: macos-latest
            file-name: macos
    steps:
      - name: Checkout
        uses: actions/checkout@v1
        # https://dart.dev/get-dart
      - name: Install Dart(windows)
        if: matrix.os == 'windows-latest'
        run: |
          choco install dart-sdk
      - name: Install Dart(ubuntu)
        if: matrix.os == 'ubuntu-latest'
        run: |
          sudo apt-get update
          sudo apt-get install apt-transport-https
          sudo sh -c 'wget -qO- https://dl-ssl.google.com/linux/linux_signing_key.pub | apt-key add -'
          sudo sh -c 'wget -qO- https://storage.googleapis.com/download.dartlang.org/linux/debian/dart_stable.list > /etc/apt/sources.list.d/dart_stable.list'
          sudo apt-get update
          sudo apt-get install dart
      - name: Install Dart(macos)
        if: matrix.os == 'macos-latest'
        run: |
          brew tap dart-lang/dart
          brew install dart
      - name: Build(windows)
        if: matrix.os == 'windows-latest'
        run: |
          $env:ChocolateyInstall = Convert-Path "$((Get-Command choco).Path)\..\.."
          Import-Module "$env:ChocolateyInstall\helpers\chocolateyProfile.psm1"
          refreshenv
          pub get
          dart2native bin/main.dart -o bin/${{ matrix.file-name }}
      - name: Build(ubuntu)
        if: matrix.os == 'ubuntu-latest'
        run: |
          echo 'export PATH="$PATH:/usr/lib/dart/bin"' >> ~/.profile
          source ~/.profile
          pub get
          dart2native bin/main.dart -o bin/${{ matrix.file-name }}
      - name: Build(macos)
        if: matrix.os == 'macos-latest'
        run: |
          pub get
          dart2native bin/main.dart -o bin/${{ matrix.file-name }}
      - name: Upload artifact
        uses: actions/upload-artifact@v1
        with:
          name: bin
          path: bin
  execute:
    name: Run artifact
    needs: build
    runs-on: ${{ matrix.os }}
    strategy:
      fail-fast: false
      matrix:
        os: [windows-latest, ubuntu-latest, macos-latest]
        include:
          - os: windows-latest
            file-name: windows.exe
          - os: ubuntu-latest
            file-name: ubuntu
          - os: macos-latest
            file-name: macos
    steps:
      - name: Download artifact
        uses: actions/download-artifact@v1
        with:
          name: bin
      - name: Run(windows)
        if: matrix.os == 'windows-latest'
        run: |
          cd bin
          .\${{ matrix.file-name }}
      - name: Run(ubuntu)
        if: matrix.os == 'ubuntu-latest'
        run: |
          cd bin
          chmod 755 ${{ matrix.file-name }}
          ./${{ matrix.file-name }}
      - name: Run(macos)
        if: matrix.os == 'macos-latest'
        run: |
          cd bin
          chmod 755 ${{ matrix.file-name }}
          ./${{ matrix.file-name }}

Dartをインストール

ホストOSで実行させたいので、JavaScriptアクションを使います。MarketPlaceにDartのJavaScriptアクションが見当たらなかったので、愚直にDartをインストールしました。

コンパイル

インストールしたDartへのPATHを通してコンパイルします。同一フォルダに出力してartifactとしたかったので、ファイル名をincludeでそれぞれ定義しています。

実行

それぞれのOSで生成したネイティブコードをそれぞれのOSで実行してみた結果です。

  • windows

f:id:sakebook:20191218192514p:plain

  • ubuntu

f:id:sakebook:20191218192541p:plain

  • macos

f:id:sakebook:20191218192602p:plain

しっかりOS名が出力されています。

手元でそれぞれ動かしてみたい方はこちらからartifactをDLできます。

各OSでしか実行できないことがわかると思います。

まとめ

テンプレート作成だったり、今回は触れませんでしたが引数をパースするライブラリも提供されており、DartでCLIツールは作りやすいです。

まだ制約もありますが、今回の様なやり方で制約を緩和することができます。

今回動作確認したリポジトリはこちらです。

github.com

参考

Get the Dart SDK | Dart

GitHub - dart-lang/stagehand: Dart project generator - web apps, console apps, servers, and more.

Write command-line apps | Dart

dart2native | Dart

Workflow syntax for GitHub Actions - GitHub Help

installation - How to refresh the environment of a PowerShell session after a Chocolatey install without needing to open a new session - Stack Overflow

Istio VirtualServiceのHost衝突を検知するAdmission Webhookをつくってみる

JX通信社Advent Calendar 2019」15日目の記事です。昨日はペイさんによるNuxt.js + firebaseで「積ん読防止」アプリを作ってみたでした。

こんにちは、SREのたっち(TatchNicolas)です。

はじめに

引き続き、KubernetesのAdmission Webhooksについて書きます。Admission Webhooksとは何か、簡単に作って動かしてみる方法については前回記事も参照してください。

tech.jxpress.net

今回はもう少し踏み込んで、ちょっとだけ役に立ちそうなAdmission Webhookを書いてみたいと思います。

TL; DR

  • Istio VirtualServiceのHostの衝突を防止するValidating Webhookを作った
  • 似たようなロジックでk8s本体のingressなどにも使えるはず

前回までのあらすじ

Kubernetesのリソース操作は、kube-apiserverにリクエストを送ることで行います。その操作がetcdに反映される前にそのリクエスト内容をValidate/MutateできるのがAdmission controllerで、Kubernetesにbuilt-inしなくても自前のValidate/Mutateの処理を足せるのがAdmission Webhooksです。

前回はWebhook作成に必要なもの(Webhookを動かすDeployment、Serviceリソース、証明書など)を作る手順をゼロから実施し、サンプルとして metadata.labels の規約が守られているかチェックしたり、 metadata.name に接頭辞を自動で付けたりする処理を実装しました。

今回やってみたこと

IstioのVirtualServiceを定義するときに、ドキュメントにもあるように、同じHostを複数のVirtualServiceに分割して書くやり方があります。

しかし、内向けのAPIなどの定義にパスではなくホスト名でマイクロサービスを定義している場合にはhostの衝突は避けたいと思います。

たとえば、 api.some-product というホスト名がすでに存在しているときに別のチームで *.some-product のようなホスト名が登録されてしまうと、複数のVirtualServiceリソース間での評価順序は保証されないため、リクエストが吸い取られて意図しない挙動をしてしまうかもしれません。

そこで、マニフェスト適用時に衝突を検出できるようなValidating Webhookを作ってみました。

サンプルコードは以下になります。

github.com

やってみる

Prerequisites

  • Istioが動いているクラスタ*1
  • 公式ドキュメント*2か前回の記事を参考に、証明書などWebhook開発のための準備ができていること
  • Webhookを動かしているPodにkube-apiserverを叩かせるための権限を与えていること*3

Webhookのリソース定義

webhooks.rules 以下にValidation対象のリソース種別を書いていきます。

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: "istio-vsvc-host"
webhooks:
- name: "istio-vsvc-host.hoge.fuga.local"
  failurePolicy: Fail
  rules:
  - apiGroups: ["networking.istio.io"]
    operations: ["CREATE","UPDATE"]
    apiVersions: ["v1alpha3"]
    resources: ["virtualservices"]
    scope: "Namespaced"
  clientConfig:
    caBundle: <server.crtの中身をbase64エンコードして貼る>
    service:
      namespace: default
      name: istio-vsvc-host
      path: /validate
  admissionReviewVersions: ["v1", "v1beta1"]
  timeoutSeconds: 5
  sideEffects: None

Webhookの実装

Webhookもいたってシンプルです。

import fnmatch

from flask import Flask, jsonify, request
from kubernetes import client, config


app = Flask(__name__)

config.load_incluster_config()
# telepresenceではこっちを使う
# config.load_kube_config()

conf=client.Configuration()
api_client=client.ApiClient(configuration=conf)

@app.route('/validate', methods=['POST'])
def validate():
    try:
        # リクエストから必要な情報を抜き出す
        req = request.get_json()
        new_hosts = req['request']['object']['spec']['hosts']
        apiserver_resp = api_client.call_api(
            '/apis/networking.istio.io/v1alpha3/namespaces/default/virtualservices',
            'GET',
            auth_settings=['BearerToken'],
            response_type='object',
        )
        nested_existing_hosts = {
            tuple(item['spec']['hosts']) for item in apiserver_resp[0]['items']
        }
        existing_hosts = {item for sublist in nested_existing_hosts for item in sublist}

        print(f'existing_hosts: {existing_hosts}')
        print(f'new_hosts: {new_hosts}')

        # UPDATEのときは、oldに入っているものは検査対象から除外する
        operation = req['request']['operation']
        if operation == 'UPDATE':
            old_hosts = req['request']['oldObject']['spec']['hosts']
            for host in old_hosts:
                existing_hosts.remove(host)
            print(f'updated existing_hosts: {existing_hosts}')

        # hostsの被りがないかチェックする
        pair = get_collision_pair(new_hosts, existing_hosts)

        if pair:
            allowed = False
            message = f'{pair[0]} collides with {pair[1]} which already exists'
        else:
            allowed = True
            message = f'No collision detected'


        # 結果を返す
        return jsonify({
            'apiVersion': 'admission.k8s.io/v1',
            'kind': 'AdmissionReview',
            'response': {
                'uid': request.get_json()['request']['uid'],
                'allowed': allowed,
                'status': {'message': message}
            }
        }), 200

    except (TypeError, KeyError):
        return jsonify({'message': 'Invalid request'}), 400

def get_collision_pair(new_hosts, existing_hosts):
    for new_host in new_hosts:
        for existing_host in existing_hosts:
            if fnmatch.fnmatch(new_host, existing_host) or fnmatch.fnmatch(existing_host, new_host):
                return new_host, existing_host

動かしてみる

以下のようなVirtualServiceが既に存在する状態で、新たにVirtualServiceのhostを増やしてみます。

$ kubectl get virtualservices
NAME              GATEWAYS   HOSTS                     AGE
existing-vsvc-1              [hoge.tatchnicolas.com]   15s
existing-vsvc-2              [fuga.tatchnicolas.com]   15s

まずはぶつからない場合。

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: new-vsvc
spec:
  hosts:
  - 'new.tatchnicolas.com'
  http:
  - name: "ClusterIP"
    route:
    - destination:
        host: new-serivce.default.svc.cluster.local

これはapplyしても普通に成功します。

$ kubectl apply -f istio-vsvc-host/new_vsvc.yaml
virtualservice.networking.istio.io/new-vsvc created

では、既存のhostと衝突させてみましょう。

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: new-vsvc
spec:
  hosts:
  - 'new.tatchnicolas.com'
  - 'fuga.tatchnicolas.com'  # これがぶつかる
  http:
  - name: "ClusterIP"
    route:
    - destination:
        host: new-serivce.default.svc.cluster.local

衝突を検知し、きちんと拒否していることがわかります。

$ kubectl apply -f istio-vsvc-host/new_vsvc.yaml
(中略)
for: "istio-vsvc-host/new_vsvc.yaml": admission webhook "istio-vsvc-host.hoge.fuga.local" denied the request: fuga.tatchnicolas.com collides with fuga.tatchnicolas.com which already exists

ワイルドカードにも対応できます。

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: new-vsvc
spec:
  hosts:
  - '*.tatchnicolas.com'
  http:
  - name: "ClusterIP"
    route:
    - destination:
        host: new-serivce.default.svc.cluster.local

こちらも拒否できていますね。

$ kubectl apply -f istio-vsvc-host/new_vsvc.yaml
(中略)
for: "istio-vsvc-host/new_vsvc.yaml": admission webhook "istio-vsvc-host.hoge.fuga.local" denied the request: *.tatchnicolas.com collides with fuga.tatchnicolas.com which already exists

さいごに

リクエストを適切に処理できれば実装言語は何でも良いので、前回のコードをベースにそのままPythonで書きました。比較のロジックもかなり簡単に書いたのでパフォーマンス面で改善の余地がありそうですし、実用レベルに持っていくにはNamespace対応したりテストもしっかり書かないといけないでしょう。

それでも、KubernetesやIstioに手を加えずに、ちょっとしたチェック/書き換え機能を追加できるのは非常に便利です。

理想的にはもっと汎用的にValidation条件を渡せるようにして、カスタムリソースとして簡単に適用できるようにするとか応用の幅は広そうです。

プロトタイプとテストをサクッと作れる言語で書いて、あとからリファクタリングしたり別の言語に書き換えることもできるので、ユースケースに合わせて上手に使っていきたいです。

*1:Minikubeのデフォルトではリソースが足りないせいでIstioのコントロールプレーンが起動せず、また手元の環境(i7/16GB RAM)ではIstio推奨のリソースを動かすのは辛かったです

*2:https://kubernetes.io/docs/tasks/tls/managing-tls-in-a-cluster/

*3:https://github.com/TatchNicolas/sample-admission-webhook/blob/master/istio-vsvc-host/rbac.yaml