60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
|
|
from fastapi import Request
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from sqlalchemy.orm import Session
|
|
from app.db.models import OperationLog,User
|
|
|
|
from functools import wraps
|
|
from fastapi import Request
|
|
from contextvars import ContextVar
|
|
import json
|
|
|
|
class LoggingMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request: Request, call_next):
|
|
if request.method in ("POST", "PUT", "PATCH","DELETE"):
|
|
try:
|
|
request.state.body = await request.json()
|
|
except json.JSONDecodeError:
|
|
request.state.body = await request.body()
|
|
else:
|
|
request.state.body = None
|
|
|
|
try:
|
|
response = await call_next(request)
|
|
state = request.state
|
|
except Exception as e:
|
|
await self.log_error(request, e)
|
|
raise
|
|
|
|
if hasattr(request.state, "user") and hasattr(request.state, "tenant"):
|
|
await self.log_request(request, response,state)
|
|
|
|
return response
|
|
|
|
async def log_request(self, request: Request, response,state):
|
|
try:
|
|
headers = dict(request.headers)
|
|
|
|
db_operation = OperationLog(tenantid =request.state.tenant,
|
|
clientip = request.client.host if request.client else None,
|
|
useragent =headers.get("user-agent", ""),
|
|
userid = request.state.user.id,
|
|
operation = request.method,
|
|
function = request.url.path,
|
|
response = f"status_code:{response.status_code }" )
|
|
|
|
db = request.state.db
|
|
if db:
|
|
await self.write_log_to_db(db_operation,db)
|
|
except Exception as e:
|
|
print(f"Logging failed: {str(e)}")
|
|
|
|
|
|
async def log_error(self, request: Request, exc: Exception):
|
|
# 错误处理逻辑
|
|
pass
|
|
|
|
async def write_log_to_db(self, db_operation,db):
|
|
db.add(db_operation)
|
|
db.commit()
|
|
|