0%

DRF实践

说明

  • 这篇文章 使用类视图实现了orm中常见的一对一,一对多,多对多。本次使用DRF的方式来实现

环境搭建

  • 发现本地的环境有问题,重新搭建个

  • 下载一个python 3.10 64位的,然后pycharm使用虚拟环境

  • 设置默认下载镜像

1
2
3
4
C:\Users\Administrator\AppData\Roaming\pip\pip.ini

[global]
index-url = https://pypi.tuna.tsinghua.edu.cn/simple
  • env环境安装各种依赖
1
2
3
4
5
6
pip install Django
pip install mysqlclient # 之前版本用的是pymysql,还要再init.py文件中加入内容,现在用这个没有问题
pip install djangorestframework
pip install django-filter # 搜索用的
pip install djangorestframework-simplejwt # jwt鉴权使用

  • 项目的setting加入
1
DEFAULT_AUTO_FIELD = 'django.db.models.AutoField'

消除该警告:表明 Django 模型未显式定义主键类型,系统自动使用了默认的 AutoField

然后执行下面生效

(venv) E:\proj\StudyDjangoAdmin>python manage.py migrate DRFDjango

python manage.py makemigrations DRFDjango

  • 看看之前的环境是否可以启动
1
(venv) E:\proj\StudyDjangoAdmin> python manage.py runserver 0.0.0.0:8000

image-20251012165148267

DRF实践

  • 新建一个应用
1
django-admin startapp DRFDjango
  • 项目下的setting注册应用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'TestModel', #
'DRFDjango',
'rest_framework',
'django_filters',

]

REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': (
'rest_framework_simplejwt.authentication.JWTAuthentication',
# 如果需要保留其他认证方式,可以添加如下
# 'rest_framework.authentication.SessionAuthentication',
# 'rest_framework.authentication.BasicAuthentication',
),

# 设置全局默认权限策略
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated', # 默认所有视图需要认证
],

'DEFAULT_FILTER_BACKENDS': [
'django_filters.rest_framework.DjangoFilterBackend',
'rest_framework.filters.SearchFilter',
'rest_framework.filters.OrderingFilter',
],
}

用默认的表做鉴权
AUTH_USER_MODEL = 'auth.User'

# JWT 配置 过期时间

from datetime import timedelta

SIMPLE_JWT = {
'ACCESS_TOKEN_LIFETIME': timedelta(minutes=5),
'REFRESH_TOKEN_LIFETIME': timedelta(days=1),
'ROTATE_REFRESH_TOKENS': True,
'BLACKLIST_AFTER_ROTATION': True,
'ALGORITHM': 'HS256',
'SIGNING_KEY': 'your-secret-key',
'VERIFYING_KEY': None,
}

  • model,这里作者详情和作者优化了,之前要先创建作者详情后再创建作者信息,不合理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from django.contrib.auth.models import AbstractUser
from django.db import models


# Create your models here.


class Book(models.Model):
title = models.CharField(max_length=32)
price = models.DecimalField(max_digits=5, decimal_places=2)
pub_date = models.DateField()
# 关联出版社一对多,意思就是一个出版社可以有印刷多本书
publish = models.ForeignKey("Publish", on_delete=models.CASCADE, default=None)
# 多对多,意思就是多个作者可以编写多本书
authors = models.ManyToManyField("Author")

def __str__(self):
return self.title


# 出版商
class Publish(models.Model):
name = models.CharField(max_length=32)
city = models.CharField(max_length=64)
email = models.EmailField()

def __str__(self):
return self.name


class Author(models.Model):
name = models.CharField(max_length=100, null=True, blank=True)
age = models.IntegerField()
# 改为可为空了
au_detail = models.OneToOneField(
'AuthorDetail',
on_delete=models.SET_NULL,
null=True, # 允许数据库存储NULL
blank=True, # 允许表单验证为空
related_name='author'
)


class AuthorDetail(models.Model):
GENDER_CHOICES = (
(0, '女'),
(1, '男'),
(2, '保密')
)

gender = models.SmallIntegerField(choices=GENDER_CHOICES)
tel = models.CharField(max_length=20)
addr = models.CharField(max_length=100)
birthday = models.DateField()
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)

  • 序列化代码serializers.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
from rest_framework import serializers

from .models import Book, Publish, Author, AuthorDetail

class AuthorDetailSerializer(serializers.ModelSerializer):
class Meta:
model = AuthorDetail
fields = '__all__'
read_only_fields = ('created_at', 'updated_at')


class AuthorSerializer(serializers.ModelSerializer):
au_detail = AuthorDetailSerializer(required=False, allow_null=True)
au_detail_id = serializers.IntegerField(
write_only=True,
required=False,
allow_null=True
)

class Meta:
model = Author
fields = ['id', 'name', 'age', 'au_detail', 'au_detail_id']
extra_kwargs = {
'au_detail': {'required': False, 'allow_null': True},
}

def validate(self, data):
# 确保至少提供一种形式的au_detail数据
if 'au_detail' not in data and 'au_detail_id' not in data:
data['au_detail'] = None # 明确设置为None
return data

def create(self, validated_data):
au_detail_data = validated_data.pop('au_detail', None)
au_detail_id = validated_data.pop('au_detail_id', None)

# 创建Author实例(不包含au_detail)
author = Author.objects.create(**validated_data)

# 处理AuthorDetail
if au_detail_data:
author.au_detail = AuthorDetail.objects.create(**au_detail_data)
author.save()
elif au_detail_id is not None: # 明确检查是否为None
author.au_detail_id = au_detail_id
author.save()

return author

def update(self, instance, validated_data):
au_detail_data = validated_data.pop('au_detail', None)
au_detail_id = validated_data.pop('au_detail_id', None)

# 更新Author字段
for attr, value in validated_data.items():
setattr(instance, attr, value)

# 处理AuthorDetail
if au_detail_data:
if instance.au_detail:
# 更新现有AuthorDetail
detail_serializer = AuthorDetailSerializer(
instance.au_detail,
data=au_detail_data,
partial=True
)
detail_serializer.is_valid(raise_exception=True)
detail_serializer.save()
else:
# 创建新AuthorDetail
instance.au_detail = AuthorDetail.objects.create(**au_detail_data)
elif au_detail_id is not None:
instance.au_detail_id = au_detail_id

instance.save()
return instance


class PublishSerializer(serializers.ModelSerializer):
class Meta:
model = Publish
fields = '__all__'


class BookSerializer(serializers.ModelSerializer):
publish = PublishSerializer(read_only=True)
authors = AuthorSerializer(many=True, read_only=True)

# 用于写入的字段
publish_id = serializers.PrimaryKeyRelatedField(
queryset=Publish.objects.all(), source='publish', write_only=True
)
author_ids = serializers.PrimaryKeyRelatedField(
queryset=Author.objects.all(), many=True, source='authors', write_only=True
)

class Meta:
model = Book
fields = '__all__'
extra_fields = ['publish_id', 'author_ids']
  • view.py 内容
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from rest_framework import viewsets, pagination
from rest_framework.response import Response

from .models import Book, Publish, Author, AuthorDetail
from .serializers import BookSerializer, PublishSerializer, AuthorSerializer, AuthorDetailSerializer
from rest_framework import filters
from django_filters.rest_framework import DjangoFilterBackend


class StandardPagination(pagination.PageNumberPagination):
page_size = 10
page_size_query_param = 'page_size'
max_page_size = 100


class BookViewSet(viewsets.ModelViewSet):
queryset = Book.objects.all().order_by('-id')
serializer_class = BookSerializer
pagination_class = StandardPagination

def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
headers = self.get_success_headers(serializer.data)
return Response(serializer.data, status=201, headers=headers)

def perform_create(self, serializer):
book = serializer.save()
author_ids = self.request.data.get('author_ids', [])
book.authors.set(author_ids)


class PublishViewSet(viewsets.ModelViewSet):
queryset = Publish.objects.all().order_by('-id')
serializer_class = PublishSerializer
pagination_class = StandardPagination
filter_backends = [DjangoFilterBackend, filters.SearchFilter]
# 修正后的搜索字段配置
search_fields = ['name', 'city'] # 可以搜索名称和城市

# 修正后的过滤字段配置
filterset_fields = {
'name': ['exact', 'icontains'], # 精确匹配或包含
'city': ['exact'],
}

def get_queryset(self):
queryset = super().get_queryset()
# 可以在这里添加自定义的查询逻辑
return queryset


class AuthorViewSet(viewsets.ModelViewSet):
queryset = Author.objects.all().order_by('-id')
serializer_class = AuthorSerializer
pagination_class = StandardPagination


class AuthorDetailViewSet(viewsets.ModelViewSet):
queryset = AuthorDetail.objects.all().order_by('-id')
serializer_class = AuthorDetailSerializer
pagination_class = StandardPagination

  • 路由url
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from rest_framework.routers import DefaultRouter

from DRFDjango.views import BookViewSet, PublishViewSet, AuthorViewSet, AuthorDetailViewSet

from rest_framework_simplejwt.views import (
TokenObtainPairView,
TokenRefreshView,
)

router = DefaultRouter()
router.register(r'books', BookViewSet)
router.register(r'publishes', PublishViewSet)
router.register(r'authors', AuthorViewSet)
router.register(r'authordetails', AuthorDetailViewSet)

urlpatterns = [
path('admin/', admin.site.urls),
path('accounts/logout/', LogoutView.as_view(next_page='/login/'), name='logout'),
path('accounts/login/', LoginView.as_view(template_name='login.html'), name='login'),
path('', include('TestModel.urls')),
path('api/', include((router.urls, 'api'), namespace='api')),
# JWT 认证路由
path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
]

  • 测试代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
BASE_URL = 'http://localhost:8000/api/'
AUTH_URL = BASE_URL + "token/"

HEADER = {
"Content-Type": "application/json"
}
def get_jwt_token():
"""获取JWT Token"""
# 测试用户凭证
USER_CREDENTIALS = {
"username": "admin", # 替换为你的测试用户名
"password": "admin" # 替换为测试密码
}
response = requests.post(AUTH_URL, json=USER_CREDENTIALS)
if response.status_code == 200:
print(response.json())
HEADER["Authorization"] = "Bearer "+ response.json()["access"]
return response.json()["access"]
raise Exception(f"获取Token失败: {response.text}")

def get_publishes():
get_jwt_token()
params = {'page': 1, 'page_size': 2, "search": "北京2121"}
url = f"{BASE_URL}publishes"
response = requests.get(url, params=params, headers=HEADER)
if response.status_code == 200:
print("Publishers retrieved successfully:")
for publish in response.json()["results"]:
print(publish)
else:
print(f"Failed to retrieve publishers. Status code: {response.status_code}")
  • 更加详测试代码就不贴了,更新到代码仓库中

总结

  • 迁移过程中,发现有些表字段迁移失败
1
2
python manage.py migrate DRFDjango zero
python manage.py migrate DRFDjango
  • 本次分别使用用DRF结合jwt,orm常见的结构进行了实践。