From 8ed800f083ad6f9518a05a0f73ca68d1b4517a7a Mon Sep 17 00:00:00 2001 From: Cxx0822 <1556464090@qq.com> Date: Sat, 12 Sep 2026 14:51:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0fluent=E6=97=A5?= =?UTF-8?q?=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/exceptions.py | 56 +++++++++++++++++++++++++ config/logging.py | 84 ++++++++++++++++++++++++++++++++++++++ main.py | 3 ++ requirements.txt | 4 +- service/comment_service.py | 2 + service/user_service.py | 4 +- 6 files changed, 150 insertions(+), 3 deletions(-) create mode 100644 config/exceptions.py create mode 100644 config/logging.py diff --git a/config/exceptions.py b/config/exceptions.py new file mode 100644 index 0000000..8257a19 --- /dev/null +++ b/config/exceptions.py @@ -0,0 +1,56 @@ +import time + +from fastapi import Request, HTTPException, status +from fastapi.responses import JSONResponse +from config.logging import logger + + +# 自定义异常类 +class AppException(Exception): + def __init__(self, message: str, details=None): + self.message = message + self.details = details + + +# 全局异常处理中间件 +async def global_exception_handler(request: Request, call_next): + try: + start_time = time.time() + + # 请求摘要 + logger.info( + f" {request.method} {request.url.path}" + ) + + response = await call_next(request) + + # 耗时 + time_taken = int((time.time() - start_time) * 1000) + logger.info( + f" Request URL {request.url.path} | Time Taken {time_taken} ms" + ) + + if response.status_code >= 400: + logger.warning(f" status={response.status_code}") + + return response + + except AppException as e: + # 记录业务异常 + logger.error(f" {e.message} | details={e.details}") + return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, + content={"message": e.message, "details": e.details}) + + except Exception as e: + # 记录未知异常（带堆栈信息） + logger.critical(f" {str(e)}", exc_info=True) + return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content={"code": 500, "message": "服务器内部错误", "details": str(e)}) + + +def get_credentials_exception() -> HTTPException: + return HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Could not validate credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) diff --git a/config/logging.py b/config/logging.py new file mode 100644 index 0000000..9279032 --- /dev/null +++ b/config/logging.py @@ -0,0 +1,84 @@ +import atexit +import os +import sys + +from dotenv import load_dotenv +from fluent import sender +from loguru import logger + +load_dotenv() + +# 日志格式 +STDOUT_FORMAT = ( + "{time:YYYY-MM-DD HH:mm:ss.SSS} | " + "{level: <8} | " + "{name}:{function}:{line} - " + "{message}" +) + +FILE_FORMAT = ( + "{time:YYYY-MM-DD HH:mm:ss.SSS} | " + "{level: <8} | " + "{name}:{function}:{line} - {message}" +) + +fluent_sender = sender.FluentSender( + tag=os.getenv("FLUENTD_TOPIC"), + host=os.getenv("FLUENTD_HOST"), + port=int(os.getenv("FLUENTD_PORT", 24224)), + buffer_max_size=8 * 1024 * 1024, + timeout=3.0, + retry_timeout=60 +) + + +def log_to_fluent(message): + try: + record = message.record + + # 构建结构化日志数据 + log_data = { + 'topic': os.getenv("FLUENTD_TOPIC"), + 'timestamp': record['time'].timestamp(), + 'level': record['level'].name.lower(), + 'message': record['message'], + 'source': f"{record['file'].path}:{record['line']}", + 'module': record['module'], + 'function': record['function'], + 'process_id': record['process'].id, + 'thread_id': record['thread'].id, + **record['extra'] + } + + if not fluent_sender.emit(os.getenv("FLUENTD_TOPIC"), log_data): + print(f"Fluentd 发送失败: {fluent_sender.last_error}") + + except Exception as e: + print(f"日志处理异常: {str(e)}") + + +# 移除默认处理器 +logger.remove() + +# 添加控制台处理器 +logger.add( + sink=sys.stdout, + level="INFO", + format=STDOUT_FORMAT, + colorize=True, + backtrace=True, # 显示完整异常堆栈 + diagnose=True, # 显示详细异常信息 +) + +logger.add( + log_to_fluent, + level="INFO", # 处理 INFO 及以上级别 + format="{message}", # 原始消息（实际使用结构化数据） + backtrace=True, # 启用堆栈回溯 + diagnose=True # 显示诊断信息 +) + +atexit.register(fluent_sender.close) + +# 导出配置好的logger +__all__ = ["logger"] diff --git a/main.py b/main.py index 4087702..45fd4a2 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,7 @@ from fastapi import FastAPI from starlette.middleware.cors import CORSMiddleware +from config.exceptions import global_exception_handler from routers import routers app = FastAPI(title="Family Service") @@ -12,5 +13,7 @@ app.add_middleware( allow_headers=["*"], ) +app.middleware("http")(global_exception_handler) + for router in routers: app.include_router(router) diff --git a/requirements.txt b/requirements.txt index 0738711..1279d2c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,4 +9,6 @@ uvicorn~=0.23.0 pymysql~=1.2.0 boto3~=1.40.59 botocore~=1.40.59 -python-multipart~=0.0.20 \ No newline at end of file +python-multipart~=0.0.20 +fluent-logger~=0.10.0 +loguru~=0.7.3 \ No newline at end of file diff --git a/service/comment_service.py b/service/comment_service.py index 37ea9fa..0c1e557 100644 --- a/service/comment_service.py +++ b/service/comment_service.py @@ -8,6 +8,7 @@ from service.record_service import get_record_by_id def get_comment_by_id(db: Session, comment_id: int) -> Comments | None: result = db.execute(select(Comments).where(Comments.id == comment_id)) + return result.scalar_one_or_none() @@ -33,6 +34,7 @@ def update_comment(db: Session, comment_id: int, comment_dto: CommentUpdate) -> comment.content = comment_dto.content db.commit() db.refresh(comment) + return True diff --git a/service/user_service.py b/service/user_service.py index 191813f..252c79e 100644 --- a/service/user_service.py +++ b/service/user_service.py @@ -19,10 +19,10 @@ def create_user(db: Session, user: UserCreate) -> Users: return user -def update_user(db: Session, user_id: int, obj_in: UserUpdate) -> Users: +def update_user(db: Session, user_id: int, obj_in: UserUpdate) -> Users | None: user = get_user_by_id(db, user_id) if not user: - return False + return None for field, value in obj_in.model_dump(exclude_unset=True).items(): setattr(user, field, value)