Make組ブログ

Python、Webサービスや製品開発、ライブラリー開発についてhirokikyが書きます

Python3.8のPEP572 (Assignment Expression) とリスト内包表記でフィボナッチ数列を作って遊んだ記録

あつおさんがPEP572でフィボナッチ数列を作って遊んでいたので、僕も遊んでみた。

Python 3.8.0a1時点の話です。

我ながら [(p2 := (p1 + (p1:=p2))) for _ in ...] はよく出来てるなと思った。 あと、PEP572のnamed assignmentは使ってみると意外と制限があるので、思ったよりも良いなと思った。

ただこの動きは直感的にはビックリした。

>>> a = 1
>>> b = 2
>>> a + (a := b)
3
>>> a
2
>>> b
2

あつおさんのブログも見てね

atsuoishimoto.hatenablog.com

aiohttp.ClientSessionのリクエストをaioresponsesでモックする。特定のホストへのアクセスは許可する

aiohttp.ClientSession() でのアクセスをテスト時にモックしたいときに、 aioresponses というライブラリーが使えます。

github.com

Pythonのrequestsライブラリーを簡単にモックできる responses というライブラリーがあるのですが、そのresponses-likeに使えれて便利です。

from aioresponses import aioresponses


def test():
    with aioresponses() as mocked:
        moked.get('http://foobar/api', payload={...})

       call_function_under_test()

特定のホストへのアクセスは許可する

この aioresponses 、良いのですがデフォルトで全てのリクエストをモックするようです (0.5.2時点) 。 問題としては例えば、 aiohttp でサーバーのテストを書いているとき、そのテストサーバーへのアクセスなどもモックして動作しなくなることです。

そこで、以下のようなpytestのフィクスチャーを書いておくと便利です。 ここでは pytest-aiohttp をインストールしている前提です。

@pytest.fixture
def responses(cli):
    base = cli.make_url('/')
    with aioresponses(passthrough=[str(base), ...]) as mocked:
        yield mocked

この例では、テストサーバーのホスト(上記の base 変数の値)へのアクセスを許可しています。 バックエンドのミドルウェアを通してテストする場合などは、 'unix://' なども許可しておくと良いです。

基本的に全てのアクセスは許可しておいて mock.get(...) のように追加したURLだけモックするようにしたい気はします。

ともあれ、aiohttp.ClientSessionを使うと aioresponses を使うと便利です。

pytest-asyncioとpytest-aiohttpを一緒に使うとRuntimeError got Future attached to a different loopがでる

pytest-asynciopytest-aiohttp を同時に使うと、テストの中で使うイベントループが別のものになってしまってエラーが出ます。

RuntimeError: Task <Task pending coro=<...> cb=[...]> got Future <Future pending> attached to a different loop

github.com

github.com

aiohttpを使ってるWebアプリのテストを書くときは pytest-aiohttp に寄せたほうが良さそうです。 pytest-aiohttp を使えば loop のフィクスチャーで、テストで使うイベントループが取れるのでそれに寄せるようにします。

github.com

asyncioとpytestでテスト用fixtureを作るときにgot Future <Future pending> attached to a different loopが出る話

Pythonでasyncioを使うアプリケーションのテストを書くとき、event loopに気を遣ってあげないとエラーになるという話です (今回はaiohttpを使ってWebアプリを作っているときに起こりました)。

こんなエラーがでました。

RuntimeError: Task <Task pending coro=<...> cb=[...]> got Future <Future pending> attached to a different loop

テスト側で動かすイベントループと、アプリ側が動かそうとする、イベントループに違いがあると発生してしまいます。 例えばアプリ側で何らかのバックエンドにアクセスするクライアントを使っていて、テスト側でもデータ(フィクスチャー)を作ったりクリーンアップしたいとします。 そのときに双方の使うクライアントが内部で使っているイベントループが違うものだと上記のエラーが発生します。

テスト用の pytest-asyncio などのライブラリーが、テスト用のイベントループを用意する場合は、アプリ側でクライアントを起動時にモジュールグローバルに保存しないようにしておきます。 そのクライアントがインスタンス化されるときに、内部で asyncio.get_event_loop を呼ぶことがあるからです。 もしくはテスト時にクライアントを差し替えるようにします。

この例だと pytest-asyncioevent_loop フィクスチャーを取っておくことでテスト用のイベントループを有効にしておきます。

import pytest

import somebackend


@pytest.fixture
async def somebackend(event_loop):
    client = somebackend.Client()
    try:
        yield client
    finally:
        do_cleanup_here()

テストでは以下のように使えます。

@pytest.mark.asyncio
async def test_foo(somebackend):
    await somebackend.create("some thing")

    actual = await call_function_under_test()

    assert actual == "expected"

今回は aiodocker を使っていて起こりました。

この情報が何かの参考になれば嬉しいです。 また、asyncioやイベントループ周りについて詳しければ教えてください。

近そうな話

github.com

Pythonのaiohttpでストリーム(aiohttp.web_ws.WebSocketResponse)をPipeする

最近Pythonの非同期処理、asyncioを使ったプログラムを書いています。 今までは非同期だとNode.jsを使っていたんですが、aiohttpや周辺ライブラリーが揃ってきたようなので使っています。

Node.jsの場合、Streamは stream.pipe(other_stream) のようにPipeできるのですが、それをaiohttpの WebSocketResponse でやる方法を書きます。

Server Reference — aiohttp 3.4.4 documentation

asyncioの Streams でも要領は同じなので参考になると思います。

(注意: Pythonのasyncioやaiohttpはまだ新しいものなので、情報が古くなってないか気をつけてください)

Pipeを1回やる(一方通行の)例

1つのストリームから読み込んで、もう片方のストリームに書き込むのは簡単です。

async for msg in ws:
    await to_ws.send_str(msg.data)

WebSocketでアクセスを受け付けて、バックエンドの別のWebSocketにつなぎ込むaiohttpのView関数を考えると、以下のようになります。

from aiohttp import web

routes = web.RouteTableDef()


@routes.post('...')
async def websocket_gateway(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    session = aiohttp.ClientSession()
    to_ws = await session.ws_connect('...')

    async for msg in ws:
        await to_ws.send_str(msg.data)

    await ws.close()
    await to_ws.close()

この場合、サーバー側のWebSocket ws がクローズすると処理が終了します。

他にも書き込み先のWebSocket to_ws が先にクローズした場合の処理や、 msg.data がバイト列の場合に to_ws.send_bytes にする処理なども必要そうです。

相互にPipeする例

上記の例ではWebSocketは一方通行にしか仲介されていません。ここで、 to_ws からの入力も、 ws に返すようにしましょう。 その場合、 2つの非同期処理を同時に実行する必要があります 。 1つの while ループの中で2つのストリームの読み書きをしても、片方が書き込まないと、もう片方の入力が受け取れないようになるので難しいです。 以下のように別々のループをして、それを asyncio.gather するとうまくいきます。

import asyncio

from aiohttp import web
from aiohttp.http import WSMsgType

WS_CLOSE_TYPES = (WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED)

routes = web.RouteTableDef()


async def pipe(f, t):
    while not f.closed and not f.closed:
        msg = await f.receive()
        if msg.type in WS_CLOSE_TYPES:
            break

        await t.send_str(msg.data)
    return


@routes.post('...')
async def websocket_gateway(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    session = aiohttp.ClientSession()
    to_ws = await session.ws_connect('...')

    await asyncio.gather(
        pipe(ws, cs),
        pipe(cs, ws)
    )

    await ws.close()
    await to_ws.close()

asyncio.gather を使うことで、2つの処理を同時に実行して、両方が終了するのを待ってくれます。 優秀なやつですが、今回は片方のWebSocketが終わったら両方終了してほしいので、 pipe 関数内に工夫をしています。

pipe 関数の中で async for msg in ws せずに while ループを使っているのは、両方のWebSocketがクローズしていない限り処理を続けるようにするためです。 そうしないと、片方のWebSocketがクローズしてるのに、反対のWebSocketから書き込まれようとしてエラーになるからです。

こんな感じで、 asyncioaiohttp はかなり刺激的で面白いのでぜひ試してみてください。 誰得情報かもしれませんが、 aiohttpasyncio についての情報を出していけたらなと思います。

プログラミングすると「イライラ」する。向いてない?への回答

ふとこの前プログラミングを教えているときに言われたことです。

プログラミングをしていると『イライラ』してしまうんですが。。

「プログラミングをしてイライラしてしまう自分はプログラミングが向いてないんじゃないか」とか、「良くないことなのかな」というニュアンスでの話でした。その心配な気持ちは分かりますが、 プログラミング中はイライラしても大丈夫なので安心してください 。そういうものです。 むしろ、そこまでイライラするほど真剣に取り組める時点でプログラミングに向いてます。

多少イライラするのは普通なので大丈夫

プログラムが動かないとき、サーバーが止まってしまうとき、環境構築がうまくいかないとき、一生懸命に頭を使ってるんで誰でもイライラすると思います。 プログラミングしているときはイライラして大丈夫です 。するものと思ってください。僕はしてます。かなり。

人間も生物なので、小さいところに閉じこもって頭だけを使っているのは異常な状態です。ソワソワとかイライラとかして当然です。 何なら部屋の中を動き回ったり、足を机に乗せたりしたら良いと思います。僕はしょっちゅうしてます。

もちろんストレスフルでシンドいなら休憩しましょう。でも多少「うーん」とか「なんだよー」とか思ったり、「ダルいなー」とか「うおおおおこのやろおおお」とか思うのは当たり前くらいに思って大丈夫です。

そんなことがあってこそ物ができる喜びとか、使って便利になった嬉しさとか、美しいものを作れた自負とか、そういうものをより大切にしたいなと思います。

フローに対する誤解

プログラマーはいっつもハイに楽しくプログラミングしてフロー状態になっていると思われている可能性がありますが、それはないです。

本当にフローな状態に入っているときなんかはイライラせず何時間でも集中できますが、そんなときが来るのは毎日プログラミングしててもホンのたまにです。 頭の中で出来上がりつつも少しチャレンジングな場合はフローになりやすいですが、世の中そうも簡単じゃないので頭を使って悩みます。

もちろん、いつもフローに近づけるように環境を整備したり体調を整えたり準備はしますが、それでもなかなか難しいレアなものだと思います。

羽生さんもイライラしているらしい

プロ棋士羽生善治さんが似たことを言っていたので紹介します。 僕の中では、この言葉を聞いたときに救われた気持ちになったので共有したいです。

以下、将棋中にどうして扇子を「パチパチ」鳴らすのか、についての答え。

将棋を考えてることってかなりイライラすることなんですよ。 こうやってもこうやってもこうやってもうまく行かないっていう状況を考え続けているんで。

そこですごくストレスが溜まるというか、イライラする。 それを扇子にぶつけてるんですよね。

プロフェッショナル 仕事の流儀 シーズン1 14. file:020 直感は経験で磨く 棋士 羽生善治

プライムにも動画があるのでぜひプライム会員になって見てください http://amzn.asia/d/crRgSRR

羽生さんみたいなプロ棋士でもやっぱり「ああでもないこうでもない」と考えるとイライラするそうです。 もちろん節度はありますが、そういうものと思って大丈夫ですよ、という話でした。

むしろ、イライラしてからが本番です。そのイライラの波に乗るなり、一旦鎮めるなり、もうやめて寝るなり、うまい扱い方をぜひ身につけてください。

DjangoGirlsTokyo 4 でコーチをしました #djangogirls

DjangoGirls Tokyo #4 でコーチをしてきました。 私がコーチをするのは 第1回目 以来です。

djangogirls.org

進め方としては、 Django Girls Tutorial というチュートリアルを参加者の方が進めて学んで、コーチがサポートします。 このチュートリアルはかなり分かりやすい内容で、プログラミング初心者の人でもWebアプリ(ブログ)を作りながら学べるようになっています。 オススメです

さらに、今回のイベントでは参加者の方2,3人あたりコーチが1人ついて、とても手厚く学べる環境だったと思います。

プログラミングやWebアプリ開発を学びたい!という女性の方はぜひ参加してみてください

DjangoGirlsは逆差別なのではないか、への答えの発表もあり良かった

ランチの間は、自由参加で短い5分程度の発表を行いました。 その中でこの発表の内容がとても良かったので共有します。

「女性向け」とするとそれは逆の差別なのではないか、という疑問はあるけど、「気持ちは分かるが、そうではないよ」という内容の発表でした。 現状、プログラマーの人口としては男性が多いです。ですので格差是正のためにこの活動が大切であり、究極なくなれば良いというのは正しいということです。

私もこの考えに共感しますし、運営の体制もバッチリなので、DjangoGirlsにはなるべく協力したいなと改めて思いました。 ぜひ、 コーチとして協力してくれる人も増えてくれると嬉しい なと思います。

他、写真

こちらに素敵なアルバムがあるので、見てみてください。

photos.app.goo.gl