forked from slackapi/bolt-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathasync_server.py
More file actions
82 lines (70 loc) · 2.8 KB
/
async_server.py
File metadata and controls
82 lines (70 loc) · 2.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import logging
from aiohttp import web
from slack_bolt.adapter.aiohttp import to_bolt_request, to_aiohttp_response
from slack_bolt.response import BoltResponse
from slack_bolt.util.utils import get_boot_message
class AsyncSlackAppServer:
port: int
path: str
bolt_app: "AsyncApp" # type:ignore
web_app: web.Application
def __init__( # type:ignore
self,
port: int,
path: str,
app: "AsyncApp", # type:ignore
):
"""Standalone AIOHTTP Web Server
Refer to AIOHTTP documents for details.
https://docs.aiohttp.org/en/stable/web.html
"""
self.port = port
self.path = path
self.bolt_app: "AsyncApp" = app
self.web_app = web.Application()
self._bolt_oauth_flow = self.bolt_app.oauth_flow
if self._bolt_oauth_flow:
self.web_app.add_routes(
[
web.get(
self._bolt_oauth_flow.install_path, self.handle_get_requests
),
web.get(
self._bolt_oauth_flow.redirect_uri_path,
self.handle_get_requests,
),
web.post(self.path, self.handle_post_requests),
]
)
else:
self.web_app.add_routes([web.post(self.path, self.handle_post_requests)])
async def handle_get_requests(self, request: web.Request) -> web.Response:
oauth_flow = self._bolt_oauth_flow
if oauth_flow:
if request.path == oauth_flow.install_path:
bolt_req = await to_bolt_request(request)
bolt_resp = await oauth_flow.handle_installation(bolt_req)
return await to_aiohttp_response(bolt_resp)
elif request.path == oauth_flow.redirect_uri_path:
bolt_req = await to_bolt_request(request)
bolt_resp = await oauth_flow.handle_callback(bolt_req)
return await to_aiohttp_response(bolt_resp)
else:
return web.Response(status=404)
else:
return web.Response(status=404)
async def handle_post_requests(self, request: web.Request) -> web.Response:
if self.path != request.path:
return web.Response(status=404)
bolt_req = await to_bolt_request(request)
bolt_resp: BoltResponse = await self.bolt_app.async_dispatch(bolt_req)
return await to_aiohttp_response(bolt_resp)
def start(self) -> None:
"""Starts a new web server process.
:return: None
"""
if self.bolt_app.logger.level > logging.INFO:
print(get_boot_message())
else:
self.bolt_app.logger.info(get_boot_message())
web.run_app(self.web_app, host="0.0.0.0", port=self.port)