【Python】asyncio と h2 で HTTP/2 クライアントをつくる

h2 の examples の中に asyncio の HTTP/2 クライアントのコードの例がなかったのでつくった。

client.py

import asyncio
from h2.config import H2Configuration
from h2.connection import H2Connection
from h2.errors import ErrorCodes
from h2.events import (
    SettingsAcknowledged, ResponseReceived, DataReceived, StreamEnded
)

class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, on_con_lost, request_headers):
        self.on_con_lost = on_con_lost
        config = H2Configuration(client_side=True, header_encoding='utf-8')
        self.conn = H2Connection(config=config)
        self.conn.initiate_connection()
        self.transport = None
        self.request_headers = request_headers

    def connection_made(self, transport):
        self.transport = transport
        self.transport.write(self.conn.data_to_send())

    def data_received(self, data):
        try:
            events = self.conn.receive_data(data)
        except ProtocolError as e:
            self.transport.write(self.conn.data_to_send())
            self.transport.close()
        else:
            for event in events:
                if isinstance(event, SettingsAcknowledged):
                     self.conn.send_headers(1, self.request_headers, end_stream=True)
                     self.transport.write(self.conn.data_to_send())
                if isinstance(event, ResponseReceived):
                    print(event.headers)
                if isinstance(event, DataReceived):
                    print(event.data)
                if isinstance(event, StreamEnded):
                    self.transport.close()

    def connection_lost(self, exc):
        print('The server closed the connection')
        self.on_con_lost.set_result(True)

async def main():
    host = '127.0.0.1'
    port = 8000
    request_headers = [
        (':method', 'GET'),
        (':scheme', 'http'),
        (':path', '/'),
        (':authority', 'localhost'),
        ('user-agent', 'h2/1.0.0')
    ]

    loop = asyncio.get_running_loop()
    on_con_lost = loop.create_future()

    transport, protocol = await loop.create_connection(
        lambda: EchoClientProtocol(on_con_lost, request_headers),
        host, port)

    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

参考

# 古い Python のバージョンのサンプルコード
# https://github.com/johnson-li/h2-example/blob/master/basic/client.py

# asyncio.Protocol を継承するクラスによる TCP クライアントのコードの例
# https://docs.python.org/ja/3.13/library/asyncio-protocol.html#tcp-echo-client

# TLS なしの HTTP/2 (h2c) サーバー
# https://zenn.dev/masakielastic/articles/618f9f44f327f4