websockets(python) 型アノテーション

websocketsを使う

websocketsはWebSocketプロトコルPythonで実装する上で便利なライブラリである。

pypi.org

# インストール
$ pip install websockets

公式ドキュメントからそのまんまサンプルコードを

# クライアント
import asyncio
import websockets

async def hello():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        name = input("What's your name? ")

        await websocket.send(name)
        print(f">>> {name}")

        greeting = await websocket.recv()
        print(f"<<< {greeting}")

asyncio.run(hello())
# サーバ
import asyncio
import websockets

async def hello(websocket, path):
    name = await websocket.recv()
    print(f"<<< {name}")

    greeting = f"Hello {name}!"

    await websocket.send(greeting)
    print(f">>> {greeting}")

async def main():
    async with websockets.serve(hello, "localhost", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())

そしてお気に入りのエディターに入力して気付くのである。
「あれ?これ型ヒントどころか、インテリセンスも効かなくね?」

型ヒントを使う

原因は__init__.pyでなかなかの事をやっているから。

from __future__ import annotations

from .imports import lazy_import
from .version import version as __version__  # noqa


__all__ = [  # noqa
    "AbortHandshake",
    "basic_auth_protocol_factory",
    "BasicAuthWebSocketServerProtocol",
    "broadcast",
    "ClientConnection",
    "connect",
    "ConnectionClosed",
    "ConnectionClosedError",
    "ConnectionClosedOK",
    "Data",
    "DuplicateParameter",
    "ExtensionName",
    "ExtensionParameter",
    "InvalidHandshake",

# 以降省略 #

Issue で「カバレッジ100%でテストされているから型ヒントなくても大丈夫!」って言ってるからビビったけど、流石に実装してあった。

ライブラリ側で大丈夫でも利用者がミスったら意味ねえ。
Djangoもそれが理由で捨てた。
型ヒントの有無はマジでライブラリ選択の決定打になり得る

次のように変更。

# クライアント
import asyncio
import websockets.legacy.client as websockets # import websockets

async def hello():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        name = input("What's your name? ")

        await websocket.send(name)
        print(f">>> {name}")

        greeting = await websocket.recv()
        print(f"<<< {greeting}")

asyncio.run(hello())
import asyncio
import websockets.legacy.server as websockets # import websockets

async def hello(websocket, path):
    name = await websocket.recv()
    print(f"<<< {name}")

    greeting = f"Hello {name}!"

    await websocket.send(greeting)
    print(f">>> {greeting}")

async def main():
    async with websockets.serve(hello, "localhost", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())

以上、型ヒントのtipsでした。