Django开发记录
约 4085 字大约 14 分钟
2025-10-02
跨域问题
浏览器跨域(Cross-Origin Resource Sharing, CORS)是浏览器实施的一种安全机制,它限制从一个源(域、协议、端口)加载的网页脚本与来自另一个源的资源进行交互。这是为了防止恶意网站窃取数据。
同源策略要求三个一致:
- 协议相同(http/https)
- 域名相同
- 端口相同
Django跨域问题表现
在Django开发中,跨域问题通常会表现为:
浏览器控制台出现
CORS错误消息:Access to fetch at 'http://api.example.com' from origin 'http://localhost:3000' has been blocked by CORS policy前端JavaScript代码无法成功向不同源的Django后端发送请求
预检请求(
OPTIONS)失败,返回403或405状态码
跨域问题解决方案
安装
django-cors-headers:pip install django-cors-headers配置
settings.py:添加
'corsheaders'到INSTALLED_APPS添加
'corsheaders.middleware.CorsMiddleware'到MIDDLEWARE的开头添加
'CORS_ORIGIN_ALLOW_ALL = True'允许全部跨域配置
CORS_ALLOWED_ORIGINS设置允许的源(可选)设置
CORS_ALLOW_CREDENTIALS为True如果需要传递Cookie(可选)配置
CORS_ALLOW_METHODS指定允许的HTTP方法(可选)配置
CORS_ALLOW_HEADERS指定允许的HTTP头

json序列化
手动序列化
# models.py
class MainMenu(models.Model):
main_menu_id = models.IntegerField()
main_menu_name = models.CharField(max_length=255)
main_menu_url = models.CharField(max_length=255, blank=True, null=True)
# 重写__str__方法, 新建一个字典,将所有字段都放在字典中,将字典传出
def __str__(self):
result = {
"main_menu_id": self.main_menu_id,
"main_menu_name": self.main_menu_name,
"main_menu_url": self.main_menu_url,
}
return json.dumps(result, ensure_ascii=False)
class Meta:
managed = False
db_table = 'main_menu'# views.py
class GoodsMainMenu(View):
def get(self, request):
# 获取MainMenu表中的所有行
main_menu = MainMenu.objects.all()
result_list = []
# 对每一行数据调用序列化方法, 转换成json格式
for m in main_menu:
result_list.append(m.__str__())
return HttpResponse(result_list)DRF序列化器
decimal类型序列化
在数据库中有表示价格的字段在json转换时会报以下错误:
TypeError: object of type Decimal is not JSON serializable- 方式一:手动对
decimal格式的数据做json转换
# models.py
class Goods(models.Model):
type_id = models.IntegerField(blank=True, null=True)
name = models.CharField(max_length=255, blank=True, null=True)
p_price = models.DecimalField(max_digits=10, decimal_places=2, blank=True, null=True)
create_time = models.DateTimeField(blank=True, null=True)
def __str__(self):
result = {
'type_id': self.type_id,
'name': self.name,
'p_price': self.p_price,
'create_time': self.create_time,
}
# `cls=DecimalEncoder` 用于转换decimal格式数据
return json.dumps(result, cls=DecimalEncoder, ensure_ascii=False)
class Meta:
managed = False
db_table = 'goods'
# 针对decimal格式做json转换(固定写法)
class DecimalEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o, Decimal):
return float(o)
return None# views.py
class GoodsCategoryAPIView(APIView):
def get(self, request):
category_data = Goods.objects.all()
result_list = []
for c in category_data:
result_list.append(c.__str__())
return HttpResponse(result_list)- 方式二:使用DRF序列化器
# serializers.py
class GoodsSerializer(serializers.ModelSerializer):
# 这里写的字段是想要进行自定义处理的字段
p_price = serializers.SerializerMethodField()
# 自定义序列化方法时,方法名必须是 get_<字段名>
def get_p_price(self, obj):
return float(obj.p_price)
class Meta:
model = Goods
fields = '__all__'# views.py
class GoodsCategoryAPIView(APIView):
def get(self, request):
category_data = Goods.objects.all()
# 序列化时参数是instance , 反序列化时参数是data
result = GoodsSerializer(instance=category_data)
return HttpResponse(result)datetime类型序列化
# serializers.py
class 模型名Serializer(serializers.ModelSerializer):
# 自定义时间格式
create_time = serializers.DateTimeField(format='%Y-%m-%d %H:%M:%S', read_only=True)
class Meta:
model = 模型名
fields = '__all__'# views.py
class 视图类名(APIView):
def get(self, request):
data = 模型名.objects.all()
# 序列化时参数是instance , 反序列化时参数是data
result = 模型名Serializer(instance=data)
return HttpResponse(result)统一Response
from django.http import JsonResponse
from rest_framework import status
from typing import Any, Dict, Optional, List
from django.utils.encoding import force_str
class APIResponse:
"""
通用API响应类
使用 force_str 确保所有消息都被正确转换为字符串,支持国际化
使用示例:return APIResponse.success(data={'id': 1}, message='操作成功')
响应示例:
{
"success": true,
"code": 200,
"message": "Success",
"data": {}
}
"""
@classmethod
def success(
cls,
data: Any = None,
message: str = "Success",
status_code: int = status.HTTP_200_OK
) -> JsonResponse:
"""
成功响应
"""
response_data = {
"success": True,
"code": status_code,
"message": force_str(message), # 使用 force_str 处理翻译
"data": data if data is not None else {}
}
return JsonResponse(response_data, status=status_code)
@classmethod
def created(
cls,
data: Any = None,
message: str = "Resource created successfully",
status_code: int = status.HTTP_201_CREATED
) -> JsonResponse:
"""
创建成功响应
"""
return cls.success(data, force_str(message), status_code) # 传递前先转换
@classmethod
def error(
cls,
message: str = "Error occurred",
status_code: int = status.HTTP_400_BAD_REQUEST,
errors: Optional[List[Dict]] = None
) -> JsonResponse:
"""
错误响应
"""
response_data = {
"success": False,
"code": status_code,
"message": force_str(message), # 使用 force_str 处理翻译
"errors": errors if errors else []
}
return JsonResponse(response_data, status=status_code)
@classmethod
def not_found(
cls,
message: str = "Resource not found",
status_code: int = status.HTTP_404_NOT_FOUND,
errors: Optional[List[Dict]] = None
) -> JsonResponse:
"""
资源未找到响应
"""
return cls.error(force_str(message), status_code, errors) # 传递前先转换
@classmethod
def validation_error(
cls,
message: str = "Validation error",
errors: Optional[List[Dict]] = None,
status_code: int = status.HTTP_422_UNPROCESSABLE_ENTITY
) -> JsonResponse:
"""
验证错误响应
"""
return cls.error(force_str(message), status_code, errors) # 传递前先转换
@classmethod
def unauthorized(
cls,
message: str = "Unauthorized access",
status_code: int = status.HTTP_401_UNAUTHORIZED,
errors: Optional[List[Dict]] = None
) -> JsonResponse:
"""
未授权响应
"""
return cls.error(force_str(message), status_code, errors) # 传递前先转换
@classmethod
def forbidden(
cls,
message: str = "Forbidden access",
status_code: int = status.HTTP_403_FORBIDDEN,
errors: Optional[List[Dict]] = None
) -> JsonResponse:
"""
禁止访问响应
"""
return cls.error(force_str(message), status_code, errors) # 传递前先转换
@classmethod
def server_error(
cls,
message: str = "Internal server error",
status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
errors: Optional[List[Dict]] = None
) -> JsonResponse:
"""
服务器错误响应
"""
return cls.error(force_str(message), status_code, errors) # 传递前先转换
@classmethod
def paginated(
cls,
data: List[Any],
total: int,
page: int,
page_size: int,
message: str = "Success",
status_code: int = status.HTTP_200_OK
) -> JsonResponse:
"""
分页响应
"""
response_data = {
"success": True,
"code": status_code,
"message": force_str(message), # 使用 force_str 处理翻译
"data": {
"items": data,
"pagination": {
"total": total,
"page": page,
"page_size": page_size,
"total_pages": (total + page_size - 1) // page_size if page_size > 0 else 0
}
}
}
return JsonResponse(response_data, status=status_code)
@classmethod
def custom(
cls,
success: bool,
code: int,
message: str,
data: Optional[Any] = None,
errors: Optional[List[Dict]] = None
) -> JsonResponse:
"""
自定义响应
"""
response_data = {
"success": success,
"code": code,
"message": force_str(message), # 使用 force_str 处理翻译
}
if data is not None:
response_data["data"] = data
if errors is not None:
response_data["errors"] = errors
else:
response_data["errors"] = []
return JsonResponse(response_data, status=code)密码加密
Django自带的密码处理框架可以提供密码加密和密码验证,Django 中用于密码处理的核心函数和方法
密码加密:
make_password(plain_password)这个函数将一个明文密码转换成一个符合Django标准的哈希字符串
from django.contrib.auth.hashers import make_password plain_password = "my123password" hashed_password = make_password(plain_password) print(hashed_password) # 输出类似:pbkdf2_sha256$600000$G3RcJbH2J6Bz6U8X$Xx4V1Z... # 格式:算法$迭代次数$盐$哈希值每次调用 make_password,即使密码相同,也会生成不同的哈希值。这是因为 Django 会自动生成一个随机的“盐”(salt),与密码一起进行哈希,这能有效防止彩虹表攻击。
验证密码:
check_password(plain_password, hashed_password)这个函数用于验证用户输入的明文密码是否与数据库中存储的哈希密码匹配。
from django.contrib.auth.hashers import check_password # 假设从数据库取出的哈希密码是: stored_hashed_password = "pbkdf2_sha256$600000$G3RcJbH2J6Bz6U8X$Xx4V1Z..." # 用户登录时输入的密码 user_input_password = "my123password" # 验证 is_correct = check_password(user_input_password, stored_hashed_password) print(is_correct) # 如果密码正确,输出 True;错误则输出 False
在DRF的序列化器中实现加密操作:
# serializers.py
import datetime
from rest_framework import serializers
from rest_framework.validators import UniqueValidator
from apps.user.models import User
from apps.user.pwd_encoder import get_md5
class UserSerializer(serializers.ModelSerializer):
# email作为账户名, 需要唯一性验证
email = serializers.EmailField(
# 必需字段
required=True,
# 不允许空
allow_null=False,
# 唯一性校验
validators=[UniqueValidator(queryset=User.objects.all(), message="用户已存在")]
)
# 时间格式化
birthday = serializers.DateTimeField('%Y-%m-%d %H:%M:%S')
create_time = serializers.DateTimeField('%Y-%m-%d %H:%M:%S', required=False)
# 在调用save()时自动被调用,在数据存储之前对数据进行加工
def create(self, validated_data):
# 密码md5加密
validated_data['password'] = get_md5(validated_data['password'])
# 取当前时间为注册时间
validated_data['create_time'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
result = User.objects.create(**validated_data)
return result
class Meta:
model = User
fields = "__all__"
# 创建用户后密码不返回给前端
extra_kwargs = {'password': {'write_only': True}}用户模块
用户模型类
Django内置了一个用户模型类 django.contrib.auth.models.User ,但当实际开发场景中,我们往往在models.py中自定义 User 模型。
当第一次运行migrate命令时,Django会自动在数据库中创建内置 User 模型对应的表(通常是auth_user), 此时,如果你在models.py中创建了一个同名的 User 模型,将会导致冲突,因为两个模型都试图创建名为User的模型类。
正确的做法是,在项目开始之初,在创建任何数据库迁移之前,就创建一个自定义用户模型,并告诉Django使用你的模型来代替它内置的模型。这样,Django的内置User模型就不会被使用,所有的认证、权限等系统都会基于你的自定义模型工作。
创建自定义用户模型
Django官方推荐的方法是使用抽象基类继承,具体是继承AbstractUser或AbstractBaseUser。
继承 AbstractUser
AbstractUser 类包含了完整的用户模型实现,主要有以下属性和方法:
个人信息字段:
username,password,email,first_name,last_name权限相关字段:
is_staff,is_active,is_superuser时间戳字段:
date_joined,last_login多对多关系字段:
groups,user_permissions重要类属性:
# 指定哪个字段作为唯一标识(默认是username) USERNAME_FIELD = 'username' # 创建超级用户时需要提供的字段(除了USERNAME_FIELD和password) REQUIRED_FIELDS = ['email'] # 指定用于管理用户的管理器 objects = UserManager() # 用于指定邮箱字段(在密码重置等功能中使用) EMAIL_FIELD = 'email'认证相关方法:
get_username(): 返回用户名set_password(raw_password): 设置密码(自动加密)check_password(raw_password): 检查密码是否正确set_unusable_password(): 设置不可用密码(用于第三方认证)has_usable_password(): 检查是否有可用密码
权限相关方法
get_user_permissions(obj=None): 返回用户直接拥有的权限(非通过组)get_group_permissions(obj=None): 返回用户通过组获得的权限get_all_permissions(obj=None): 返回用户的所有权限(直接+通过组)has_perm(perm, obj=None): 检查用户是否有特定权限has_perms(perm_list, obj=None): 检查用户是否有所有指定权限has_module_perms(app_label): 检查用户是否有某个app的权限
其他实用方法
get_full_name(): 返回用户全名get_short_name(): 返回用户短名(通常是first_name)email_user(subject, message, form_email=None, **kwargs): 发送邮件给用户
管理器方法:
# 创建普通用户(自动加密密码) User.objects.create_user(username, email=None, password=None, **extra_fields) # 创建超级用户 User.objects.create_superuser(username, email=None, password=None, **extra_fields) # 根据用户名获取用户(用于认证后端) User.objects.get_by_natural_key(username) # 使用手机号创建用户(如果自定义了USERNAME_FIELD) User.objects.create_user(phone_number='13800138000', password='xxx')
使用示例:
在models.py中创建模型:
from django.contrib.auth.models import AbstractUser, BaseUserManager from django.db import models from django.utils.translation import gettext_lazy as _ class CustomUserManager(BaseUserManager): """自定义用户管理器,支持邮箱和手机号作为用户名""" def create_user(self, email, mobile, password=None, **extra_fields): """创建普通用户""" if not email and not mobile: raise ValueError(_('必须提供邮箱或手机号')) email = self.normalize_email(email) if email else None user = self.model(email=email, mobile=mobile, **extra_fields) user.set_password(password) user.save(using=self._db) return user def create_superuser(self, email, mobile, password=None, **extra_fields): """创建超级用户""" extra_fields.setdefault('is_staff', True) extra_fields.setdefault('is_superuser', True) extra_fields.setdefault('is_active', True) if extra_fields.get('is_staff') is not True: raise ValueError(_('超级用户必须设置 is_staff=True')) if extra_fields.get('is_superuser') is not True: raise ValueError(_('超级用户必须设置 is_superuser=True')) return self.create_user(email, mobile, password, **extra_fields) class User(AbstractUser): """自定义用户模型""" username = None # 移除默认的 username 字段 email = models.EmailField(_('邮箱地址'), unique=True, blank=True, null=True) mobile = models.CharField(_('手机号'), max_length=11, unique=True, blank=True, null=True) # 设置认证字段为邮箱或手机号 USERNAME_FIELD = 'email' REQUIRED_FIELDS = ['mobile'] # 创建超级用户时需要填写的字段 objects = CustomUserManager() class Meta: db_table = 'user' # 可以自定义表名,也可以不写使用默认表名"auth_user" verbose_name = _('用户') verbose_name_plural = _('用户') def __str__(self): return self.email or self.mobile在
settings.py中指定自定义模型:这是最关键的一步,告诉Django你现在用哪个模型作为默认的用户模型。
# project/settings.py AUTH_USER_MODEL = 'myapp.User' # 格式:'应用名.模型名'执行数据库迁移:
python manage.py makemigrations python manage.py migrate在其它模型中引用用户模型:
from django.contrib.auth import get_user_model User = get_user_model() # 这会获取到当前激活的用户模型,即你的自定义模型 user = User.objects.get(id=1)
重要提醒
设置
AUTH_USER_MODEL和创建自定义用户模型的操作,务必在执行第一次makemigrations和migrate命令之前完成。如果已经创建了数据库表,再做这件事会非常麻烦。在任何地方需要引用
User模型时,养成使用get_user_model()或settings.AUTH_USER_MODEL的习惯。
继承 AbstractBaseUser(用于高度定制)
这种方式只包含最核心的认证信息(密码、最后一次登录时间等),你需要从头定义所有字段(如用户名、邮箱等),并重写管理器。这更强大但也更复杂,除非你有非常特殊的需求(比如只用手机号登录),否则建议使用方法一。
创建自定义User模型
# models.py from django.db import models from django.contrib.auth.models import AbstractBaseUser, BaseUserManager, PermissionsMixin from django.utils import timezone from django.core.validators import RegexValidator class CustomUserManager(BaseUserManager): def create_user(self, email=None, mobile=None, password=None, **extra_fields): if not email and not mobile: raise ValueError('必须提供邮箱或手机号') if email: email = self.normalize_email(email) user = self.model(email=email, mobile=mobile, **extra_fields) user.set_password(password) user.save(using=self._db) return user def create_superuser(self, email=None, mobile=None, password=None, **extra_fields): extra_fields.setdefault('is_staff', True) extra_fields.setdefault('is_superuser', True) if extra_fields.get('is_staff') is not True: raise ValueError('超级用户必须设置 is_staff=True') if extra_fields.get('is_superuser') is not True: raise ValueError('超级用户必须设置 is_superuser=True') return self.create_user(email, mobile, password, **extra_fields) class User(AbstractBaseUser, PermissionsMixin): # 手机号验证器 mobile_validator = RegexValidator( regex=r'^1[3-9]\d{9}$', message="手机号格式不正确" ) email = models.EmailField(unique=True, null=True, blank=True, verbose_name='邮箱地址') mobile = models.CharField( max_length=11, unique=True, null=True, blank=True, validators=[mobile_validator], verbose_name='手机号' ) first_name = models.CharField(max_length=30, verbose_name='名') last_name = models.CharField(max_length=30, verbose_name='姓') is_active = models.BooleanField(default=True, verbose_name='激活状态') is_staff = models.BooleanField(default=False, verbose_name='员工状态') date_joined = models.DateTimeField(default=timezone.now, verbose_name='加入日期') objects = CustomUserManager() # 使用邮箱作为主要的USERNAME_FIELD,但实际登录时两者都支持 USERNAME_FIELD = 'email' REQUIRED_FIELDS = ['first_name', 'last_name'] class Meta: db_table = 'User' # 指定数据表名为 User verbose_name = '用户' verbose_name_plural = '用户' def __str__(self): return self.mobile or self.email or str(self.id) def get_full_name(self): return f"{self.first_name} {self.last_name}" def get_short_name(self): return self.first_name def clean(self): super().clean() if not self.email and not self.mobile: raise ValueError('邮箱和手机号不能同时为空')更新配置文件
# settings.py AUTH_USER_MODEL = 'yourapp.User' # 替换yourapp为你的应用名
创建序列化器
- 创建序列化器
# serializers.py
from rest_framework import serializers
from django.contrib.auth import authenticate
from .models import User
class UserSerializer(serializers.ModelSerializer):
"""用户序列化器"""
class Meta:
model = User
fields = '__all__'
read_only_fields = ('id', 'date_joined', 'password')
class UserRegistrationSerializer(serializers.ModelSerializer):
"""用户注册序列化器"""
password = serializers.CharField(write_only=True, min_length=6)
class Meta:
model = User
fields = ('email', 'mobile', 'password')
def validate(self, attrs):
email = attrs.get('email')
mobile = attrs.get('mobile')
if not email and not mobile:
raise serializers.ValidationError('必须提供邮箱或手机号')
return attrs
def create(self, validated_data):
return User.objects.create_user(**validated_data)
class CustomAuthSerializer(serializers.Serializer):
"""自定义认证序列化器,支持邮箱或手机号登录"""
login = serializers.CharField()
password = serializers.CharField(style={'input_type': 'password'})
def validate(self, attrs):
login = attrs.get('login')
password = attrs.get('password')
if login and password:
# 判断是邮箱还是手机号
if '@' in login:
user = authenticate(request=self.context.get('request'), email=login, password=password)
else:
user = authenticate(request=self.context.get('request'), mobile=login, password=password)
if not user:
raise serializers.ValidationError('无效的登录凭证')
if not user.is_active:
raise serializers.ValidationError('用户账户已被禁用')
attrs['user'] = user
return attrs
else:
raise serializers.ValidationError('必须提供登录凭证和密码')自定义认证后端
认证后端(Authentication Backend)的核心作用是将认证逻辑从视图/序列化器中解耦,提供灵活、可扩展的用户身份验证机制。
- 创建自定义认证后端
# backends.py
from django.contrib.auth.backends import ModelBackend
from .models import User
class EmailOrMobileBackend(ModelBackend):
def authenticate(self, request, username=None, password=None, **kwargs):
# 支持多种参数传递方式
login_id = username or kwargs.get('email') or kwargs.get('mobile')
if not login_id or not password:
return None
try:
# 判断是邮箱还是手机号
if '@' in login_id:
user = User.objects.get(email=login_id)
else:
user = User.objects.get(mobile=login_id)
except User.DoesNotExist:
# 运行默认密码哈希器以减少计时攻击
User().set_password(password)
return None
if user.check_password(password) and self.user_can_authenticate(user):
return user
return None- 配置认证后端:
# settings.py
AUTHENTICATION_BACKENDS = [
'your_app.backends.EmailOrMobileBackend',
'django.contrib.auth.backends.ModelBackend', # 保持默认后端作为备用
]视图开发
包含用户注册、用户登录、用户登出、查看个人信息、更新个人信息视图
from rest_framework import status
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework_simplejwt.tokens import RefreshToken
from .models import User
from .serializers import UserSerializer, UserRegistrationSerializer, CustomAuthSerializer
class UserRegistrationView(APIView):
"""用户注册视图"""
permission_classes = [AllowAny]
def post(self, request):
serializer = UserRegistrationSerializer(data=request.data)
if serializer.is_valid():
user = serializer.save()
user_data = UserSerializer(user).data
return Response({
'message': '用户注册成功',
'user': user_data
}, status=status.HTTP_201_CREATED)
return Response({
'message': '注册失败',
'errors': serializer.errors
}, status=status.HTTP_400_BAD_REQUEST)
class UserLoginView(APIView):
"""用户登录视图"""
permission_classes = [AllowAny]
def post(self, request):
serializer = CustomAuthSerializer(data=request.data, context={'request': request})
if serializer.is_valid():
user = serializer.validated_data['user']
refresh = RefreshToken.for_user(user)
user_data = UserSerializer(user).data
return Response({
'message': '登录成功',
'user': user_data,
'tokens': {
'refresh': str(refresh),
'access': str(refresh.access_token),
}
}, status=status.HTTP_200_OK)
return Response({
'message': '登录失败',
'errors': serializer.errors
}, status=status.HTTP_401_UNAUTHORIZED)
class UserProfileView(APIView):
"""用户个人信息视图"""
permission_classes = [IsAuthenticated]
def get(self, request):
"""获取用户信息"""
serializer = UserSerializer(request.user)
return Response({
'user': serializer.data
}, status=status.HTTP_200_OK)
def put(self, request):
"""更新用户信息"""
serializer = UserSerializer(
request.user,
data=request.data,
partial=True
)
if serializer.is_valid():
serializer.save()
return Response({
'message': '用户信息更新成功',
'user': serializer.data
}, status=status.HTTP_200_OK)
return Response({
'message': '更新失败',
'errors': serializer.errors
}, status=status.HTTP_400_BAD_REQUEST)
class UserLogoutView(APIView):
"""用户登出视图(可选)"""
permission_classes = [IsAuthenticated]
def post(self, request):
# 在客户端删除token即可实现登出
# 如果需要服务端黑名单功能,可以在这里实现
return Response({
'message': '登出成功'
}, status=status.HTTP_200_OK)