SQLite for user data
0阅读
我将为你创建一个基于 Django + Redis + 微信开放平台扫码登录的最小化实例。这是一个完整的 RESTful API 风格实现。
项目结构
django_login_wxredis/
├── django_login_wxredis/
│ ├── __init__.py
│ ├── settings.py
│ ├── urls.py
│ ├── asgi.py
│ └── wsgi.py
├── api_redis/
│ ├── __init__.py
│ ├── models.py
│ ├── views.py
│ ├── urls.py
│ ├── serializers.py
│ └── utils.py
├── manage.py
└── requirements.txt
1. 配置文件
requirements.txt
Django>=4.2,<5.0
djangorestframework>=3.14
redis>=4.5
requests>=2.28
python-dotenv>=1.0
django_login_wxredis/settings.py
import os
from pathlib import Path
BASE_DIR = Path(__file__).resolve().parent.parent
SECRET_KEY = 'your-secret-key-change-in-production'
DEBUG = True
ALLOWED_HOSTS = ['*']
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'rest_framework',
'api_redis',
]
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
ROOT_URLCONF = 'django_login_wxredis.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
WSGI_APPLICATION = 'django_login_wxredis.wsgi.application'
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': BASE_DIR / 'db.sqlite3',
}
}
AUTH_PASSWORD_VALIDATORS = [
{'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator'},
{'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator'},
{'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator'},
{'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator'},
]
LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True
USE_TZ = True
STATIC_URL = 'static/'
DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
# Redis Configuration
REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
REDIS_DB = int(os.getenv('REDIS_DB', 0))
REDIS_PASSWORD = os.getenv('REDIS_PASSWORD', None)
# WeChat Open Platform Config
WECHAT_APPID = os.getenv('WECHAT_APPID', 'your-appid')
WECHAT_SECRET = os.getenv('WECHAT_SECRET', 'your-secret')
# JWT / Token settings
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.SessionAuthentication',
'api_redis.authentication.BearerTokenAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
],
'DEFAULT_RENDERER_CLASSES': [
'rest_framework.renderers.JSONRenderer',
],
}
django_login_wxredis/urls.py
from django.contrib import admin
from django.urls import path, include
urlpatterns = [
path('admin/', admin.site.urls),
path('api/', include('api_redis.urls')),
]
2. 数据模型与序列化器
api_redis/models.py
from django.contrib.auth.models import AbstractUser
from django.db import models
import uuid
class User(AbstractUser):
"""自定义用户模型，支持微信登录"""
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
openid = models.CharField(max_length=128, unique=True, null=True, blank=True, verbose_name='微信OpenID')
unionid = models.CharField(max_length=128, unique=True, null=True, blank=True, verbose_name='微信UnionID')
nickname = models.CharField(max_length=64, blank=True, verbose_name='昵称')
avatar_url = models.URLField(blank=True, verbose_name='头像URL')
phone = models.CharField(max_length=20, blank=True, verbose_name='手机号')
is_wechat_bound = models.BooleanField(default=False, verbose_name='是否绑定微信')
last_login_ip = models.GenericIPAddressField(null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
db_table = 'users'
verbose_name = '用户'
verbose_name_plural = verbose_name
def __str__(self):
return self.nickname or self.username or str(self.id)
api_redis/serializers.py
from rest_framework import serializers
from .models import User
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ['id', 'username', 'nickname', 'avatar_url', 'phone', 'is_wechat_bound', 'created_at']
read_only_fields = ['id', 'created_at']
class QRCodeStatusSerializer(serializers.Serializer):
scene_id = serializers.CharField()
status = serializers.ChoiceField(choices=['pending', 'scanned', 'confirmed', 'expired', 'cancelled'])
user_info = UserSerializer(required=False)
token = serializers.CharField(required=False)
message = serializers.CharField(required=False)
3. Redis 工具类
api_redis/utils.py
import json
import redis
import uuid
from datetime import datetime
from django.conf import settings
class RedisQRCodeManager:
"""Redis 二维码状态管理器"""
def __init__(self):
self.redis_client = redis.Redis(
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
db=settings.REDIS_DB,
password=settings.REDIS_PASSWORD,
decode_responses=True
)
self.key_prefix = "wx:qrcode:"
self.expire_time = 300 # 5分钟过期
def _get_key(self, scene_id: str) -> str:
return f"{self.key_prefix}{scene_id}"
def create_qrcode(self) -> dict:
"""创建新的二维码场景"""
scene_id = str(uuid.uuid4())[:8] # 短ID便于扫码
data = {
'scene_id': scene_id,
'status': 'pending', # pending, scanned, confirmed, expired, cancelled
'created_at': datetime.now().isoformat(),
'openid': None,
'user_info': None,
'token': None
}
key = self._get_key(scene_id)
self.redis_client.setex(key, self.expire_time, json.dumps(data))
return {
'scene_id': scene_id,
'expire_seconds': self.expire_time,
'qrcode_url': f"weixin://dl/business/?t={scene_id}" # 模拟微信扫码URL
}
def get_status(self, scene_id: str) -> dict:
"""获取二维码状态"""
key = self._get_key(scene_id)
data = self.redis_client.get(key)
if not data:
return {'status': 'expired', 'message': '二维码已过期'}
return json.loads(data)
def update_status(self, scene_id: str, status: str, **kwargs) -> bool:
"""更新二维码状态"""
key = self._get_key(scene_id)
data = self.redis_client.get(key)
if not data:
return False
data = json.loads(data)
data['status'] = status
data.update(kwargs)
# 确认或取消后，延长短暂时间供前端获取结果，然后销毁
if status in ['confirmed', 'cancelled']:
self.redis_client.setex(key, 10, json.dumps(data)) # 10秒后销毁
else:
self.redis_client.setex(key, self.expire_time, json.dumps(data))
return True
def mark_scanned(self, scene_id: str, openid: str) -> bool:
"""标记为已扫描"""
return self.update_status(scene_id, 'scanned', openid=openid)
def confirm_login(self, scene_id: str, user_info: dict, token: str) -> bool:
"""确认登录，返回用户信息"""
return self.update_status(
scene_id,
'confirmed',
user_info=user_info,
token=token
)
def cancel_login(self, scene_id: str) -> bool:
"""取消登录"""
return self.update_status(scene_id, 'cancelled')
def delete_qrcode(self, scene_id: str):
"""立即删除二维码"""
key = self._get_key(scene_id)
self.redis_client.delete(key)
# 全局实例
redis_qr = RedisQRCodeManager()
4. 自定义认证
api_redis/authentication.py
from rest_framework import authentication, exceptions
from rest_framework.authtokens.models import Token
from .models import User
class BearerTokenAuthentication(authentication.BaseAuthentication):
"""Bearer Token 认证"""
def authenticate(self, request):
auth_header = request.headers.get('Authorization', '')
if not auth_header.startswith('Bearer '):
return None
token_key = auth_header.split(' ')[1]
try:
token = Token.objects.get(key=token_key)
return (token.user, token)
except Token.DoesNotExist:
raise exceptions.AuthenticationFailed('Invalid token')
5. 核心视图
api_redis/views.py
import requests
from rest_framework import status, permissions
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.authtokens.models import Token
from django.contrib.auth import get_user_model
from django.db import IntegrityError
from .utils import redis_qr
from .serializers import UserSerializer, QRCodeStatusSerializer
User = get_user_model()
class QRCodeGenerateView(APIView):
"""生成登录二维码"""
permission_classes = [permissions.AllowAny]
def post(self, request):
"""创建新的二维码"""
qr_data = redis_qr.create_qrcode()
return Response({
'code': 200,
'message': 'success',
'data': {
'scene_id': qr_data['scene_id'],
'qrcode_url': qr_data['qrcode_url'],
'expire_seconds': qr_data['expire_seconds']
}
})
class QRCodeStatusView(APIView):
"""轮询查询二维码状态"""
permission_classes = [permissions.AllowAny]
def get(self, request, scene_id):
"""查询状态"""
status_data = redis_qr.get_status(scene_id)
response_data = {
'code': 200,
'data': {
'scene_id': scene_id,
'status': status_data.get('status'),
}
}
# 如果已确认登录，返回token
if status_data.get('status') == 'confirmed':
response_data['data']['token'] = status_data.get('token')
response_data['data']['user_info'] = status_data.get('user_info')
if status_data.get('message'):
response_data['message'] = status_data['message']
return Response(response_data)
class WeChatScanCallbackView(APIView):
"""微信扫码回调（模拟微信服务器回调）"""
permission_classes = [permissions.AllowAny]
def post(self, request):
"""
微信扫码后回调此接口
实际应由微信服务器调用，这里模拟处理
"""
scene_id = request.data.get('scene_id')
openid = request.data.get('openid')
if not scene_id or not openid:
return Response(
{'code': 400, 'message': 'Missing parameters'},
status=status.HTTP_400_BAD_REQUEST
)
# 更新为已扫描状态
success = redis_qr.mark_scanned(scene_id, openid)
if not success:
return Response(
{'code': 404, 'message': 'QR code expired or not found'},
status=status.HTTP_404_NOT_FOUND
)
return Response({
'code': 200,
'message': 'Scanned successfully, waiting for confirm'
})
class WeChatConfirmLoginView(APIView):
"""用户确认登录（手机端操作）"""
permission_classes = [permissions.AllowAny]
def post(self, request):
"""
用户在手机端点击确认登录
1. 获取或创建用户
2. 生成Token
3. 更新Redis状态
"""
scene_id = request.data.get('scene_id')
openid = request.data.get('openid')
user_info = request.data.get('user_info', {})
if not scene_id or not openid:
return Response(
{'code': 400, 'message': 'Missing parameters'},
status=status.HTTP_400_BAD_REQUEST
)
# 获取或创建用户
try:
user, created = User.objects.get_or_create(
openid=openid,
defaults={
'username': f"wx_{openid[-8:]}",
'nickname': user_info.get('nickname', f"微信用户{openid[-6:]}"),
'avatar_url': user_info.get('headimgurl', ''),
'is_wechat_bound': True
}
)
if not created:
# 更新用户信息
user.nickname = user_info.get('nickname', user.nickname)
user.avatar_url = user_info.get('headimgurl', user.avatar_url)
user.save()
except IntegrityError:
return Response(
{'code': 500, 'message': 'User creation failed'},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
# 生成或获取Token
token, _ = Token.objects.get_or_create(user=user)
# 序列化用户信息
user_serializer = UserSerializer(user)
# 更新Redis状态为已确认
success = redis_qr.confirm_login(
scene_id,
user_info=user_serializer.data,
token=token.key
)
if not success:
return Response(
{'code': 404, 'message': 'QR code expired'},
status=status.HTTP_404_NOT_FOUND
)
return Response({
'code': 200,
'message': 'Login confirmed',
'data': {
'token': token.key,
'user': user_serializer.data
}
})
class WeChatCancelLoginView(APIView):
"""用户取消登录"""
permission_classes = [permissions.AllowAny]
def post(self, request):
scene_id = request.data.get('scene_id')
if not scene_id:
return Response(
{'code': 400, 'message': 'Missing scene_id'},
status=status.HTTP_400_BAD_REQUEST
)
success = redis_qr.cancel_login(scene_id)
if not success:
return Response(
{'code': 404, 'message': 'QR code not found'},
status=status.HTTP_404_NOT_FOUND
)
return Response({
'code': 200,
'message': 'Login cancelled'
})
class UserProfileView(APIView):
"""获取当前用户信息"""
permission_classes = [permissions.IsAuthenticated]
def get(self, request):
serializer = UserSerializer(request.user)
return Response({
'code': 200,
'data': serializer.data
})
class LogoutView(APIView):
"""退出登录"""
permission_classes = [permissions.IsAuthenticated]
def post(self, request):
# 删除Token
Token.objects.filter(user=request.user).delete()
return Response({
'code': 200,
'message': 'Logged out successfully'
})
6. URL 路由
api_redis/urls.py
from django.urls import path
from . import views
urlpatterns = [
# 二维码相关
path('auth/qrcode/generate/', views.QRCodeGenerateView.as_view(), name='qrcode-generate'),
path('auth/qrcode/status/<str:scene_id>/', views.QRCodeStatusView.as_view(), name='qrcode-status'),
# 微信回调（模拟）
path('auth/wechat/callback/', views.WeChatScanCallbackView.as_view(), name='wechat-callback'),
path('auth/wechat/confirm/', views.WeChatConfirmLoginView.as_view(), name='wechat-confirm'),
path('auth/wechat/cancel/', views.WeChatCancelLoginView.as_view(), name='wechat-cancel'),
# 用户相关
path('user/profile/', views.UserProfileView.as_view(), name='user-profile'),
path('auth/logout/', views.LogoutView.as_view(), name='logout'),
]
7. 管理配置
api_redis/admin.py
from django.contrib import admin
from django.contrib.auth.admin import UserAdmin
from .models import User
@admin.register(User)
class CustomUserAdmin(UserAdmin):
list_display = ['id', 'username', 'nickname', 'openid', 'is_wechat_bound', 'created_at']
list_filter = ['is_wechat_bound', 'is_staff', 'created_at']
search_fields = ['username', 'nickname', 'openid', 'phone']
readonly_fields = ['id', 'created_at', 'updated_at']
fieldsets = (
(None, {'fields': ('id', 'username', 'password')}),
('微信信息', {'fields': ('openid', 'unionid', 'is_wechat_bound')}),
('个人信息', {'fields': ('nickname', 'avatar_url', 'phone')}),
('权限', {'fields': ('is_active', 'is_staff', 'is_superuser', 'groups', 'user_permissions')}),
('重要日期', {'fields': ('last_login', 'date_joined', 'created_at', 'updated_at')}),
)
api_redis/init.py
default_app_config = 'api_redis.apps.ApiRedisConfig'
api_redis/apps.py
from django.apps import AppConfig
class ApiRedisConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'api_redis'
verbose_name = '微信登录API'
8. 启动脚本
manage.py
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main():
"""Run administrative tasks."""
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_login_wxredis.settings')
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == '__main__':
main()
9. 初始化命令
创建 init_project.sh：
#!/bin/bash
# 创建项目目录结构
mkdir -p django_login_wxredis/django_login_wxredis
mkdir -p django_login_wxredis/api_redis
# 进入项目目录
cd django_login_wxredis
# 创建虚拟环境
python3 -m venv venv
source venv/bin/activate
# 安装依赖
pip install -r requirements.txt
# 执行迁移
python manage.py makemigrations
python manage.py migrate
# 创建超级用户（可选）
echo "from django.contrib.auth import get_user_model; User = get_user_model(); User.objects.create_superuser('admin', '[email protected]', 'admin123') if not User.objects.filter(username='admin').exists() else print('Admin exists')" | python manage.py shell
echo "项目初始化完成！"
echo "运行: python manage.py runserver"
API 端点说明
| 方法 | 端点 | 描述 | 认证 |
|---|---|---|---|
| POST | /api/auth/qrcode/generate/ |
生成登录二维码 | 否 |
| GET | /api/auth/qrcode/status/{scene_id}/ |
轮询二维码状态 | 否 |
| POST | /api/auth/wechat/callback/ |
微信扫码回调（模拟） | 否 |
| POST | /api/auth/wechat/confirm/ |
确认登录（手机端） | 否 |
| POST | /api/auth/wechat/cancel/ |
取消登录 | 否 |
| GET | /api/user/profile/ |
获取用户信息 | Bearer Token |
| POST | /api/auth/logout/ |
退出登录 | Bearer Token |
登录流程时序
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Web端 │ │ Redis │ │ 手机端 │
└────┬────┘ └────┬────┘ └────┬────┘
│ │ │
│ 1. POST /qrcode/generate │ │
│─────────────────────────────>│ │
│ │ 创建scene_id, status=pending │
│ │ (TTL 5分钟) │
│ 返回 scene_id + QR码 │ │
│<─────────────────────────────│ │
│ │ │
│ 2. 轮询 GET /status/{id} │ │
│─────────────────────────────>│ │
│ 返回 pending │ │
│<─────────────────────────────│ │
│ │ │
│ │ 3. 扫码 POST /callback │
│ │<─────────────────────────────│
│ │ 更新 status=scanned │
│ │ │
│ 4. 轮询返回 scanned │ │
│<─────────────────────────────│ │
│ │ │
│ │ 5. 确认 POST /confirm │
│ │<─────────────────────────────│
│ │ 更新 status=confirmed │
│ │ 存储 token + user_info │
│ │ (TTL 10秒后销毁) │
│ │ │
│ 6. 轮询返回 confirmed + token│ │
│<─────────────────────────────│ │
│ │ │
│ 7. 使用 Bearer Token 访问API │ │
│────────────────────────────────────────────────────────────>│
快速启动
# 1. 确保 Redis 运行中
redis-server
# 2. 初始化项目
chmod +x init_project.sh
./init_project.sh
# 3. 启动服务
python manage.py runserver 0.0.0.0:8000
这个最小化实例实现了：
- ✅ 二维码生成与 Redis 存储（5分钟过期）
- ✅ 扫码状态管理（pending → scanned → confirmed/cancelled）
- ✅ 登录完成后立即销毁二维码（10秒缓冲）
- ✅ RESTful API 风格
- ✅ Bearer Token 认证
- ✅ 前后端分离架构
- ✅ 自定义 User 模型支持微信信息