Merge pull request #22 from toshi-pono/feature/webhookpost

WebhookでPOSTのAPIがなかった問題を修正

README.md

@@ -43,6 +43,9 @@ poetry install
 make api_update
 ```
 
+> [!WARNING]
+> `post_webhook` についてはライブラリ未対応のため動的に生成できず、手動で追加しています
+
 ## License
 
 This project is licensed under the terms of the MIT license.

examples/.env.example

@@ -1,2 +1,4 @@
 BOT_ACCESS_TOKEN=SuperSecretToken
 BOT_VERIFICATION_TOKEN=SuperSecretVerificationToken
+WEBHOOK_SECRET=SuperSecretWebhookSecret
+WEBHOOK_ID=SuperSecretWebhookId

examples/webhook.py

@@ -0,0 +1,40 @@
+import os
+from os.path import join, dirname
+from dotenv import load_dotenv
+from aiotraq import Client
+from aiotraq.api.webhook import post_webhook
+import hashlib
+import hmac
+
+
+dotenv_path = join(dirname(__file__), ".env")
+load_dotenv(dotenv_path)
+
+WEBHOOK_ID = os.getenv("WEBHOOK_ID")
+WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
+
+
+def main() -> None:
+    client = Client(base_url="https://q.trap.jp/api/v3")
+
+    with client as client:
+        if WEBHOOK_ID is None or WEBHOOK_SECRET is None:
+            print("WEBHOOK_ID or WEBHOOK_SECRET is not set.")
+            return
+
+        message = "Hello World!"
+        # メッセージ本文をWebhookシークレットでHMAC-SHA1でハッシュ化した結果をhex形式で表した文字列
+        # https://bot-console.trap.jp/docs/webhook/send
+        secret = hmac.new(WEBHOOK_SECRET.encode(), message.encode(), hashlib.sha1).hexdigest()
+
+        result = post_webhook.sync_detailed(
+            client=client,
+            webhook_id=WEBHOOK_ID,
+            body=message,
+            x_traq_signature=secret,
+        )
+        print(result)
+
+
+if __name__ == "__main__":
+    main()

libs/aiotraq/aiotraq/api/webhook/post_webhook.py

@@ -0,0 +1,159 @@
+"""
+HACK: ライブラリが text/plain に対応していないため、手動で追加しています
+"""
+
+from http import HTTPStatus
+from typing import Any, Dict, Optional, Union
+
+import httpx
+
+from ... import errors
+from ...client import AuthenticatedClient, Client
+from ...types import UNSET, Response, Unset
+
+
+def _get_kwargs(
+    webhook_id: str,
+    *,
+    body: str,
+    embed: Union[Unset, int] = 0,
+    x_traq_signature: Union[Unset, str] = UNSET,
+    x_traq_channel_id: Union[Unset, str] = UNSET,
+) -> Dict[str, Any]:
+    headers: Dict[str, Any] = {}
+    if not isinstance(x_traq_signature, Unset):
+        headers["X-TRAQ-Signature"] = x_traq_signature
+
+    if not isinstance(x_traq_channel_id, Unset):
+        headers["X-TRAQ-Channel-Id"] = x_traq_channel_id
+
+    params: Dict[str, Any] = {}
+
+    params["embed"] = embed
+
+    params = {k: v for k, v in params.items() if v is not UNSET and v is not None}
+
+    _kwargs: Dict[str, Any] = {
+        "method": "post",
+        "url": f"/webhooks/{webhook_id}",
+        "params": params,
+    }
+
+    _body = body
+
+    _kwargs["content"] = _body
+    headers["Content-Type"] = "text/plain"
+
+    _kwargs["headers"] = headers
+
+    return _kwargs
+
+
+def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[Any]:
+    if response.status_code == HTTPStatus.NO_CONTENT:
+        return None
+    if response.status_code == HTTPStatus.BAD_REQUEST:
+        return None
+    if response.status_code == HTTPStatus.NOT_FOUND:
+        return None
+    if client.raise_on_unexpected_status:
+        raise errors.UnexpectedStatus(response.status_code, response.content)
+    else:
+        return None
+
+
+def _build_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Response[Any]:
+    return Response(
+        status_code=HTTPStatus(response.status_code),
+        content=response.content,
+        headers=response.headers,
+        parsed=_parse_response(client=client, response=response),
+    )
+
+
+def sync_detailed(
+    webhook_id: str,
+    *,
+    client: Union[AuthenticatedClient, Client],
+    body: str,
+    embed: Union[Unset, int] = 0,
+    x_traq_signature: Union[Unset, str] = UNSET,
+    x_traq_channel_id: Union[Unset, str] = UNSET,
+) -> Response[Any]:
+    """Webhookを送信
+
+    Webhookにメッセージを投稿します。
+    secureなウェブフックに対しては`X-TRAQ-Signature`ヘッダーが必須です。
+    アーカイブされているチャンネルには投稿できません。
+
+    Args:
+        webhook_id (str):
+        embed (Union[Unset, int]):  Default: 0.
+        x_traq_signature (Union[Unset, str]):
+        x_traq_channel_id (Union[Unset, str]):
+        body (str): メッセージ文字列
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[Any]
+    """
+
+    kwargs = _get_kwargs(
+        webhook_id=webhook_id,
+        body=body,
+        embed=embed,
+        x_traq_signature=x_traq_signature,
+        x_traq_channel_id=x_traq_channel_id,
+    )
+
+    response = client.get_httpx_client().request(
+        **kwargs,
+    )
+
+    return _build_response(client=client, response=response)
+
+
+async def asyncio_detailed(
+    webhook_id: str,
+    *,
+    client: Union[AuthenticatedClient, Client],
+    body: str,
+    embed: Union[Unset, int] = 0,
+    x_traq_signature: Union[Unset, str] = UNSET,
+    x_traq_channel_id: Union[Unset, str] = UNSET,
+) -> Response[Any]:
+    """Webhookを送信
+
+    Webhookにメッセージを投稿します。
+    secureなウェブフックに対しては`X-TRAQ-Signature`ヘッダーが必須です。
+    アーカイブされているチャンネルには投稿できません。
+
+    Args:
+        webhook_id (str):
+        embed (Union[Unset, int]):  Default: 0.
+        x_traq_signature (Union[Unset, str]):
+        x_traq_channel_id (Union[Unset, str]):
+        body (str): メッセージ文字列
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[Any]
+    """
+
+    kwargs = _get_kwargs(
+        webhook_id=webhook_id,
+        body=body,
+        embed=embed,
+        x_traq_signature=x_traq_signature,
+        x_traq_channel_id=x_traq_channel_id,
+    )
+
+    response = await client.get_async_httpx_client().request(**kwargs)
+
+    return _build_response(client=client, response=response)