PushKitのプッシュ通知をPython3(hyper)から送信する - 『入る学科間違えた高専生』の日記

from hyper import HTTP20Connection
from pathlib import Path

import json
import os
import ssl

BASE_DIR = Path(__file__).parent
CERT_FILE = BASE_DIR / f'YOUR_VOIP_CERTIFICATE.pem'

サンドボックスを使うかどうかの設定と送信するbodyとヘッダー

if bool(int(os.environ.get('USE_SANDBOX'))):
    host = 'api.development.push.apple.com'
else:
    host = 'api.push.apple.com'

body = json.dumps(
    {'key': 'send by hyper\' HTTP20Connection'}  # ここはなんでもいいです。容量以内であれば。
)
headers = {
    'apns-expiration': '0',  # 使用期限。詳しくは公式ドキュメント
    'apns-priority': '10',  # 優先度。詳しくは公式ドキュメント
    'apns-topic': os.environ.get('BUNDLE_ID')
}

クライアント証明書の設定及び、HTTP/2.0での通信をするコネクションの生成

ssl_context = ssl.create_default_context()
ssl_context.load_cert_chain(CERT_FILE)
connection = HTTP20Connection(host, post=443, ssl_context=ssl_context, force_proto='h2')

curl コマンドで使用する -E , --cert にあたるものの設定です。

HTTP20Connection のコンストラクタ内で port=443 の場合は secure=True になる実装だったので特に secure については設定しませんでした。

https://github.com/Lukasa/hyper/blob/master/hyper/http20/connection.py#L107

実際に送信

def send_request(registration_id):
    connection.request(
        'POST', f'/3/device/{registration_id}', body, headers
    )
    response = connection.get_response()

    if response.status == 200:
        return
    response_body = json.loads(response.read().decode('utf-8'))
    reason = response_body.get('reason')
    if reason:
        print(reason)

最後にコネクションを閉じる

connection.close()

ちょっと補足

from contextlib import closing


def create_connection():
    ssl_context = ssl.create_default_context()
    ssl_context.load_cert_chain(CERT_FILE)
    return HTTP20Connection(host, post=443, ssl_context=ssl_context, force_proto='h2')


with closing(create_connection()) as connection:
    send_request("YOUR_REGISTRATION_ID")

のようにすると最後に HTTP20Connection の close() 忘れを防げますね。