浅念
f22786d746
auth.py 的 _EnvelopeMixin/_ok 提取到 tracker/api/envelope.py（去下划线前缀）， auth.py 改为 import 使用；collect.py 的 finalize_response 含 ACAO 回显逻辑 保留在自身视图，仅注释指向未来收敛。 Co-Authored-By: Claude Code <noreply@anthropic.com>
61 lines
3.0 KiB
Python
61 lines
3.0 KiB
Python
import re
|
||
from django.contrib.auth import get_user_model
|
||
from rest_framework.views import APIView
|
||
from rest_framework.response import Response
|
||
from rest_framework.permissions import AllowAny, IsAdminUser
|
||
from rest_framework.throttling import ScopedRateThrottle
|
||
from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
|
||
from tracker.api.envelope import EnvelopeMixin, ok
|
||
|
||
# 对齐 Django UnicodeUsernameValidator（\w 默认 Unicode）——create_user 不跑校验器，入口自防
|
||
_USERNAME_RE = re.compile(r"[\w.@+-]+")
|
||
|
||
|
||
class LoginView(EnvelopeMixin, TokenObtainPairView):
|
||
"""登录。JWT 取舍（M1 决策）：access 12h / refresh 24h、无黑名单轮换——
|
||
token 泄露唯一应急是换 SECRET_KEY（全体登出），README-deploy 写明。"""
|
||
throttle_classes = [ScopedRateThrottle]
|
||
throttle_scope = "login" # 10/min/IP 防爆破（settings 已配）
|
||
|
||
def post(self, request, *args, **kwargs):
|
||
resp = super().post(request, *args, **kwargs) # 失败由 simplejwt 返回 401（信封由 mixin 补）
|
||
if resp.status_code == 200:
|
||
return ok({"access": resp.data["access"], "refresh": resp.data["refresh"]})
|
||
return resp
|
||
|
||
|
||
class RefreshView(EnvelopeMixin, TokenRefreshView):
|
||
permission_classes = [AllowAny] # TokenRefreshView 自带 AllowAny，显式声明表意
|
||
|
||
def post(self, request, *args, **kwargs):
|
||
resp = super().post(request, *args, **kwargs)
|
||
if resp.status_code == 200: # 成功同样走信封——与 LoginView 对齐，前端只需一种解析路径
|
||
return ok({"access": resp.data["access"]})
|
||
return resp
|
||
|
||
|
||
class MeView(EnvelopeMixin, APIView):
|
||
def get(self, request):
|
||
return ok({"username": request.user.username,
|
||
"is_admin": request.user.is_staff})
|
||
|
||
|
||
class UserCreateView(EnvelopeMixin, APIView):
|
||
permission_classes = [IsAdminUser] # is_staff
|
||
|
||
def post(self, request):
|
||
data = request.data if isinstance(request.data, dict) else {} # JSON 数组等畸形 body → 空表（.get 会 500）
|
||
username, password = data.get("username"), data.get("password")
|
||
if not isinstance(username, str) or not isinstance(password, str): # 123.strip() / len(None) 会 500
|
||
return Response({"code": 400, "data": None, "message": "invalid_input"}, status=400)
|
||
username = username.strip()
|
||
# 字符集对齐 UnicodeUsernameValidator（create_user 不跑校验器）；长度上限防 MySQL DataError（sqlite 不报错）
|
||
if (not username or len(username) > 150 or not _USERNAME_RE.fullmatch(username)
|
||
or len(password) < 8):
|
||
return Response({"code": 400, "data": None, "message": "invalid_input"}, status=400)
|
||
User = get_user_model()
|
||
if User.objects.filter(username=username).exists():
|
||
return Response({"code": 409, "data": None, "message": "username_exists"}, status=409)
|
||
User.objects.create_user(username=username, password=password, is_staff=False)
|
||
return ok({"username": username}, "created")
|