python 数据库 Django前后端分离之后端实践-进阶

管道技术-pipeline

redis的C-S架构

基于客户端-服务端模型及请求、响应协议的TCP服务客户端向服务端发送一个查询请求，并监听socket返回通常是阻塞模式，等待服务端的响应服务端处理命令，并将结果返回给客户端

管道技术可以解决的问题

当redis服务端需要同时处理多个请求，加上网络延迟，可用管道来提升服务端的利用率，提高效率

pipeline的介绍

定义：

可以一次性发送多个命令并发送完成后一次性将结果返回

pipeline通过减少客户端与redis的通信次数来降低往返的延迟时间

实现的原理-队列的使用

client 可以将三个命令放到一个tcp报文一起发送server可以将三个命令的处理结果放到一个tcp报文返回利用队列的先进先出的原理实现完成，可以保证数据的顺序性

pipeline操作redis数据库

实现步骤

创建redis管道将redis请求添加到队列执行请求

 代码的实现

from redis import StrictRedis

redis_cli = StrictRedis(host="xx", port=xx, password="xx", db=xx, decode_responses=True)

# 去自己的队列中获取数据

# 创建管道

pl = redis_cli.pipeline()

# 管道收集指令

pl.setex('xxx',300,'code')

# 添加一个发送标记，有效时间60s

pl.setex('data',60,1)

pl.execute()

更多的redis信息请参考博客地址https://blog.csdn.net/myli_binbin/article/details/121323231?ops_request_misc=&request_id=&biz_id=102&utm_term=pl%20=%20redis_cli.pipeline&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduweb~default-0-121323231.142^v99^pc_search_result_base2&spm=1018.2226.3001.4187https://blog.csdn.net/myli_binbin/article/details/121323231?ops_request_misc=&request_id=&biz_id=102&utm_term=pl%20=%20redis_cli.pipeline&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduweb~default-0-121323231.142^v99^pc_search_result_base2&spm=1018.2226.3001.4187

小实践：实现用户登录、注销及ORM管理功能、事务开启小实践

models.py

class Books(models.Model):

id = models.CharField(primary_key=True,max_length=20,verbose_name="图书ID")

name = models.CharField(max_length=20,verbose_name='图书名称')

status = models.BooleanField(default=False,verbose_name='是否出借',blank=True)

class Meta:

db_table = 'books'

verbose_name = '图书表'

def __str__(self):

return self.name

class Record(models.Model):

book = models.ForeignKey('Books',on_delete=models.CASCADE,verbose_name="图书")

name = models.CharField(max_length=20,verbose_name='借书姓名')

s_time = models.DateTimeField(auto_created=True,verbose_name='借书时间',auto_now=True) # auto_created 自动创建当前时间

e_time = models.DateTimeField(auto_created=True,verbose_name='还书时间',auto_now=True) # auto_now 任意一个字段变更都会自动更新时间

state = models.BooleanField(default=False,verbose_name='是否归还',blank=True)

class Meta:

db_table = 'records'

verbose_name = '图书记录'

 urls.py[注意：路由结尾要记得加$符号，负责会匹配其他的视图函数]

from django.contrib import admin

from django.urls import path,re_path,include

from news import views

urlpatterns = [

path('admin/', admin.site.urls),

re_path(r'^news/',include('news.urls')),

re_path(r'logout/$',views.LogoutView.as_view()),

re_path(r'login/$',views.LoginView.as_view()),

re_path(r'book/$',views.BookView.as_view()),

re_path(r'book/hanlde/$',views.RecordView.as_view())

]

用户登录注销代码的实现

from django.http import HttpResponse,JsonResponse

from django.views import View

from django.contrib.auth import authenticate,logout,login

class LogoutView(View):

def get(self,request):

logout(request)

return JsonResponse({'code':200,'message':'已退出登录'},status=200)

class LoginView(View):

def post(self,request):

username = request.POST.get('username')

password = request.POST.get('password')

user = authenticate(request,username=username,password=password)

if user is not None:

login(request,user)

return JsonResponse({'code':200,'message':'登录成功'})

else:

return JsonResponse({'code':400,'message':'登录失败'})

admin.py  可后台查看数据

from .models import Books,Record

class BookAdmin(admin.ModelAdmin):

list_display = ['id','name','status']

class RecordAdmin(admin.ModelAdmin):

list_display = ['id','book','name','s_time','e_time','state']

admin.site.register(Books,BookAdmin)

admin.site.register(Record,RecordAdmin)

 ORM功能的增删查实现

class BookView(View):

def get(self,request):

if not request.user.is_authenticated:

return JsonResponse({'code':2002,'messages':"认证失败你没有权限访问，请重新登录"})

bs= Books.objects.all()

result = []

for i in bs:

item = dict(id=i.id,name=i.name,status=i.status)

result.append(item)

return JsonResponse({'code':200,'data':result,'message':'ok'},safe=False)

def post(self,request):

import json

# param =request.POST if len(request.POST)> 0 else json.loads(request.body.decode())

id = request.POST.get('id')

name = request.POST.get('name')

try:

Books.objects.create(id=id,name=name)

except Exception as e:

return JsonResponse({'code':2001,'message':'书籍编号已存在,异常：{}'.format(e)},safe=False)

else:

return JsonResponse({'code':1001,'message':'添加成功'},safe=False)

def delete(self,request):

id = request.POST.get('id')

if not id:

return JsonResponse({'code':1001,'message':'删除失败'})

try:

book = Books.objects.get(id=id)

except Exception as e:

return JsonResponse({'code': 1001, 'message': '删除失败:{}'.format(e)})

else:

book.delete()

return JsonResponse({'code': 200, 'message': '删除成功'})

 应用结果：

事务实现

eg: 当进行批量文件上传时，避免造成数据冗余，可以使用事务的原理：若一次操作的失败将执行回滚，不执行任何的操作

class RecordView(View):

def get(self, request):

pass

def post(self, request):

book_id = request.POST.get('id')

name = request.POST.get('name')

try:

book = Books.objects.get(id=book_id)

except Exception as e:

return JsonResponse({'code': 2001, 'messages': "{}".format(e)})

if book.status:

return JsonResponse({'code': 2001, 'messages': "该书籍已借书，请确认书籍的状态是否已出借"})

with transaction.atomic():

# update status

book.status = True

book.save()

# add record

Record.objects.create(name=name, book=book)

return JsonResponse({'code': 1000, 'messages': "出借成功"})

精彩内容