Python实战:5分钟搞定阿里云短信验证码发送(Django项目集成版)

张开发
2026/4/10 12:03:06 15 分钟阅读

分享文章

Python实战:5分钟搞定阿里云短信验证码发送(Django项目集成版)
Python实战5分钟搞定阿里云短信验证码发送Django项目集成版短信验证码已经成为现代Web应用身份验证的标配功能。作为Python开发者如何在Django项目中快速集成阿里云短信服务同时确保安全性和可维护性本文将带你从零开始5分钟内完成从配置到部署的全流程。1. 环境准备与基础配置在开始编码之前我们需要确保开发环境已经准备就绪。首先创建一个新的Django项目如果已有项目可跳过此步pip install django django-admin startproject sms_project cd sms_project接下来安装阿里云Python SDK这是与阿里云短信服务交互的核心依赖pip install aliyun-python-sdk-core-v3 aliyun-python-sdk-dysmsapi安全提示永远不要在代码中硬编码敏感信息。我们将使用Django的环境变量管理方案# settings.py import os ALIYUN_ACCESS_KEY_ID os.getenv(ALIYUN_ACCESS_KEY_ID) ALIYUN_ACCESS_KEY_SECRET os.getenv(ALIYUN_ACCESS_KEY_SECRET) ALIYUN_SMS_SIGN_NAME os.getenv(ALIYUN_SMS_SIGN_NAME) ALIYUN_SMS_TEMPLATE_CODE os.getenv(ALIYUN_SMS_TEMPLATE_CODE)对应的.env文件示例ALIYUN_ACCESS_KEY_IDyour_access_key_id ALIYUN_ACCESS_KEY_SECRETyour_access_key_secret ALIYUN_SMS_SIGN_NAME你的签名 ALIYUN_SMS_TEMPLATE_CODESMS_1234567892. 核心服务层封装良好的架构应该分离关注点。我们创建一个独立服务层处理短信发送逻辑# services/sms_service.py import random from aliyunsdkcore.client import AcsClient from aliyunsdkcore.request import CommonRequest from django.conf import settings class SMSService: staticmethod def generate_verification_code(length4): return .join(random.choices(0123456789, klength)) classmethod def send_verification_code(cls, phone_number): client AcsClient( settings.ALIYUN_ACCESS_KEY_ID, settings.ALIYUN_ACCESS_KEY_SECRET, cn-hangzhou ) code cls.generate_verification_code() request CommonRequest() request.set_accept_format(json) request.set_domain(dysmsapi.aliyuncs.com) request.set_method(POST) request.set_protocol_type(https) request.set_version(2017-05-25) request.set_action_name(SendSms) request.add_query_param(RegionId, cn-hangzhou) request.add_query_param(PhoneNumbers, phone_number) request.add_query_param(SignName, settings.ALIYUN_SMS_SIGN_NAME) request.add_query_param(TemplateCode, settings.ALIYUN_SMS_TEMPLATE_CODE) request.add_query_param(TemplateParam, f{{code:{code}}}) response client.do_action_with_exception(request) return { success: True, code: code, raw_response: response.decode(utf-8) }这个服务类具有以下特点验证码生成与发送逻辑分离完全基于Django配置系统返回结构化的响应数据易于扩展和测试3. 视图层与API设计现在我们将服务集成到Django的视图层。首先创建一个专门处理短信验证的apppython manage.py startapp sms_auth然后实现API视图# sms_auth/views.py from django.views.decorators.csrf import csrf_exempt from django.http import JsonResponse from services.sms_service import SMSService import json csrf_exempt def send_verification_code(request): if request.method POST: try: data json.loads(request.body) phone_number data.get(phone_number) if not phone_number: return JsonResponse( {error: 手机号不能为空}, status400 ) result SMSService.send_verification_code(phone_number) return JsonResponse(result) except Exception as e: return JsonResponse( {error: str(e)}, status500 )配置URL路由# sms_auth/urls.py from django.urls import path from .views import send_verification_code urlpatterns [ path(send-code/, send_verification_code, namesend_verification_code), ] # 项目主urls.py from django.urls import include, path urlpatterns [ path(api/sms/, include(sms_auth.urls)), ]4. 高级优化与最佳实践4.1 异步发送实现同步发送短信会阻塞请求响应我们可以使用Celery实现异步发送# tasks.py from celery import shared_task from services.sms_service import SMSService shared_task def async_send_verification_code(phone_number): return SMSService.send_verification_code(phone_number) # 修改视图 csrf_exempt def send_verification_code(request): if request.method POST: try: data json.loads(request.body) phone_number data.get(phone_number) if not phone_number: return JsonResponse( {error: 手机号不能为空}, status400 ) async_send_verification_code.delay(phone_number) return JsonResponse({success: True}) except Exception as e: return JsonResponse( {error: str(e)}, status500 )4.2 验证码验证与存储发送验证码只是第一步我们还需要验证用户输入的验证码# services/sms_service.py from django.core.cache import cache class SMSService: classmethod def send_verification_code(cls, phone_number): # ...之前的发送逻辑... cache.set(fsms_verification_{phone_number}, code, timeout300) # 5分钟有效期 return { success: True, code: code, raw_response: response.decode(utf-8) } classmethod def verify_code(cls, phone_number, code): cached_code cache.get(fsms_verification_{phone_number}) return cached_code code对应的验证API# views.py csrf_exempt def verify_code(request): if request.method POST: try: data json.loads(request.body) phone_number data.get(phone_number) code data.get(code) if not all([phone_number, code]): return JsonResponse( {error: 参数不完整}, status400 ) is_valid SMSService.verify_code(phone_number, code) return JsonResponse({is_valid: is_valid}) except Exception as e: return JsonResponse( {error: str(e)}, status500 )4.3 频率限制与安全防护防止短信轰炸是必须考虑的安全措施# decorators.py from django.core.cache import cache from django.http import JsonResponse from functools import wraps import json def sms_rate_limit(max_requests1, timeout60): def decorator(view_func): wraps(view_func) def wrapped_view(request, *args, **kwargs): try: data json.loads(request.body) phone_number data.get(phone_number) if not phone_number: return JsonResponse( {error: 手机号不能为空}, status400 ) cache_key fsms_rate_limit_{phone_number} request_count cache.get(cache_key, 0) if request_count max_requests: return JsonResponse( {error: 请求过于频繁请稍后再试}, status429 ) cache.set(cache_key, request_count 1, timeouttimeout) return view_func(request, *args, **kwargs) except Exception as e: return JsonResponse( {error: str(e)}, status500 ) return wrapped_view return decorator # 使用装饰器 sms_rate_limit(max_requests3, timeout300) csrf_exempt def send_verification_code(request): # ...原有逻辑...5. 测试与部署5.1 本地测试方案编写单元测试确保功能正常# tests.py from django.test import TestCase from services.sms_service import SMSService from django.core.cache import cache class SMSServiceTestCase(TestCase): def test_code_generation(self): code SMSService.generate_verification_code() self.assertEqual(len(code), 4) self.assertTrue(code.isdigit()) def test_code_verification(self): test_phone 13800000000 test_code 1234 cache.set(fsms_verification_{test_phone}, test_code) self.assertTrue( SMSService.verify_code(test_phone, test_code) ) self.assertFalse( SMSService.verify_code(test_phone, 0000) )5.2 生产环境部署要点部署到生产环境时需要注意密钥管理使用专门的密钥管理服务如AWS KMS、阿里云KMS或者使用Django的加密设置功能监控与日志# services/sms_service.py import logging logger logging.getLogger(__name__) class SMSService: classmethod def send_verification_code(cls, phone_number): try: # ...发送逻辑... logger.info(f短信发送成功: {phone_number}) return result except Exception as e: logger.error(f短信发送失败: {str(e)}) raise性能优化使用连接池管理阿里云客户端实例对于高并发场景考虑批量发送接口多环境配置# settings/production.py ALIYUN_SMS_SIGN_NAME 正式环境签名 ALIYUN_SMS_TEMPLATE_CODE SMS_正式模板 # settings/development.py ALIYUN_SMS_SIGN_NAME 测试环境签名 ALIYUN_SMS_TEMPLATE_CODE SMS_测试模板5.3 前端集成示例前端如何调用这些API这里是一个简单的React示例async function sendVerificationCode(phoneNumber) { try { const response await fetch(/api/sms/send-code/, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ phone_number: phoneNumber }), }); if (!response.ok) { throw new Error(await response.text()); } return await response.json(); } catch (error) { console.error(发送验证码失败:, error); throw error; } } async function verifyCode(phoneNumber, code) { try { const response await fetch(/api/sms/verify-code/, { method: POST, headers: { Content-Type: application/json, }, body: JSON.stringify({ phone_number: phoneNumber, code: code, }), }); if (!response.ok) { throw new Error(await response.text()); } return await response.json(); } catch (error) { console.error(验证码验证失败:, error); throw error; } }在实际项目中我们通常会添加倒计时、错误提示等增强用户体验的功能。

更多文章