Web API with SlackSDK

게시 2024/03/16 업데이트 2024/03/19

By Jeek Lee 5 분읽는 시간

I. SlackSDK

(그림 I-1) SlackSDK

SlackSDK는 Slack API 사용을 위해 Slack에서 제공하는 모듈이다. Web API부터 UI Builder까지 상당히 많은 기능을 제공하는데, 이번엔 Web API를 통해 간단한 알림 로직을 구현했다.

(그림 I-2) Flow chart

알림 로직이 적용된 부분은 위와 같다. Flow Chart에서 *Lambda는 핵심 비즈니스 로직을 수행한다. 문제는 해당 Lambda가 안정적이지 않다는 점에 있다. 시간 관계상 Stress Test나 QA를 정상적으로 진행할 수 없고, 핵심 기능인 만큼 오류가 발생할 때마다 즉각적인 조치가 필요하다. 이러한 점에서 Lambda 호출 결과가 빠르게 전파되어야 한다 판단했고, Notion API와 Slack API를 활용했다.

*Lamda: AWS에서 제공하는 서버리스 컴퓨팅 플랫폼. 주로 주기적으로 호출되거나, 단발적(트리거)으로 동작 후 종료되는 코드에 많이 쓰인다.

Slack의 Web API Metod 페이지에서 활용할 수 있는 다양한 Web API를 확인할 수 있는데, 단순히 특정 채널에 원하는 메세지를 전송하는 건 chat.postMessage API로 충분하다.

II. WebClient

SlackSDK는 Slack API 사용을 위해 WebClient 객체를 사용한다.

  
# Example of recommended usage:
client = WebClient(token=os.environ['SLACK_API_TOKEN'])

WebClient 객체는 BaseClient 객체를 상속하는 형태인데, BaseClient.__init__()은 다음과 같다.

  
def __init__(
        self,
        token: Optional[str] = None,
        base_url: str = BASE_URL,
        timeout: int = 30,
        ssl: Optional[SSLContext] = None,
        proxy: Optional[str] = None,
        headers: Optional[dict] = None,
        user_agent_prefix: Optional[str] = None,
        user_agent_suffix: Optional[str] = None,
        # for Org-Wide App installation
        team_id: Optional[str] = None,
        logger: Optional[logging.Logger] = None,
        retry_handlers: Optional[List[RetryHandler]] = None,
    ):
        self.token = None if token is None else token.strip()
        """A string specifying an `xoxp-*` or `xoxb-*` token."""
        (...)

WebClient 객체 생성에 필요한 token은 xoxp- 혹은 xoxb- 형태로 시작하는 유저 토큰 혹은 봇 토큰이다. SlackAPI 페이지에서 앱 생성 및 워크스페이스 적용 이후에 OAuth & Permissions 메뉴에서 OAuth 토큰을 확인할 수 있다.

(그림 II-1) Slack Bot OAuth

Web API Metod의 각 API 명세서를 통해 API 실행 시 필요한 scope를 확인할 수 있다. 이번에 사용할 chat.postMessage의 경우 기본적으로 명시되어 있는 chat:write 외에도 문서를 쭉 확인하다 보면 추가적으로 필요한 scope들을 알 수 있다.

(그림 II-2) 문서를 끝까지 읽자

OAuth & Permissions의 스크롤을 내리다보면, 해당 앱이 접근할 수 있는 scope를 설정할 수 있다. 사용하려는 API에 맞춰 추가해주어야 정상적인 호출이 가능하다.

(그림 II-3) Scopes

III. chat_postMessage()

chat_postMessage()의 파라미터를 확인하면 다음과 같다.

  
def chat_postMessage(
    self,
    *,
    channel: str,
    text: Optional[str] = None,
    as_user: Optional[bool] = None,
    attachments: Optional[Union[str, Sequence[Union[Dict, Attachment]]]] = None,
    blocks: Optional[Union[str, Sequence[Union[Dict, Block]]]] = None,
    thread_ts: Optional[str] = None,
    reply_broadcast: Optional[bool] = None,
    unfurl_links: Optional[bool] = None,
    unfurl_media: Optional[bool] = None,
    container_id: Optional[str] = None,
    icon_emoji: Optional[str] = None,
    icon_url: Optional[str] = None,
    mrkdwn: Optional[bool] = None,
    link_names: Optional[bool] = None,
    username: Optional[str] = None,
    parse: Optional[str] = None,  # none, full
    metadata: Optional[Union[Dict, Metadata]] = None,
    **kwargs,
) -> SlackResponse:

해당 def를 사용하기 위해선, 메세지를 전송하고자 하는 channel의 ID를 알아야한다. channel ID는 conversations.list API를 통해 간단하게 확인할 수 있다.

(그림 III-1) conversations.list

channel ID를 확인했다면, 다음과 같은 방식으로 text 파라미터에 메세지를 설정해 메세지를 원하는 채널에 전송할 수 있다.

  
self.client.chat_postMessage(channel=self.channel_id, text=message)

chat_postMessage()는 전송된 메세지의 정보를 모두 반환하는데, 이를 통해 다음과 같이 댓글을 다는 것도 가능하다.

  
self.client.chat_postMessage(channel=self.channel_id, text=reply, thread_ts=message_ts)

IV. Result

chat_postMessage()를 통해 Lambda 관련 로그를 Slack으로 전송한 결과이다. (가독성을 위해 CloudWatch 로그 파싱 일부 진행)

(그림 IV-1) conversations.list

V. Reference

FastAPI