-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path__init__.py
More file actions
155 lines (130 loc) · 4.61 KB
/
__init__.py
File metadata and controls
155 lines (130 loc) · 4.61 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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
"""Decatur Makers Machine Access Control."""
import argparse
import logging
import os
import sys
from asyncio import AbstractEventLoop
from asyncio import get_event_loop
from time import time
from quart import Quart
from quart import has_request_context
from quart import request
from quart.logging import default_handler
from quart_schema import QuartSchema
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from dm_mac.models.machine import MachinesConfig
from dm_mac.models.users import UsersConfig
from dm_mac.slack_handler import SlackHandler
from dm_mac.utils import set_log_debug
from dm_mac.utils import set_log_info
from dm_mac.views.api import api
from dm_mac.views.machine import machineapi
from dm_mac.views.prometheus import prometheus_route
logger: logging.Logger = logging.getLogger()
logging.basicConfig(
level=logging.WARNING, format="[%(asctime)s %(levelname)s] %(message)s"
)
# BEGIN adding request information to logs
class RequestFormatter(logging.Formatter):
"""Custom log formatter to add request information."""
def format(self, record: logging.LogRecord) -> str:
"""Custom log formatter to add request information."""
if has_request_context():
record.url = request.url
record.remote_addr = request.remote_addr
else:
record.url = None
record.remote_addr = None
return super().format(record)
formatter = RequestFormatter(
"%(asctime)s %(levelname)s:[%(remote_addr)s]:%(name)s:%(message)s"
)
default_handler.setFormatter(formatter)
# END adding request information to logs
# enable logging from libraries (i.e. everything but the views)
logging.getLogger().addHandler(default_handler)
logging.getLogger("AUTH").setLevel(logging.INFO)
logging.getLogger().setLevel(logging.INFO)
# see: https://github.com/pallets/flask/issues/4786#issuecomment-1416354177
api.register_blueprint(machineapi)
app: Quart
def asyncio_exception_handler(_, context):
# get details of the exception
exception = context["exception"]
message = context["message"]
# log exception
logger.error(f"Task failed, msg={message}, exception={exception}")
def create_app() -> Quart:
"""Factory to create the app."""
app: Quart = Quart("dm_mac")
QuartSchema(
app,
info={
"title": "Decatur Makers Machine Access Control API",
"version": "0.9.1",
"description": (
"HTTP API for the Decatur Makers Machine Access Control system. "
"Used by ESP32-based Machine Control Units (MCUs) to report state "
"and receive commands, and by administrators for machine management."
),
},
tags=[
{"name": "Machine", "description": "Machine state and control endpoints"},
{"name": "Admin", "description": "Administrative endpoints"},
{"name": "Monitoring", "description": "Monitoring and metrics endpoints"},
],
)
app.config.update({"MACHINES": MachinesConfig()})
app.config.update({"USERS": UsersConfig()})
app.config.update({"START_TIME": time()})
app.config.update({"SLACK_HANDLER": None})
app.register_blueprint(api)
app.add_url_rule("/metrics", view_func=prometheus_route)
return app
def main() -> None:
global app
p = argparse.ArgumentParser(description="Run Machine Access Control (MAC) server")
p.add_argument(
"-d",
"--debug",
dest="debug",
action="store_true",
default=False,
help="Debug mode",
)
p.add_argument(
"-v",
"--verbose",
dest="verbose",
action="store_true",
default=False,
help="verbose output",
)
p.add_argument(
"-P",
"--port",
dest="port",
action="store",
type=int,
default=5000,
help="Port number to listen on (default 5000)",
)
args = p.parse_args(sys.argv[1:])
if args.verbose:
set_log_debug(logger)
else:
set_log_info(logger)
loop: AbstractEventLoop = get_event_loop()
loop.set_exception_handler(asyncio_exception_handler)
app = create_app()
token: str = os.environ.get("SLACK_APP_TOKEN", "").strip()
if token:
slack: SlackHandler = SlackHandler(app)
app.config.update({"SLACK_HANDLER": slack})
handler = AsyncSocketModeHandler(
slack.app, os.environ["SLACK_APP_TOKEN"], loop=loop
)
loop.create_task(handler.start_async())
app.run(loop=loop, debug=args.debug, host="0.0.0.0", port=args.port)
if __name__ == "__main__": # pragma: no cover
main()