Slack API (fastapi - slack_bolt)

본 게시글은 slackbot 를 만들어 볼건데요
지원을 잘 해주는 slack_bolt를 사용하고 fastAPI 기반 입니다.

 

아래 링크에서 개선된게 본 게시글이고 디테일한 설명은 아래 게시글을 참고해주세요

https://kimjanghyun1010.tistory.com/55

 

 

Event Subscriptions에 최초 등록시 challenge 요청을 통해 응답을 확인 합니다.

body에 challenge가 있다면 200을 리턴하여 Verified 시켜야 합니다.

 

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from slack_bolt import App
from slack_bolt.adapter.fastapi import SlackRequestHandler

app = FastAPI()
slack_app = App(token="xoxb-slack-bot-token", signing_secret="signing-secret")
handler = SlackRequestHandler(app=slack_app)

@app.post("/slack/message")
async def get_message(request: Request):
    body = await request.json()
    challenge = body.get('challenge')
    if challenge is not None:
        logger.info("Received challenge: %s", challenge)
        return JSONResponse(content=challenge, status_code=200)
    event = body.get('event', {})
    channel_id = event.get('channel')
    user_id = event.get('user')
    ts = event.get('event_ts')
    try:
        result = slack_app.client.conversations_history(channel=channel_id, limit=1)
        messages = result['messages'][0]['text']
    except SlackApiError as e:
        logger.error("Slack API Error: %s", str(e))
        raise HTTPException(status_code=500, detail=str(e))
    if bot_id != user_id and messages is not None and "!" in messages:
	'''
	비즈니스 작성부분
	'''
    return JSONResponse(content={"message": "Data received successfully"}, status_code=203)

@app.post('/slack/event')
async def incoming_slack_message(payload: str = Form(...)):
    '''
    event 처리
    '''
    form_json = json.loads(payload)
    selected_user_name = form_json["user"]["name"]
    selected_value = form_json["actions"][0]["selected_options"][0]["value"]
    '''
    비즈니스 작성부분
    '''
  
    return JSONResponse(content=incoming_format, status_code=200)

 

 

실행

import uvicorn

if __name__ == "__main__":
    uvicorn.run("fastapi_web.__init__:app", host="0.0.0.0", port=80, reload=True)