-
-
Notifications
You must be signed in to change notification settings - Fork 1
/
run.py
139 lines (108 loc) · 3.75 KB
/
run.py
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
import sys
import logging
from typing import Optional
from app.app import app
from loguru import logger
from fastapi import FastAPI
from types import FrameType
from multiprocessing import cpu_count
from gunicorn.glogging import Logger
from gunicorn.app.base import BaseApplication
LOG_LEVEL = "INFO"
WORKERS = int(cpu_count())
BIND = "0.0.0.0:8000"
class InterceptHandler(logging.Handler):
"""Intercept logs and forward them to Loguru.
Args:
logging.Handler (Filterer): Handler to filter logs
"""
def emit(self, record: logging.LogRecord):
"""Emit a log record."""
# Get corresponding Loguru level if it exists
level: str | int
frame: Optional[FrameType]
depth: int
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
# Find caller from where originated the logged message
frame, depth = logging.currentframe(), 2
while frame and frame.f_code.co_filename == logging.__file__:
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(
level, record.getMessage()
)
class StubbedGunicornLogger(Logger):
"""Defining a custom logger class to prevent gunicorn from logging to stdout
Args:
Logger (object): Gunicon logger class
"""
def setup(self, cfg):
"""Setup logger."""
handler: logging.NullHandler = logging.NullHandler()
self.error_logger: Logger = logging.getLogger("gunicorn.error")
self.error_logger.addHandler(handler)
self.access_logger: Logger = logging.getLogger("gunicorn.access")
self.access_logger.addHandler(handler)
self.error_logger.setLevel(LOG_LEVEL)
self.access_logger.setLevel(LOG_LEVEL)
class StandaloneApplication(BaseApplication):
"""Defines a Guicorn application
Args:
BaseApplication (object): Base class for Gunicorn applications
"""
def __init__(self, app: FastAPI, options: Optional[dict] = None):
"""Initialize the application
Args:
app (fastapi.FastAPI): FastAPI application
options (dict, optional): Gunicorn options. Defaults to None.
"""
self.options: dict = options or {}
self.application: FastAPI = app
super().__init__()
def load_config(self):
"""Load Gunicorn configuration."""
config: dict = {
key: value
for key, value in self.options.items()
if key in self.cfg.settings and value is not None
}
for key, value in config.items():
self.cfg.set(key.lower(), value)
def load(self) -> FastAPI:
"""Load the application
Returns:
FastAPI: FastAPI application
"""
return self.application
if __name__ == "__main__":
intercept_handler = InterceptHandler()
logging.root.setLevel(LOG_LEVEL)
seen: set = set()
for name in [
*logging.root.manager.loggerDict.keys(),
"gunicorn",
"gunicorn.access",
"gunicorn.error",
"uvicorn",
"uvicorn.access",
"uvicorn.error",
]:
if name not in seen:
seen.add(name.split(".")[0])
logging.getLogger(name).handlers = [intercept_handler]
logger.configure(handlers=[{"sink": sys.stdout, "serialize": False}])
options: dict = {
"bind": BIND,
"workers": WORKERS,
"accesslog": "-",
"errorlog": "-",
"worker_class": "uvicorn.workers.UvicornWorker",
"logger_class": StubbedGunicornLogger,
"preload": True,
"forwarded_allow_ips": "*",
"proxy_allow_ips": "*",
}
StandaloneApplication(app, options).run()