drfベースjwt認証およびユーザ登録実戦
drf TokenAuthentication認証 settings.py増加: データベース移行の実行: urls.py増加:
drf jwt認証インストール: settings.py増加: urls.py増加:
カスタムdjango認証
settings.py増加
common/auth.pyの追加:
ユーザー登録実戦
認証コード送信
utils.sms.py
serializers.py
views.py
urls.py
携帯番号+認証コード登録
serializers.py
views.py
信号量方式設定パスワード
signals.py
apps.py
ユーザ登録に成功しjwtに戻る
views.py
INSTALLED_APPS = [
...
'rest_framework.authtoken'
]
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.BasicAuthentication',
'rest_framework.authentication.SessionAuthentication',
'rest_framework.authentication.TokenAuthentication', #
],
]
python manage.py makemigrations
、python manage.py migrate
from rest_framework.authtoken import views
urlpatterns = [
# drf
path(r'api-token-auth/', views.obtain_auth_token),
]
drf jwt認証
pip install djangorestframework-jwt
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework_jwt.authentication.JSONWebTokenAuthentication',
...
],
]
from rest_framework_jwt.views import obtain_jwt_token
urlpatterns = [
# jwt token
path(r'api-token-auth/', obtain_jwt_token),
]
カスタムdjango認証
settings.py増加
AUTHENTICATION_BACKENDS = (
'common.auth.CustomBackend',
)
common/auth.pyの追加:
from django.contrib.auth.backends import ModelBackend
from django.contrib.auth import get_user_model
from django.db.models import Q
User = get_user_model()
class CustomBackend(ModelBackend):
def authenticate(self, request, username=None, password=None, **kwargs):
try:
user = User.objects.get(Q(username=username) | Q(mobile=username))
if user.check_password(password):
return user
except Exception as e:
return None
ユーザー登録実戦
認証コード送信
utils.sms.py
class Sms:
def __init__(self, api_key=''):
self.api_key = api_key
self.single_send_url = "#"
def send_sms(self, code, mobile):
return {'code': 0, 'msg': ' '}
serializers.py
from rest_framework import serializers
from django.contrib.auth import get_user_model
from django.conf import settings
from .models import VerifyCode
import datetime
import re
User = get_user_model()
class SmsSerializer(serializers.Serializer):
mobile = serializers.CharField(max_length=11)
def validate_mobile(self, mobile):
"""
:param data:
:return:
"""
#
if not re.match(settings.REGEX_MOBILE, mobile):
raise serializers.ValidationError(' ')
#
if User.objects.filter(mobile=mobile).count():
raise serializers.ValidationError(' ')
#
one_mintes_ago = datetime.datetime.now() - datetime.timedelta(hours=0, minutes=1, seconds=0)
if VerifyCode.objects.filter(add_time__gt=one_mintes_ago, mobile=mobile).count():
raise serializers.ValidationError(' 1 ')
return mobile
views.py
import random
from rest_framework.mixins import CreateModelMixin
from rest_framework import viewsets, status
from rest_framework.response import Response
from utils.sms import Sms
from .serializers import SmsSerializer
from .models import VerifyCode
class SmsCodeViewSet(CreateModelMixin, viewsets.GenericViewSet):
serializer_class = SmsSerializer
def generate_code(self):
seeds = "1234567890"
random_str = []
for i in range(6):
random_str.append(random.choice(seeds))
return ''.join(random_str)
"""
Create a model instance.
"""
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# validated_data
mobile = serializer.validated_data['mobile']
code = self.generate_code()
smsapi = Sms()
sms_status = smsapi.send_sms(code, mobile)
if sms_status['code'] != 0:
return Response({
'mobile': sms_status['msg']
}, status=status.HTTP_400_BAD_REQUEST)
else:
code_record = VerifyCode(code=code, mobile=mobile)
code_record.save()
return Response({
'mobile': mobile,
'code': code,
}, status=status.HTTP_201_CREATED)
urls.py
router.register(r'codes', SmsCodeViewset, base_name="codes")
携帯番号+認証コード登録
serializers.py
from rest_framework.validators import UniqueValidator
class UserRegSerializer(serializers.ModelSerializer):
code = serializers.CharField(label=' ', required=True, write_only=True, min_length=4, max_length=4,
error_messages={
'blank': ' ',
'required': ' ',
'max_length': ' ',
'min_length': ' ',
})
username = serializers.CharField(label=' ', help_text=' ', required=True, allow_blank=False,
validators=[UniqueValidator(queryset=User.objects.all(), message=' ')])
password = serializers.CharField(label=' ', help_text=' ', write_only=True, style={'input_type': 'password'})
def validate_code(self, code):
verify_row = VerifyCode.objects.filter(mobile=self.initial_data['username']).order_by('-add_time')
if verify_row:
last_row = verify_row[0]
mintes_ago = datetime.datetime.now() - datetime.timedelta(hours=0, minutes=5, seconds=0)
if last_row.add_time < mintes_ago:
raise serializers.ValidationError(' ')
if code != last_row.code:
raise serializers.ValidationError(' ')
else:
raise serializers.ValidationError(' ')
return code
# model
def validate(self, attrs): # validata
attrs['mobile'] = attrs['username']
del attrs['code']
return attrs
class Meta:
model = User
fields = ('username', 'code', 'password')
#
def create(self, validated_data):
user = super(UserRegSerializer, self).create(validated_data=validated_data)
user.set_password(validated_data['password'])
user.save()
return user
views.py
from rest_framework.mixins import CreateModelMixin
from rest_framework import viewsets
class UserViewSet(CreateModelMixin, viewsets.GenericViewSet):
serializer_class = UserRegSerializer
信号量方式設定パスワード
signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth import get_user_model
User = get_user_model()
@receiver(post_save, sender=User)
def create_user(sender, instance=None, created=False, **kwargs):
if created:
password = instance.password
instance.set_password(password)
instance.save()
apps.py
class UsersConfig(AppConfig):
...
def ready(self):
import users.signals
ユーザ登録に成功しjwtに戻る
views.py
class UserViewSet(CreateModelMixin, viewsets.GenericViewSet):
serializer_class = UserRegSerializer
queryset = User.objects.all()
def perform_create(self, serializer):
""" , """
return serializer.save()
def create(self, request, *args, **kwargs):
"" , jwt, ""
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
user = self.perform_create(serializer)
re_dict = serializer.data
# jwt
payload = jwt_payload_handler(user)
re_dict['token'] = jwt_encode_handler(payload)
re_dict['name'] = user.name if user.name else user.username
headers = self.get_success_headers(serializer.data)
return Response(re_dict, status=status.HTTP_201_CREATED, headers=headers)