You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
65 lines
3.2 KiB
Python
65 lines
3.2 KiB
Python
import logging
|
|
import logging.handlers
|
|
import json
|
|
import datetime
|
|
import os
|
|
import sys
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
except ImportError:
|
|
from backports.zoneinfo import ZoneInfo
|
|
from src.config import LOG_LEVEL, SERVICE_NAME
|
|
from src.enums import OptimumOHEnum
|
|
_TZ_LOCAL = ZoneInfo('Asia/Jakarta')
|
|
RESET = '\x1b[0m'
|
|
COLORS = {'DEBUG': '\x1b[36m', 'INFO': '\x1b[32m', 'WARNING': '\x1b[33m', 'WARN': '\x1b[33m', 'ERROR': '\x1b[31m', 'CRITICAL': '\x1b[1;31m'}
|
|
_REQUEST_FIELDS = frozenset({'method', 'path', 'status_code', 'duration_ms', 'request_id', 'user_id', 'user_ip', 'error_id', 'result', 'error_type', 'error_message'})
|
|
|
|
class LogLevels(OptimumOHEnum):
|
|
info = 'INFO'
|
|
warn = 'WARN'
|
|
error = 'ERROR'
|
|
debug = 'DEBUG'
|
|
|
|
class JSONFormatter(logging.Formatter):
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
dt = datetime.datetime.fromtimestamp(record.created, tz=_TZ_LOCAL)
|
|
timestamp = dt.isoformat(timespec='microseconds')
|
|
status_code = getattr(record, 'status_code', None)
|
|
result = getattr(record, 'result', None)
|
|
if result is None and status_code is not None:
|
|
result = 'Request completed' if int(status_code) < 400 else 'Request failed'
|
|
log_record = {'timestamp': timestamp, 'service_name': SERVICE_NAME, 'level': record.levelname, 'method': getattr(record, 'method', None), 'path': getattr(record, 'path', None), 'status_code': status_code, 'duration_ms': getattr(record, 'duration_ms', None), 'request_id': getattr(record, 'request_id', None), 'user_id': getattr(record, 'user_id', None), 'user_ip': getattr(record, 'user_ip', None), 'error_id': getattr(record, 'error_id', None), 'result': result, 'error_type': getattr(record, 'error_type', None), 'error_message': getattr(record, 'error_message', None)}
|
|
log_json = json.dumps(log_record, default=str)
|
|
if sys.stdout.isatty():
|
|
color = COLORS.get(record.levelname, '')
|
|
return f'{color}{log_json}{RESET}'
|
|
return log_json
|
|
|
|
def configure_logging() -> None:
|
|
log_level = str(LOG_LEVEL).upper()
|
|
if log_level not in list(LogLevels):
|
|
log_level = LogLevels.error
|
|
root_logger = logging.getLogger()
|
|
root_logger.setLevel(log_level)
|
|
if root_logger.hasHandlers():
|
|
root_logger.handlers.clear()
|
|
formatter = JSONFormatter()
|
|
stream_handler = logging.StreamHandler(sys.stdout)
|
|
stream_handler.setFormatter(formatter)
|
|
root_logger.addHandler(stream_handler)
|
|
os.makedirs('logs', exist_ok=True)
|
|
file_handler = logging.handlers.RotatingFileHandler('logs/app.log', maxBytes=10 * 1024 * 1024, backupCount=5, encoding='utf-8')
|
|
file_handler.setFormatter(formatter)
|
|
root_logger.addHandler(file_handler)
|
|
for logger_name in ['uvicorn', 'uvicorn.access', 'uvicorn.error', 'fastapi']:
|
|
uvicorn_logger = logging.getLogger(logger_name)
|
|
uvicorn_logger.handlers = []
|
|
uvicorn_logger.propagate = True
|
|
logging.getLogger('uvicorn.access').setLevel(logging.WARNING)
|
|
slowapi_logger = logging.getLogger('slowapi')
|
|
slowapi_logger.setLevel(logging.ERROR)
|
|
slowapi_logger.propagate = False
|
|
logging.getLogger('slack_sdk.web.base_client').setLevel(logging.CRITICAL)
|