当前位置： 首页 > news >正文

Python全栈项目：基于Django的电子商务平台开发

news 2025/11/7 7:24:01

项目概述

随着互联网经济的快速发展，电子商务已成为商业活动的重要组成部分。本文将详细介绍如何使用Python的Django框架开发一个功能完整的电子商务平台，涵盖从项目架构设计到具体实现的全过程。

这个项目将实现一个现代化的B2C电商平台，包括商品展示、购物车、订单管理、支付集成、用户系统等核心功能。通过这个项目，我们可以深入理解全栈开发的完整流程。

技术栈选型

后端技术

  • Django 4.2+: 成熟稳定的Web框架，提供完善的ORM、认证系统和管理后台
  • Django REST Framework: 构建RESTful API，实现前后端分离
  • Celery: 处理异步任务，如发送邮件、生成报表
  • Redis: 缓存系统和消息队列
  • PostgreSQL: 主数据库，提供强大的数据完整性保证

前端技术

  • Vue.js 3 / React: 构建动态用户界面
  • Bootstrap 5 / Tailwind CSS: 响应式UI框架
  • Axios: HTTP客户端，处理API请求
  • Webpack: 模块打包工具

第三方集成

  • Stripe / PayPal: 支付网关
  • Aliyun OSS / AWS S3: 对象存储服务
  • SendGrid / 阿里云邮件: 邮件服务
  • Elasticsearch: 全文搜索引擎

项目架构设计

整体架构

ecommerce-platform/
├── apps/
│   ├── users/              # 用户管理模块
│   ├── products/           # 商品管理模块
│   ├── orders/             # 订单管理模块
│   ├── cart/               # 购物车模块
│   ├── payment/            # 支付模块
│   └── reviews/            # 评论模块
├── config/
│   ├── settings/
│   │   ├── base.py
│   │   ├── development.py
│   │   └── production.py
│   ├── urls.py
│   └── wsgi.py
├── static/
├── media/
├── templates/
├── requirements/
└── manage.py

数据库设计原则

采用规范化设计，确保数据一致性和完整性。主要遵循以下原则：

  1. 第三范式设计，避免数据冗余
  2. 合理使用外键关系
  3. 为高频查询字段建立索引
  4. 敏感信息加密存储

核心功能模块实现

1. 用户管理系统

用户系统是电商平台的基础，需要实现完整的认证和授权机制。

# users/models.py
from django.contrib.auth.models import AbstractUser
from django.db import modelsclass User(AbstractUser):"""扩展用户模型"""phone = models.CharField(max_length=11, unique=True, null=True, blank=True)avatar = models.ImageField(upload_to='avatars/', null=True, blank=True)birthday = models.DateField(null=True, blank=True)is_verified = models.BooleanField(default=False)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_nameclass UserAddress(models.Model):"""用户收货地址"""user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='addresses')receiver = models.CharField(max_length=50, verbose_name='收货人')province = models.CharField(max_length=50, verbose_name='省份')city = models.CharField(max_length=50, verbose_name='城市')district = models.CharField(max_length=50, verbose_name='区县')detail = models.CharField(max_length=200, verbose_name='详细地址')phone = models.CharField(max_length=11, verbose_name='联系电话')is_default = models.BooleanField(default=False, verbose_name='默认地址')class Meta:db_table = 'user_addresses'verbose_name = '用户地址'verbose_name_plural = verbose_name

实现JWT认证机制：

# users/views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from rest_framework_simplejwt.tokens import RefreshToken
from django.contrib.auth import authenticateclass UserLoginView(APIView):"""用户登录"""def post(self, request):username = request.data.get('username')password = request.data.get('password')user = authenticate(username=username, password=password)if user is not None:refresh = RefreshToken.for_user(user)return Response({'refresh': str(refresh),'access': str(refresh.access_token),'user': {'id': user.id,'username': user.username,'email': user.email,}})return Response({'error': '用户名或密码错误'}, status=status.HTTP_401_UNAUTHORIZED)

2. 商品管理模块

商品模块是电商平台的核心，需要支持分类、SPU/SKU、库存管理等功能。

# products/models.py
from django.db import models
from django.contrib.auth import get_user_modelUser = get_user_model()class Category(models.Model):"""商品分类"""name = models.CharField(max_length=50, verbose_name='分类名称')parent = models.ForeignKey('self', on_delete=models.CASCADE, null=True, blank=True, related_name='children')sort = models.IntegerField(default=0, verbose_name='排序')is_active = models.BooleanField(default=True, verbose_name='是否启用')class Meta:db_table = 'categories'verbose_name = '商品分类'verbose_name_plural = verbose_nameclass Product(models.Model):"""商品SPU"""name = models.CharField(max_length=200, verbose_name='商品名称')category = models.ForeignKey(Category, on_delete=models.PROTECT, related_name='products')description = models.TextField(verbose_name='商品描述')main_image = models.ImageField(upload_to='products/', verbose_name='主图')is_on_sale = models.BooleanField(default=True, verbose_name='是否上架')created_at = models.DateTimeField(auto_now_add=True)updated_at = models.DateTimeField(auto_now=True)class Meta:db_table = 'products'verbose_name = '商品'verbose_name_plural = verbose_nameindexes = [models.Index(fields=['category', 'is_on_sale']),]class ProductSKU(models.Model):"""商品SKU"""product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name='skus')sku_code = models.CharField(max_length=50, unique=True, verbose_name='SKU编码')price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='价格')stock = models.IntegerField(default=0, verbose_name='库存')specs = models.JSONField(verbose_name='规格信息')image = models.ImageField(upload_to='skus/', null=True, verbose_name='SKU图片')class Meta:db_table = 'product_skus'verbose_name = '商品SKU'verbose_name_plural = verbose_nameclass ProductImage(models.Model):"""商品图片"""product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name='images')image = models.ImageField(upload_to='products/gallery/')sort = models.IntegerField(default=0)class Meta:db_table = 'product_images'ordering = ['sort']

实现商品搜索功能：

# products/views.py
from rest_framework.generics import ListAPIView
from rest_framework.filters import SearchFilter, OrderingFilter
from django_filters.rest_framework import DjangoFilterBackend
from .models import Product
from .serializers import ProductSerializer
from .filters import ProductFilterclass ProductListView(ListAPIView):"""商品列表"""queryset = Product.objects.filter(is_on_sale=True).select_related('category')serializer_class = ProductSerializerfilter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]filterset_class = ProductFiltersearch_fields = ['name', 'description']ordering_fields = ['created_at', 'price']ordering = ['-created_at']

3. 购物车系统

购物车需要支持未登录用户（基于Session）和已登录用户（基于数据库）两种场景。

# cart/models.py
from django.db import models
from django.contrib.auth import get_user_model
from apps.products.models import ProductSKUUser = get_user_model()class CartItem(models.Model):"""购物车商品"""user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='cart_items')sku = models.ForeignKey(ProductSKU, on_delete=models.CASCADE)quantity = models.IntegerField(default=1, verbose_name='数量')is_selected = models.BooleanField(default=True, verbose_name='是否选中')created_at = models.DateTimeField(auto_now_add=True)updated_at = models.DateTimeField(auto_now=True)class Meta:db_table = 'cart_items'unique_together = ['user', 'sku']verbose_name = '购物车商品'verbose_name_plural = verbose_name

购物车业务逻辑：

# cart/services.py
from django.core.cache import cache
from .models import CartItem
from apps.products.models import ProductSKUclass CartService:"""购物车服务"""@staticmethoddef add_to_cart(user, sku_id, quantity=1):"""添加商品到购物车"""try:sku = ProductSKU.objects.get(id=sku_id)if sku.stock < quantity:return None, "库存不足"cart_item, created = CartItem.objects.get_or_create(user=user,sku=sku,defaults={'quantity': quantity})if not created:cart_item.quantity += quantityif cart_item.quantity > sku.stock:return None, "超出库存限制"cart_item.save()return cart_item, Noneexcept ProductSKU.DoesNotExist:return None, "商品不存在"@staticmethoddef get_cart_items(user):"""获取购物车商品列表"""return CartItem.objects.filter(user=user).select_related('sku__product').order_by('-created_at')@staticmethoddef calculate_total(user):"""计算购物车总价"""items = CartItem.objects.filter(user=user, is_selected=True)total = sum(item.sku.price * item.quantity for item in items)return total

4. 订单管理系统

订单系统需要处理订单创建、状态流转、库存扣减等复杂业务逻辑。

# orders/models.py
from django.db import models
from django.contrib.auth import get_user_model
from apps.products.models import ProductSKUUser = get_user_model()class Order(models.Model):"""订单"""STATUS_CHOICES = (('PENDING', '待支付'),('PAID', '已支付'),('SHIPPED', '已发货'),('COMPLETED', '已完成'),('CANCELLED', '已取消'),('REFUNDING', '退款中'),('REFUNDED', '已退款'),)order_no = models.CharField(max_length=32, unique=True, verbose_name='订单号')user = models.ForeignKey(User, on_delete=models.PROTECT, related_name='orders')status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='PENDING', verbose_name='订单状态')total_amount = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='订单总额')shipping_fee = models.DecimalField(max_digits=10, decimal_places=2, default=0, verbose_name='运费')actual_amount = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='实付金额')# 收货信息receiver = models.CharField(max_length=50, verbose_name='收货人')phone = models.CharField(max_length=11, verbose_name='联系电话')province = models.CharField(max_length=50, verbose_name='省份')city = models.CharField(max_length=50, verbose_name='城市')district = models.CharField(max_length=50, verbose_name='区县')address = models.CharField(max_length=200, verbose_name='详细地址')remark = models.TextField(blank=True, verbose_name='备注')paid_at = models.DateTimeField(null=True, blank=True, verbose_name='支付时间')shipped_at = models.DateTimeField(null=True, blank=True, verbose_name='发货时间')completed_at = models.DateTimeField(null=True, blank=True, verbose_name='完成时间')created_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')updated_at = models.DateTimeField(auto_now=True, verbose_name='更新时间')class Meta:db_table = 'orders'verbose_name = '订单'verbose_name_plural = verbose_nameordering = ['-created_at']indexes = [models.Index(fields=['user', 'status']),models.Index(fields=['order_no']),]class OrderItem(models.Model):"""订单商品"""order = models.ForeignKey(Order, on_delete=models.CASCADE, related_name='items')sku = models.ForeignKey(ProductSKU, on_delete=models.PROTECT)product_name = models.CharField(max_length=200, verbose_name='商品名称')product_image = models.CharField(max_length=500, verbose_name='商品图片')specs = models.JSONField(verbose_name='规格信息')price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='单价')quantity = models.IntegerField(verbose_name='数量')subtotal = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='小计')class Meta:db_table = 'order_items'verbose_name = '订单商品'verbose_name_plural = verbose_name

订单创建服务（使用事务确保数据一致性）：

# orders/services.py
from django.db import transaction
from django.utils import timezone
import uuid
from .models import Order, OrderItem
from apps.cart.models import CartItem
from apps.products.models import ProductSKUclass OrderService:"""订单服务"""@staticmethod@transaction.atomicdef create_order(user, address_id, cart_item_ids):"""创建订单"""# 获取购物车商品cart_items = CartItem.objects.filter(id__in=cart_item_ids,user=user,is_selected=True).select_related('sku__product')if not cart_items:raise ValueError("没有选中的商品")# 检查库存并锁定库存for item in cart_items:sku = ProductSKU.objects.select_for_update().get(id=item.sku.id)if sku.stock < item.quantity:raise ValueError(f"商品{sku.product.name}库存不足")# 获取收货地址from apps.users.models import UserAddressaddress = UserAddress.objects.get(id=address_id, user=user)# 计算订单金额total_amount = sum(item.sku.price * item.quantity for item in cart_items)shipping_fee = 0  # 运费计算逻辑actual_amount = total_amount + shipping_fee# 创建订单order = Order.objects.create(order_no=OrderService.generate_order_no(),user=user,total_amount=total_amount,shipping_fee=shipping_fee,actual_amount=actual_amount,receiver=address.receiver,phone=address.phone,province=address.province,city=address.city,district=address.district,address=address.detail,)# 创建订单商品order_items = []for item in cart_items:order_items.append(OrderItem(order=order,sku=item.sku,product_name=item.sku.product.name,product_image=item.sku.image.url if item.sku.image else '',specs=item.sku.specs,price=item.sku.price,quantity=item.quantity,subtotal=item.sku.price * item.quantity))# 扣减库存item.sku.stock -= item.quantityitem.sku.save()OrderItem.objects.bulk_create(order_items)# 清空购物车cart_items.delete()return order@staticmethoddef generate_order_no():"""生成订单号"""return timezone.now().strftime('%Y%m%d%H%M%S') + str(uuid.uuid4().hex[:8]).upper()

5. 支付集成

集成第三方支付平台，以Stripe为例：

# payment/services.py
import stripe
from django.conf import settings
from apps.orders.models import Orderstripe.api_key = settings.STRIPE_SECRET_KEYclass PaymentService:"""支付服务"""@staticmethoddef create_payment_intent(order_id):"""创建支付意图"""try:order = Order.objects.get(id=order_id, status='PENDING')intent = stripe.PaymentIntent.create(amount=int(order.actual_amount * 100),  # 转换为分currency='cny',metadata={'order_id': order.id, 'order_no': order.order_no},description=f'订单支付 - {order.order_no}')return {'client_secret': intent.client_secret,'payment_intent_id': intent.id}except Order.DoesNotExist:raise ValueError("订单不存在")@staticmethod@transaction.atomicdef handle_payment_success(payment_intent_id):"""处理支付成功"""intent = stripe.PaymentIntent.retrieve(payment_intent_id)order_id = intent.metadata.get('order_id')order = Order.objects.select_for_update().get(id=order_id)order.status = 'PAID'order.paid_at = timezone.now()order.save()# 发送支付成功通知from .tasks import send_payment_notificationsend_payment_notification.delay(order.id)return order

6. 异步任务处理

使用Celery处理耗时任务：

# payment/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from django.conf import settings
from apps.orders.models import Order@shared_task
def send_payment_notification(order_id):"""发送支付成功通知"""try:order = Order.objects.get(id=order_id)subject = f'订单支付成功 - {order.order_no}'message = f'''尊敬的{order.receiver}：您的订单{order.order_no}已支付成功！订单金额：¥{order.actual_amount}我们会尽快为您安排发货。'''send_mail(subject,message,settings.DEFAULT_FROM_EMAIL,[order.user.email],fail_silently=False,)return f"通知邮件已发送至{order.user.email}"except Order.DoesNotExist:return f"订单{order_id}不存在"@shared_task
def generate_sales_report():"""生成销售报表"""from django.utils import timezonefrom datetime import timedeltaend_date = timezone.now()start_date = end_date - timedelta(days=30)orders = Order.objects.filter(status='COMPLETED',completed_at__range=[start_date, end_date])total_sales = sum(order.actual_amount for order in orders)total_orders = orders.count()# 生成报表逻辑report_data = {'period': f"{start_date.date()} 至 {end_date.date()}",'total_sales': total_sales,'total_orders': total_orders,'average_order_value': total_sales / total_orders if total_orders > 0 else 0}return report_data

前端实现

Vue.js 商品列表组件

<!-- ProductList.vue -->
<template><div class="product-list"><div class="filters"><select v-model="filters.category" @change="loadProducts"><option value="">全部分类</option><option v-for="cat in categories" :key="cat.id" :value="cat.id">{{ cat.name }}</option></select><select v-model="filters.ordering" @change="loadProducts"><option value="-created_at">最新</option><option value="price">价格从低到高</option><option value="-price">价格从高到低</option></select></div><div class="products-grid"><div v-for="product in products" :key="product.id" class="product-card"><img :src="product.main_image" :alt="product.name" /><h3>{{ product.name }}</h3><p class="price">¥{{ product.min_price }}</p><button @click="addToCart(product)" class="btn-add-cart">加入购物车</button></div></div><div class="pagination"><button @click="prevPage" :disabled="!hasPrev">上一页</button><span>第 {{ currentPage }} 页</span><button @click="nextPage" :disabled="!hasNext">下一页</button></div></div>
</template><script>
import { ref, reactive, onMounted } from 'vue'
import axios from 'axios'export default {name: 'ProductList',setup() {const products = ref([])const categories = ref([])const currentPage = ref(1)const hasPrev = ref(false)const hasNext = ref(false)const filters = reactive({category: '',ordering: '-created_at',search: ''})const loadProducts = async () => {try {const params = {page: currentPage.value,...filters}const response = await axios.get('/api/products/', { params })products.value = response.data.resultshasPrev.value = !!response.data.previoushasNext.value = !!response.data.next} catch (error) {console.error('加载商品失败:', error)}}const loadCategories = async () => {try {const response = await axios.get('/api/categories/')categories.value = response.data} catch (error) {console.error('加载分类失败:', error)}}const addToCart = async (product) => {try {await axios.post('/api/cart/', {sku_id: product.default_sku_id,quantity: 1})alert('添加成功！')} catch (error) {console.error('添加失败:', error)}}const prevPage = () => {if (hasPrev.value) {currentPage.value--loadProducts()}}const nextPage = () => {if (hasNext.value) {currentPage.value++loadProducts()}}onMounted(() => {loadProducts()loadCategories()})return {products,categories,filters,currentPage,hasPrev,hasNext,loadProducts,addToCart,prevPage,nextPage}}
}
</script><style scoped>
.product-list {padding: 20px;
}.products-grid {display: grid;grid-template-columns: repeat(auto-fill, minmax(250px, 1fr));gap: 20px;margin: 20px 0;
}.product-card {border: 1px solid #ddd;border-radius: 8px;padding: 15px;text-align: center;transition: transform 0.2s;
}.product-card:hover {transform: translateY(-5px);box-shadow: 0 4px 12px rgba(0,0,0,0.1);
}.product-card img {width: 100%;height: 200px;object-fit: cover;border-radius: 4px;
}.price {color: #e4393c;font-size: 24px;font-weight: bold;margin: 10px 0;
}.btn-add-cart {background-color: #ff6700;color: white;border: none;padding: 10px 20px;border-radius: 4px;cursor: pointer;width: 100%;
}.btn-add-cart:hover {background-color: #ff5500;
}
</style>

性能优化策略

1. 数据库优化

# 使用select_related和prefetch_related减少查询次数
products = Product.objects.select_related('category').prefetch_related('skus', 'images'
).filter(is_on_sale=True)# 使用索引
class Product(models.Model):class Meta:indexes = [models.Index(fields=['category', '-created_at']),models.Index(fields=['is_on_sale', '-created_at']),]# 使用数据库连接池
DATABASES = {'default': {'ENGINE': 'django.db.backends.postgresql','CONN_MAX_AGE': 600,  # 连接池'OPTIONS': {'connect_timeout': 10,}}
}

2. 缓存策略

# settings.py - Redis缓存配置
CACHES = {'default': {'BACKEND': 'django_redis.cache.RedisCache','LOCATION': 'redis://127.0.0.1:6379/1','OPTIONS': {'CLIENT_CLASS': 'django_redis.client.DefaultClient','PARSER_CLASS': 'redis.connection.HiredisParser','CONNECTION_POOL_KWARGS': {'max_connections': 50},}}
}# 使用缓存
from django.core.cache import cache
from django.views.decorators.cache import cache_page@cache_page(60 * 15)  # 缓存15分钟
def product_list(request):products = Product.objects.filter(is_on_sale=True)return render(request, 'products/list.html', {'products': products})# 手动缓存
def get_hot_products():cache_key = 'hot_products'products = cache.get(cache_key)if products is None:products = list(Product.objects.filter(is_on_sale=True).order_by('-sales')[:10])cache.set(cache_key, products, 60 * 30)  # 缓存30分钟return products

3. 静态资源优化

# settings.py
STATIC_URL = '/static/'
STATIC_ROOT = os.path.join(BASE_DIR, 'staticfiles')# 使用CDN加速
if not DEBUG:STATIC_URL = 'https://cdn.example.com/static/'MEDIA_URL = 'https://cdn.example.com/media/'# 启用Gzip压缩
MIDDLEWARE = ['django.middleware.gzip.GZipMiddleware',# ...
]# 图片压缩和缩略图
from PIL import Image
from io import BytesIO
from django.core.files.uploadedfile import InMemoryUploadedFiledef compress_image(image, quality=85):"""压缩图片"""img = Image.open(image)if img.mode != 'RGB':img = img.convert('RGB')output = BytesIO()img.save(output, format='JPEG', quality=quality, optimize=True)output.seek(0)return InMemoryUploadedFile(output, 'ImageField', f"{image.name.split('.')[0]}.jpg",'image/jpeg', output.tell(), None)

部署方案

1. Docker容器化部署

# Dockerfile
FROM python:3.11-slimWORKDIR /app# 安装系统依赖
RUN apt-get update && apt-get install -y \gcc \postgresql-client \&& rm -rf /var/lib/apt/lists/*# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt# 复制项目文件
COPY . .# 收集静态文件
RUN python manage.py collectstatic --noinputEXPOSE 8000CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "config.wsgi:application"]

# docker-compose.yml
version: '3.8'services:db:image: postgres:15volumes:- postgres_data:/var/lib/postgresql/dataenvironment:POSTGRES_DB: ecommercePOSTGRES_USER: ecommerce_userPOSTGRES_PASSWORD: ${DB_PASSWORD}ports:- "5432:5432"redis:image: redis:7-alpineports:- "6379:6379"web:build: .command: gunicorn config.wsgi:application --bind 0.0.0.0:8000volumes:- .:/app- static_volume:/app/staticfiles- media_volume:/app/mediaports:- "8000:8000"depends_on:- db- redisenvironment:- DEBUG=False- DATABASE_URL=postgresql://ecommerce_user:${DB_PASSWORD}@db:5432/ecommerce- REDIS_URL=redis://redis:6379/0celery:build: .command: celery -A config worker -l infovolumes:- .:/appdepends_on:- db- redisnginx:image: nginx:alpineports:- "80:80"- "443:443"volumes:- ./nginx.conf:/etc/nginx/nginx.conf- static_volume:/app/staticfiles- media_volume:/app/media- ./ssl:/etc/nginx/ssldepends_on:- webvolumes:postgres_data:static_volume:media_volume:

2. Nginx配置

# nginx.conf
upstream django {server web:8000;
}server {listen 80;server_name example.com;return 301 https://$server_name$request_uri;
}server {listen 443 ssl http2;server_name example.com;ssl_certificate /etc/nginx/ssl/cert.pem;ssl_certificate_key /etc/nginx/ssl/key.pem;client_max_body_size 20M;location / {proxy_pass http://django;proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;proxy_set_header Host $host;proxy_set_header X-Forwarded-Proto $scheme;proxy_redirect off;}location /static/ {alias /app/staticfiles/;expires 30d;add_header Cache-Control "public, immutable";}location /media/ {alias /app/media/;expires 7d;}
}

安全措施

1. 基础安全配置

# settings/production.py
DEBUG = FalseALLOWED_HOSTS = ['example.com', 'www.example.com']# HTTPS设置
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SECURE_BROWSER_XSS_FILTER = True
SECURE_CONTENT_TYPE_NOSNIFF = True
X_FRAME_OPTIONS = 'DENY'# 密码验证
AUTH_PASSWORD_VALIDATORS = [{'NAME': 'django.contrib.auth.password_validation.UserAttributeSimilarityValidator',},{'NAME': 'django.contrib.auth.password_validation.MinimumLengthValidator','OPTIONS': {'min_length': 8}},{'NAME': 'django.contrib.auth.password_validation.CommonPasswordValidator',},{'NAME': 'django.contrib.auth.password_validation.NumericPasswordValidator',},
]

2. API限流

# settings.py
REST_FRAMEWORK = {'DEFAULT_THROTTLE_CLASSES': ['rest_framework.throttling.AnonRateThrottle','rest_framework.throttling.UserRateThrottle'],'DEFAULT_THROTTLE_RATES': {'anon': '100/hour','user': '1000/hour'}
}# 自定义限流类
from rest_framework.throttling import UserRateThrottleclass PaymentRateThrottle(UserRateThrottle):rate = '10/hour'  # 支付接口每小时最多10次

3. SQL注入防护

# 始终使用Django ORM的参数化查询
# 正确方式
products = Product.objects.filter(name__icontains=search_term)# 错误方式 - 容易SQL注入
# cursor.execute(f"SELECT * FROM products WHERE name LIKE '%{search_term}%'")# 如果必须使用原生SQL
from django.db import connection
with connection.cursor() as cursor:cursor.execute("SELECT * FROM products WHERE name LIKE %s", [f'%{search_term}%'])

监控与日志

日志配置

# settings.py
LOGGING = {'version': 1,'disable_existing_loggers': False,'formatters': {'verbose': {'format': '{levelname} {asctime} {module} {message}','style': '{',},},'handlers': {'file': {'level': 'INFO','class': 'logging.handlers.RotatingFileHandler','filename': '/var/log/django/ecommerce.log','maxBytes': 1024 * 1024 * 10,  # 10MB'backupCount': 5,'formatter': 'verbose',},'error_file': {'level': 'ERROR','class': 'logging.handlers.RotatingFileHandler','filename': '/var/log/django/error.log','maxBytes': 1024 * 1024 * 10,'backupCount': 5,'formatter': 'verbose',},},'loggers': {'django': {'handlers': ['file', 'error_file'],'level': 'INFO','propagate': True,},'apps': {'handlers': ['file', 'error_file'],'level': 'INFO','propagate': False,},},
}

测试策略

单元测试示例

# tests/test_orders.py
from django.test import TestCase
from django.contrib.auth import get_user_model
from apps.orders.services import OrderService
from apps.products.models import Product, ProductSKU
from apps.users.models import UserAddressUser = get_user_model()class OrderServiceTest(TestCase):def setUp(self):"""测试前准备"""self.user = User.objects.create_user(username='testuser',email='test@example.com',password='testpass123')self.product = Product.objects.create(name='测试商品',description='测试描述')self.sku = ProductSKU.objects.create(product=self.product,sku_code='TEST001',price=99.99,stock=100)self.address = UserAddress.objects.create(user=self.user,receiver='张三',province='北京',city='北京市',district='朝阳区',detail='xxx街道',phone='13800138000')def test_create_order_success(self):"""测试创建订单成功"""from apps.cart.models import CartItemcart_item = CartItem.objects.create(user=self.user,sku=self.sku,quantity=2)order = OrderService.create_order(user=self.user,address_id=self.address.id,cart_item_ids=[cart_item.id])self.assertIsNotNone(order)self.assertEqual(order.user, self.user)self.assertEqual(order.items.count(), 1)self.assertEqual(order.total_amount, 199.98)# 验证库存扣减self.sku.refresh_from_db()self.assertEqual(self.sku.stock, 98)# 验证购物车清空self.assertEqual(CartItem.objects.filter(user=self.user).count(), 0)def test_create_order_insufficient_stock(self):"""测试库存不足"""from apps.cart.models import CartItemcart_item = CartItem.objects.create(user=self.user,sku=self.sku,quantity=101  # 超过库存)with self.assertRaises(ValueError):OrderService.create_order(user=self.user,address_id=self.address.id,cart_item_ids=[cart_item.id])

API测试

# tests/test_api.py
from rest_framework.test import APITestCase
from rest_framework import status
from django.contrib.auth import get_user_modelUser = get_user_model()class ProductAPITest(APITestCase):def setUp(self):self.user = User.objects.create_user(username='testuser',password='testpass123')def test_get_product_list(self):"""测试获取商品列表"""response = self.client.get('/api/products/')self.assertEqual(response.status_code, status.HTTP_200_OK)def test_add_to_cart_authenticated(self):"""测试已登录用户添加购物车"""self.client.force_authenticate(user=self.user)data = {'sku_id': 1,'quantity': 2}response = self.client.post('/api/cart/', data)self.assertEqual(response.status_code, status.HTTP_201_CREATED)

项目总结

通过这个完整的Django电子商务平台项目，我们实现了以下核心功能：

技术亮点

  1. 前后端分离架构: 使用Django REST Framework构建API，前端使用Vue.js实现响应式界面
  2. 分布式任务处理: 使用Celery处理异步任务，提升系统响应速度
  3. 高性能缓存: 通过Redis实现多层缓存策略，大幅提升查询效率
  4. 安全防护: 实施多层安全措施，包括HTTPS、SQL注入防护、CSRF防护等
  5. 容器化部署: 使用Docker实现一键部署，简化运维流程

可扩展功能

  1. 搜索优化: 集成Elasticsearch实现全文搜索和搜索建议
  2. 推荐系统: 基于用户行为的协同过滤推荐算法
  3. 秒杀系统: 高并发场景下的库存控制和限流策略
  4. 物流跟踪: 集成第三方物流API实现物流实时追踪
  5. 数据分析: 用户行为分析和销售数据可视化
  6. 移动端适配: 开发React Native或Flutter移动应用
  7. 国际化: 支持多语言和多货币

最佳实践

  1. 代码规范: 遵循PEP 8编码规范，使用pylint和black进行代码检查
  2. 版本控制: 使用Git进行版本管理，采用Git Flow工作流
  3. 文档完善: 编写详细的API文档和开发文档
  4. 持续集成: 配置CI/CD流程，自动化测试和部署
  5. 性能监控: 使用APM工具监控应用性能

学习收获

开发这样一个完整的电商平台，可以让我们深入理解：

  • Web应用的完整开发流程
  • 前后端分离架构的设计思想
  • 数据库设计和优化技巧
  • 分布式系统的架构设计
  • 高并发场景的处理方案
  • 生产环境的部署和运维

这个项目为进一步学习微服务架构、云原生应用开发打下了坚实的基础。

参考资源

  • Django官方文档
  • Django REST Framework文档
  • Vue.js官方文档
  • Celery文档
  • Docker文档

项目代码

下载链接

 

http://www.dtcms.com/a/576993.html

相关文章：

  • 网站建设怎么开票网站设计网页设计公司
  • Python实现GPT自动问答与保存
  • 深度强化学习，用神经网络代替 Q-table
  • seo网站建设技巧电线电缆技术支持中山网站建设
  • supabase外键查询语句
  • 【linux端cursor CLI常用命令】
  • 表的增删改查
  • Git 工作区、暂存区和版本库
  • MIT-矩阵链相乘
  • Go语言实战：入门篇-5：函数、服务接口和Swagger UI
  • 国产化Excel处理控件Spire.XLS教程：使用Java将CSV转换为PDF（含格式设置）
  • 【Hot100｜3 LeetCode 128. 最长连续序列】
  • 一键搭建 Coze 智能体对话页面：支持流式输出 + 图片直显，开发效率拉满！
  • 十大免费ae模板网站短视频素材下载网站
  • 那里做直播网站网页打不开是什么问题
  • 论文分享 | AirRoom：物体是关键！革新室内房间重识别的新范式
  • 游戏 IPA 如何防修改，面向开发者的多工具实战（IPA 加固/无源码混淆/Ipa Guard CLI）
  • 从0到1做一个“字母拼词”Unity小游戏（含源码/GIF）- 项目的创建及准备
  • 在扣子上搭建测试用例自动编写智能体
  • 2023年第二十届五一数学建模竞赛-A题 无人机定点投放问题-基于抛体运动的无人机定点投放问题研究
  • 影刀RPA一键生成销售日报！AI智能分析，效率提升1000%[特殊字符]
  • Rust开发实战之密码学基础——哈希计算与对称加密实战
  • 技术解析：清洗无人机在高空清洁中的应用与优势
  • Linux LVM NAT 模式部署实践
  • 使用 DVC（Data Version Control）进行数据版本管理
  • 网站建设选择哪种开发语言最好从哪里下载wordpress
  • 微服务之网关（Spring Cloud Gateway）
  • ES脚本语言Painless介绍
  • 基于MATLAB的雨流计数法疲劳计算GUI可视化系统
  • WiFi 协议精读：IEEE 802.11-2012，IEEE Std 802.11w™-2009: Protected Management Frames