95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""Mokee Gateway 用户上下文中间件
|
||
|
||
网关转发请求时注入以下请求头，本中间件读取并存入 request.state.user：
|
||
X-User-Id, X-User-Name, X-User-Account, X-User-Email,
|
||
X-User-Post, X-User-Roles, X-System-Id, X-System-Code, X-System-Name
|
||
|
||
本地开发时（无网关），使用 DEV_MODE=true 环境变量或 X-Dev-Mock-User 请求头
|
||
启用默认 admin 用户，无需启动网关即可调试。
|
||
"""
|
||
|
||
import os
|
||
from starlette.middleware.base import BaseHTTPMiddleware
|
||
from fastapi import Request
|
||
|
||
# 网关注入的所有请求头
|
||
GATEWAY_HEADERS = [
|
||
"X-User-Id",
|
||
"X-User-Name",
|
||
"X-User-Account",
|
||
"X-User-Email",
|
||
"X-User-Post",
|
||
"X-User-Roles",
|
||
"X-System-Id",
|
||
"X-System-Code",
|
||
"X-System-Name",
|
||
]
|
||
|
||
# 本地开发默认用户（模拟网关注入）
|
||
DEV_DEFAULT_USER = {
|
||
"X-User-Id": "admin",
|
||
"X-User-Name": "管理员",
|
||
"X-User-Account": "admin",
|
||
"X-User-Email": "admin@zgitm.com",
|
||
"X-User-Post": "",
|
||
"X-User-Roles": '[{"roleId":"1","roleName":"管理员","roleCode":"admin"}]',
|
||
"X-System-Id": "1",
|
||
"X-System-Code": "mokeetext-edit",
|
||
"X-System-Name": "mokee编辑器",
|
||
}
|
||
|
||
|
||
class UserContext:
|
||
"""用户上下文 — 提供快捷访问方法"""
|
||
|
||
@staticmethod
|
||
def from_request(request: Request) -> dict:
|
||
"""从 request.state 获取用户信息"""
|
||
return getattr(request.state, "user", {})
|
||
|
||
@staticmethod
|
||
def get_user_id(request: Request) -> str:
|
||
"""获取当前用户账号名（用于 created_by 显示），未登录返回 'admin'"""
|
||
user = UserContext.from_request(request)
|
||
# 优先用登录账号名（如 admin），其次用用户名，最后用 ID
|
||
return user.get("X-User-Account") or user.get("X-User-Name") or user.get("X-User-Id", "admin")
|
||
|
||
@staticmethod
|
||
def get_user_name(request: Request) -> str:
|
||
return UserContext.from_request(request).get("X-User-Name", "")
|
||
|
||
@staticmethod
|
||
def get_user_account(request: Request) -> str:
|
||
return UserContext.from_request(request).get("X-User-Account", "")
|
||
|
||
@staticmethod
|
||
def get_system_code(request: Request) -> str:
|
||
return UserContext.from_request(request).get("X-System-Code", "")
|
||
|
||
|
||
class UserContextMiddleware(BaseHTTPMiddleware):
|
||
"""将网关注入的用户信息存入 request.state.user
|
||
|
||
生产环境：读取网关转发的 X-User-* 请求头
|
||
本地开发：DEV_MODE=true 时使用 DEV_DEFAULT_USER
|
||
"""
|
||
|
||
async def dispatch(self, request: Request, call_next):
|
||
user_info = {}
|
||
|
||
# 读取网关注入的请求头
|
||
for header in GATEWAY_HEADERS:
|
||
value = request.headers.get(header)
|
||
if value:
|
||
user_info[header] = value
|
||
|
||
# 本地开发：无网关时使用默认用户
|
||
if not user_info.get("X-User-Id"):
|
||
dev_mode = os.getenv("DEV_MODE", "false").lower() == "true"
|
||
if dev_mode or request.headers.get("X-Dev-Mock-User"):
|
||
user_info = DEV_DEFAULT_USER.copy()
|
||
|
||
request.state.user = user_info
|
||
response = await call_next(request)
|
||
return response
|