通过给大家分享一个最近我用python写的一个web项目，对于web项目，用java或者php来开发的比较多，不过为了快速的掌握一门编程语言，我们也能够尝试用别的语言来发开web任务，这篇文章就给大家介绍一下 使用 python fastApi框架 来开发一个仓库管理系统。
技术介绍：
后端框架：FastAPI
数据库：mysql8
前端：Vue2+Element UI
主要实现的能力有：
分类管理、产品管理、客户管理、供应商管理、仓库与库位管理、入库出库操作、管理员管理 。
页面介绍：
首页：
产品页：
入库记录：
出库记录：
这个项目一共实现了8个菜单的功能，假设你对这个项目有兴趣允许去看看。（非开源）
https://wwwoop.com/home/Index/projectInfo?goodsId=125&typeParam=1&subKey=-1
部分代码：
from typing import Optional
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from core.response import ResponseModel
from core.exceptions import CustomException
from core.deps import get_current_admin
from db.database import get_db
from models.admin import Admin
from models.product_category import ProductCategory
from schemas.product_category import ProductCategoryCreate, ProductCategoryUpdate, ProductCategoryResponse, ProductCategorySearchRequest
router = APIRouter()
@router.get("/list")
async def get_product_category_list(
db: Session = Depends(get_db),
current_admin: Admin = Depends(get_current_admin),
page: int = Query(1, ge=1),
page_size: int = Query(10, ge=1, le=100),
category_name: Optional[str] = None,
status: Optional[int] = None
):
"""
获取产品分类列表（分页，带搜索条件）
"""
# 构建查询条件，只查询未删除的记录
query = db.query(ProductCategory).filter(ProductCategory.is_deleted == 0)
# 分类名称搜索条件
if category_name:
query = query.filter(ProductCategory.category_name.like(f"%{category_name}%"))
# 状态搜索条件
if status is not None:
query = query.filter(ProductCategory.status == status)
# 按序号和创建时间排序
query = query.order_by(ProductCategory.sort_order.asc(), ProductCategory.created_time.desc())
# 分页查询
total = query.count()
categories = query.offset((page - 1) * page_size).limit(page_size).all()
return ResponseModel.success(
data={
"items": [ProductCategoryResponse.model_validate(category) for category in categories],
"total": total,
"page": page,
"page_size": page_size
},
msg="获取成功"
)
@router.get("/detail/{category_id}")
async def get_product_category_detail(
category_id: int,
db: Session = Depends(get_db),
current_admin: Admin = Depends(get_current_admin)
):
"""
获取产品分类详情
"""
category = db.query(ProductCategory).filter(
ProductCategory.id == category_id,
ProductCategory.is_deleted == 0
).first()
if not category:
raise CustomException(msg="分类不存在")
return ResponseModel.success(
data=ProductCategoryResponse.model_validate(category),
msg="获取成功"
)
@router.post("/create")
async def create_product_category(
category_in: ProductCategoryCreate,
db: Session = Depends(get_db),
current_admin: Admin = Depends(get_current_admin)
):
"""
创建产品分类
"""
# 检查分类名称是否已存在
existing_category = db.query(ProductCategory).filter(
ProductCategory.category_name == category_in.category_name,
ProductCategory.is_deleted == 0
).first()
if existing_category:
raise CustomException(msg="分类名称已存在")
# 创建新分类
db_category = ProductCategory(
category_name=category_in.category_name,
sort_order=category_in.sort_order,
status=category_in.status,
remark=category_in.remark,
created_id=current_admin.id,
updated_id=current_admin.id
)
db.add(db_category)
db.commit()
db.refresh(db_category)
return ResponseModel.success(
data=ProductCategoryResponse.model_validate(db_category),
msg="创建成功"
)
@router.put("/update/{category_id}")
async def update_product_category(
category_id: int,
category_in: ProductCategoryUpdate,
db: Session = Depends(get_db),
current_admin: Admin = Depends(get_current_admin)
):
"""
更新产品分类
"""
# 查找要更新的分类
db_category = db.query(ProductCategory).filter(
ProductCategory.id == category_id,
ProductCategory.is_deleted == 0
).first()
if not db_category:
raise CustomException(msg="分类不存在")
# 如果要更新分类名称，检查是否与其他分类重名
if category_in.category_name and category_in.category_name != db_category.category_name:
existing_category = db.query(ProductCategory).filter(
ProductCategory.category_name == category_in.category_name,
ProductCategory.id != category_id,
ProductCategory.is_deleted == 0
).first()
if existing_category:
raise CustomException(msg="分类名称已存在")
# 更新字段
update_data = category_in.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(db_category, field, value)
# 更新操作者信息
db_category.updated_id = current_admin.id
db.commit()
db.refresh(db_category)
return ResponseModel.success(
data=ProductCategoryResponse.model_validate(db_category),
msg="更新成功"
)
@router.delete("/delete/{category_id}")
async def delete_product_category(
category_id: int,
db: Session = Depends(get_db),
current_admin: Admin = Depends(get_current_admin)
):
"""
删除产品分类（软删除）
"""
# 查找要删除的分类
db_category = db.query(ProductCategory).filter(
ProductCategory.id == category_id,
ProductCategory.is_deleted == 0
).first()
if not db_category:
raise CustomException(msg="分类不存在")
# 软删除
db_category.is_deleted = 1
db_category.updated_id = current_admin.id
db.commit()
return ResponseModel.success(
data=None,
msg="删除成功"
)