update. 使用django-ninja-jwt作为Token管理
This commit is contained in:
+3
-1
@@ -11,8 +11,9 @@ https://docs.djangoproject.com/en/6.0/ref/settings/
|
||||
"""
|
||||
import os
|
||||
from pathlib import Path
|
||||
from loguru import logger
|
||||
|
||||
import environ
|
||||
from loguru import logger
|
||||
|
||||
env = environ.Env()
|
||||
|
||||
@@ -76,6 +77,7 @@ INSTALLED_APPS = [
|
||||
'django.contrib.sessions',
|
||||
'django.contrib.messages',
|
||||
'django.contrib.staticfiles',
|
||||
'ninja_extra',
|
||||
'import_export',
|
||||
'django_ckeditor_5',
|
||||
'base.apps.BaseConfig',
|
||||
|
||||
+26
-70
@@ -14,83 +14,40 @@ Including another URLconf
|
||||
1. Import the include() function: from django.urls import include, path
|
||||
2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))
|
||||
"""
|
||||
import importlib
|
||||
import pkgutil
|
||||
import traceback
|
||||
|
||||
from django.conf import settings
|
||||
from django.conf.urls.static import static
|
||||
from django.contrib import admin
|
||||
from django.contrib.staticfiles.urls import staticfiles_urlpatterns
|
||||
from django.urls import path, include, re_path
|
||||
|
||||
import logging
|
||||
import os
|
||||
import pkgutil
|
||||
import traceback
|
||||
import importlib
|
||||
|
||||
import jwt
|
||||
from django.contrib.auth.models import User
|
||||
from django.http import JsonResponse
|
||||
from jwt import exceptions
|
||||
from ninja import NinjaAPI
|
||||
from django.urls import path, include
|
||||
from loguru import logger
|
||||
from ninja.errors import ValidationError
|
||||
from ninja.security import HttpBearer
|
||||
from ninja_extra import NinjaExtraAPI
|
||||
from ninja_jwt.controller import NinjaJWTDefaultController
|
||||
|
||||
from base.views import custom_upload_file
|
||||
from common.common_response import CommonResponse
|
||||
from .settings import SECRET_KEY
|
||||
|
||||
logger = logging.getLogger('ptools')
|
||||
|
||||
|
||||
class GlobalAuth(HttpBearer):
|
||||
def authenticate(self, request, token):
|
||||
try:
|
||||
logger.debug('用户认证中')
|
||||
logger.debug(f'解析 Auth Token：{token}')
|
||||
res = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
|
||||
user = User.objects.filter(id=res.get('id'), username=res.get('username')).first()
|
||||
logger.debug(user)
|
||||
if not user:
|
||||
raise exceptions.PyJWKError('用户不存在')
|
||||
request.user = user
|
||||
return token
|
||||
except exceptions.PyJWKError:
|
||||
msg = 'Token 认证失败，请重新登陆！'
|
||||
logger.error(msg)
|
||||
raise
|
||||
except exceptions.ExpiredSignatureError:
|
||||
msg = '认证令牌已过期，请重新登陆！'
|
||||
logger.error(msg)
|
||||
raise
|
||||
except Exception as e:
|
||||
msg = '认证失败，请检查用户是否存在'
|
||||
logger.error(msg)
|
||||
raise
|
||||
|
||||
|
||||
# 2. 获取当前文件所在的目录路径 (假设此文件与各个 app 文件夹同级，或者在特定配置目录下)
|
||||
# 如果你的此文件在项目根目录，而 app 文件夹也在根目录：
|
||||
from django.conf import settings
|
||||
|
||||
project_root = settings.BASE_DIR
|
||||
|
||||
|
||||
# 如果 app 文件夹是 Django 的已安装应用，可能需要通过 settings.INSTALLED_APPS 获取，
|
||||
# 但通常直接扫描物理目录更灵活。这里假设扫描当前目录下的所有子文件夹。
|
||||
|
||||
# 3. 自动发现并注册路由
|
||||
def auto_load_routers(base_path, package_name_prefix=None):
|
||||
def auto_load_routers(package_name_prefix=None):
|
||||
"""
|
||||
base_path: 物理路径，例如 /path/to/project
|
||||
package_name_prefix: 包名前缀，例如 'myproject.' (如果这些文件夹是包的一部分)
|
||||
"""
|
||||
|
||||
base_path = settings.BASE_DIR
|
||||
# 获取该路径下所有子模块/包
|
||||
# 注意：这需要这些文件夹是合法的 Python 包 (即包含 __init__.py)，或者在 Python 路径中
|
||||
for importer, modname, ispkg in pkgutil.iter_modules([base_path]):
|
||||
for importer, modname, is_pkg in pkgutil.iter_modules([base_path]):
|
||||
# 过滤条件：
|
||||
# 1. 必须是包 (文件夹)
|
||||
# 2. 不以 '_' 开头 (排除 __pycache__, _private 等)
|
||||
# 3. 排除当前文件所在的脚本名 (防止递归或错误)
|
||||
if ispkg and not modname.startswith('_'):
|
||||
if is_pkg and not modname.startswith('_'):
|
||||
|
||||
# 构造完整的模块路径：例如 "authorize.views"
|
||||
# 如果这些文件夹直接位于根路径且不是大包的一部分，package_name_prefix 可能为空
|
||||
@@ -126,8 +83,19 @@ def auto_load_routers(base_path, package_name_prefix=None):
|
||||
print(f"❌ 注册失败 {modname}: {e}")
|
||||
|
||||
|
||||
api_v1 = NinjaAPI(version='1.0.0', auth=GlobalAuth())
|
||||
auto_load_routers(project_root)
|
||||
class UnifiedNinjaAPI(NinjaExtraAPI):
|
||||
def create_response(self, request, data, *, status=None, headers=None, temporal_response=None):
|
||||
if not isinstance(data, CommonResponse):
|
||||
if isinstance(data, tuple) and len(data) == 2:
|
||||
data, status = data
|
||||
data = CommonResponse.success(data=data)
|
||||
return super().create_response(request, data, status=status, temporal_response=temporal_response)
|
||||
|
||||
|
||||
api_v1 = UnifiedNinjaAPI(version='1.0.0')
|
||||
api_v1.register_controllers(NinjaJWTDefaultController)
|
||||
|
||||
auto_load_routers()
|
||||
|
||||
|
||||
@api_v1.exception_handler(ValidationError)
|
||||
@@ -138,18 +106,6 @@ def validation_errors(request, exc):
|
||||
return JsonResponse(data=CommonResponse.error(msg=f"输入参数验证失败！").to_dict(), status=422, safe=False)
|
||||
|
||||
|
||||
@api_v1.exception_handler(exceptions.PyJWKError) # 捕获 HttpBearer 认证异常
|
||||
def custom_http_bearer_exception_handler(request, exc):
|
||||
logger.error(exc)
|
||||
return JsonResponse(data=CommonResponse.error(msg=f"用户Token认证失败：{exc}").to_dict(), status=401)
|
||||
|
||||
|
||||
@api_v1.exception_handler(exceptions.ExpiredSignatureError) # 捕获 HttpBearer 认证异常
|
||||
def custom_http_bearer_exception_handler(request, exc):
|
||||
logger.error(exc)
|
||||
return JsonResponse(data=CommonResponse.error(msg=f"用户登录信息已过期，请重新登录！！").to_dict(), status=401)
|
||||
|
||||
|
||||
urlpatterns = [
|
||||
path('', include("front.urls")),
|
||||
path('admin/', admin.site.urls),
|
||||
|
||||
Reference in New Issue
Block a user