《Python 购物车之商家部分代码》
在电商系统中，购物车功能是连接用户与商品的核心模块之一。本文将聚焦于购物车系统的商家端实现，通过Python语言构建一个完整的商家管理接口，涵盖商品上下架、库存管理、订单处理及数据统计等功能。商家端代码需具备高可扩展性、数据安全性和异常处理能力，以适应不同规模的电商业务需求。
一、系统架构设计
商家端购物车系统采用分层架构设计，分为数据访问层（DAO）、业务逻辑层（Service）和接口展示层（API）。数据存储使用SQLite轻量级数据库，适合中小型商家快速部署。整体流程如下：
- 商家通过API提交操作请求
- Service层处理业务逻辑并调用DAO层
- DAO层执行数据库操作并返回结果
- API层将结果封装为JSON响应
二、数据库模型设计
核心数据表包括商品表（products）、库存表（inventory）、订单表（orders）和商家表（merchants）。以下是关键表的SQL定义：
-- 商品表
CREATE TABLE products (
product_id INTEGER PRIMARY KEY AUTOINCREMENT,
merchant_id INTEGER NOT NULL,
name VARCHAR(100) NOT NULL,
description TEXT,
price DECIMAL(10,2) NOT NULL,
category VARCHAR(50),
create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_active BOOLEAN DEFAULT TRUE,
FOREIGN KEY (merchant_id) REFERENCES merchants(merchant_id)
);
-- 库存表
CREATE TABLE inventory (
inventory_id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id INTEGER NOT NULL,
quantity INTEGER NOT NULL DEFAULT 0,
warning_threshold INTEGER DEFAULT 10,
last_update TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (product_id) REFERENCES products(product_id)
);
-- 订单表（简化版）
CREATE TABLE orders (
order_id INTEGER PRIMARY KEY AUTOINCREMENT,
merchant_id INTEGER NOT NULL,
product_id INTEGER NOT NULL,
quantity INTEGER NOT NULL,
total_price DECIMAL(10,2) NOT NULL,
status VARCHAR(20) DEFAULT 'pending',
create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (merchant_id) REFERENCES merchants(merchant_id),
FOREIGN KEY (product_id) REFERENCES products(product_id)
);
三、核心功能实现
1. 商品管理模块
商品管理包含增删改查和上下架功能。以下是关键代码实现：
import sqlite3
from datetime import datetime
class ProductManager:
def __init__(self, db_path='ecommerce.db'):
self.conn = sqlite3.connect(db_path)
self.cursor = self.conn.cursor()
def add_product(self, merchant_id, name, price, category, description=''):
try:
self.cursor.execute(
"INSERT INTO products (merchant_id, name, price, category, description) "
"VALUES (?, ?, ?, ?, ?)",
(merchant_id, name, price, category, description)
)
product_id = self.cursor.lastrowid
# 初始化库存
self.cursor.execute(
"INSERT INTO inventory (product_id, quantity) VALUES (?, ?)",
(product_id, 0)
)
self.conn.commit()
return product_id
except sqlite3.Error as e:
self.conn.rollback()
raise Exception(f"添加商品失败: {str(e)}")
def update_product_status(self, product_id, is_active):
try:
self.cursor.execute(
"UPDATE products SET is_active = ? WHERE product_id = ?",
(is_active, product_id)
)
self.conn.commit()
return True
except sqlite3.Error as e:
self.conn.rollback()
raise Exception(f"更新商品状态失败: {str(e)}")
def get_products_by_merchant(self, merchant_id, is_active=None):
query = "SELECT * FROM products WHERE merchant_id = ?"
params = [merchant_id]
if is_active is not None:
query += " AND is_active = ?"
params.append(is_active)
self.cursor.execute(query, params)
return self.cursor.fetchall()
2. 库存管理模块
库存模块需处理并发更新和库存预警功能：
class InventoryManager:
def __init__(self, db_path='ecommerce.db'):
self.conn = sqlite3.connect(db_path)
self.conn.execute("PRAGMA busy_timeout = 3000") # 设置超时防止死锁
def update_stock(self, product_id, quantity_change):
try:
# 使用事务确保原子性
with self.conn:
# 获取当前库存
self.cursor.execute(
"SELECT quantity FROM inventory WHERE product_id = ?",
(product_id,)
)
result = self.cursor.fetchone()
if not result:
raise ValueError("商品不存在")
current_stock = result[0]
new_stock = current_stock + quantity_change
if new_stock 3. 订单处理模块
订单处理需实现状态流转和商家确认功能：
class OrderManager:
def __init__(self, db_path='ecommerce.db'):
self.conn = sqlite3.connect(db_path)
def get_pending_orders(self, merchant_id):
self.cursor.execute(
"SELECT o.*, p.name, p.price "
"FROM orders o JOIN products p ON o.product_id = p.product_id "
"WHERE o.merchant_id = ? AND o.status = 'pending'",
(merchant_id,)
)
return self.cursor.fetchall()
def confirm_order(self, order_id):
try:
with self.conn:
# 检查订单是否存在且属于当前商家
self.cursor.execute(
"SELECT merchant_id, product_id, quantity "
"FROM orders WHERE order_id = ? AND status = 'pending'",
(order_id,)
)
order_data = self.cursor.fetchone()
if not order_data:
raise ValueError("订单不存在或已处理")
merchant_id, product_id, quantity = order_data
# 更新订单状态
self.cursor.execute(
"UPDATE orders SET status = 'confirmed' WHERE order_id = ?",
(order_id,)
)
# 减少库存
inventory_mgr = InventoryManager()
inventory_mgr.update_stock(product_id, -quantity)
return True
except sqlite3.Error as e:
raise Exception(f"订单确认失败: {str(e)}")
4. 数据统计模块
商家需要销售数据统计功能：
class SalesAnalyzer:
def __init__(self, db_path='ecommerce.db'):
self.conn = sqlite3.connect(db_path)
def get_daily_sales(self, merchant_id, start_date, end_date):
self.cursor.execute(
"SELECT DATE(o.create_time) as sale_date, SUM(o.total_price) as total_sales, "
"COUNT(o.order_id) as order_count "
"FROM orders o "
"WHERE o.merchant_id = ? AND o.status = 'confirmed' "
"AND DATE(o.create_time) BETWEEN ? AND ? "
"GROUP BY sale_date "
"ORDER BY sale_date",
(merchant_id, start_date, end_date)
)
return self.cursor.fetchall()
def get_top_selling_products(self, merchant_id, limit=5):
self.cursor.execute(
"SELECT p.product_id, p.name, SUM(o.quantity) as total_sold "
"FROM orders o JOIN products p ON o.product_id = p.product_id "
"WHERE o.merchant_id = ? AND o.status = 'confirmed' "
"GROUP BY p.product_id, p.name "
"ORDER BY total_sold DESC "
"LIMIT ?",
(merchant_id, limit)
)
return self.cursor.fetchall()
四、API接口设计
使用Flask框架构建RESTful API，示例代码如下：
from flask import Flask, request, jsonify
app = Flask(__name__)
# 初始化管理器
product_mgr = ProductManager()
inventory_mgr = InventoryManager()
order_mgr = OrderManager()
sales_analyzer = SalesAnalyzer()
@app.route('/api/merchant/products', methods=['POST'])
def add_product():
data = request.json
try:
product_id = product_mgr.add_product(
data['merchant_id'],
data['name'],
data['price'],
data['category'],
data.get('description', '')
)
return jsonify({"success": True, "product_id": product_id}), 201
except Exception as e:
return jsonify({"success": False, "error": str(e)}), 400
@app.route('/api/merchant/products/', methods=['PUT'])
def update_product_status(product_id):
data = request.json
try:
product_mgr.update_product_status(product_id, data['is_active'])
return jsonify({"success": True}), 200
except Exception as e:
return jsonify({"success": False, "error": str(e)}), 400
@app.route('/api/merchant/inventory/', methods=['POST'])
def adjust_inventory(product_id):
data = request.json
try:
new_stock = inventory_mgr.update_stock(product_id, data['quantity_change'])
return jsonify({"success": True, "current_stock": new_stock}), 200
except Exception as e:
return jsonify({"success": False, "error": str(e)}), 400
@app.route('/api/merchant/orders', methods=['GET'])
def get_pending_orders():
merchant_id = request.args.get('merchant_id', type=int)
try:
orders = order_mgr.get_pending_orders(merchant_id)
# 转换为字典列表
result = []
for order in orders:
result.append({
"order_id": order[0],
"product_id": order[1],
"quantity": order[2],
"total_price": order[3],
"status": order[4],
"create_time": order[5],
"product_name": order[6],
"product_price": order[7]
})
return jsonify({"success": True, "orders": result}), 200
except Exception as e:
return jsonify({"success": False, "error": str(e)}), 400
if __name__ == '__main__':
app.run(debug=True)
五、系统优化与扩展
1. 性能优化：
- 添加数据库索引：在product_id、merchant_id等字段上创建索引
- 实现缓存机制：使用Redis缓存热门商品数据
- 异步处理：使用Celery处理库存预警通知等耗时操作
2. 安全增强：
- 添加JWT认证：确保只有授权商家能访问API
- 输入验证：对所有用户输入进行严格校验
- SQL注入防护：使用参数化查询
3. 功能扩展：
- 添加商品图片管理功能
- 实现多级商品分类
- 增加商家财务报表生成功能
六、完整示例：商家端操作流程
以下是一个完整的商家操作示例：
# 1. 商家添加商品
POST /api/merchant/products
{
"merchant_id": 1,
"name": "Python编程从入门到实践",
"price": 89.90,
"category": "书籍",
"description": "Python编程权威指南"
}
# 2. 商家调整库存
POST /api/merchant/inventory/1
{
"quantity_change": 50 # 进货50本
}
# 3. 商家查看待处理订单
GET /api/merchant/orders?merchant_id=1
# 4. 商家确认订单
PUT /api/merchant/orders/1001/confirm
# 5. 商家查看销售统计
GET /api/merchant/sales/daily?merchant_id=1&start_date=2023-01-01&end_date=2023-01-31
七、总结与展望
本文实现的商家端购物车系统涵盖了电商业务的核心功能，采用分层架构设计保证了系统的可维护性和扩展性。通过SQLite数据库实现了轻量级部署，同时预留了向MySQL/PostgreSQL迁移的接口。未来可以进一步集成微服务架构，将商品服务、订单服务和库存服务拆分为独立服务，提高系统的可伸缩性。
关键词：Python购物车系统、商家端开发、电商系统架构、Flask API、数据库设计、库存管理、订单处理、销售统计
简介：本文详细介绍了使用Python开发电商购物车系统商家端的全过程，包括数据库设计、核心功能实现（商品管理、库存控制、订单处理、数据分析）、API接口构建以及系统优化方案。通过分层架构和模块化设计，实现了高可扩展性的商家管理后台，适用于中小型电商业务场景。
《Python 购物车之商家部分代码.doc》
将本文以doc文档格式下载到电脑，方便收藏和打印
推荐度：
点击下载文档
