【MongoDB实战】8.2 简易商品管理系统-核心功能实现

简易商品管理系统（MongoDB + Flask 实现）

本文将完整实现基于MongoDB + Python Flask的简易商品管理系统，聚焦 MongoDB 核心操作（CRUD + 聚合），同时封装业务层逻辑（库存校验、分类统计），并提供 HTTP 接口层。

一、环境准备

1. 安装依赖

pipinstallpymongo flask python-dotenv

2. 配置文件（.env）

存放 MongoDB 连接信息，避免硬编码：

# MongoDB 连接配置MONGO_URI=mongodb://localhost:27017/MONGO_DB_NAME=product_managementMONGO_COLLECTION_NAME=products

二、核心代码实现

整体结构

# main.pyimportosfromdotenvimportload_dotenvfromflaskimportFlask,request,jsonifyfrompymongoimportMongoClientfrombson.objectidimportObjectIdfrombson.errorsimportInvalidId# 加载环境变量load_dotenv()# 初始化 Flask 应用app=Flask(__name__)# ====================== 数据层：MongoDB 操作封装 ======================classProductDB:def__init__(self):"""初始化 MongoDB 连接"""self.client=MongoClient(os.getenv("MONGO_URI"))self.db=self.client[os.getenv("MONGO_DB_NAME")]self.collection=self.db[os.getenv("MONGO_COLLECTION_NAME")]# 1. 创建商品（C）defcreate_product(self,product_data):""" 新增商品 :param product_data: 商品字典（name, category, price, stock, description） :return: 新增商品的 ID """result=self.collection.insert_one(product_data)returnstr(result.inserted_id)# 2. 查询商品（R）defget_product_by_id(self,product_id):"""根据 ID 查询单个商品"""try:product=self.collection.find_one({"_id":ObjectId(product_id)})ifproduct:# 转换 ObjectId 为字符串，方便 JSON 序列化product["_id"]=str(product["_id"])returnproductexceptInvalidId:returnNonedefget_products_by_condition(self,condition=None):"""根据条件查询多个商品（默认查询全部）"""condition=conditionor{}products=[]forproductinself.collection.find(condition):product["_id"]=str(product["_id"])products.append(product)returnproducts# 3. 更新商品（U）defupdate_product(self,product_id,update_data):""" 更新商品信息 :param product_id: 商品 ID :param update_data: 要更新的字段字典 :return: 是否更新成功（True/False） """<