fastApi框架开发一个web端仓库管理系统

给大家分享一个最近我用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, ProductCategorySearchRequestrouter = 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.iddb.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 = 1db_category.updated_id = current_admin.iddb.commit()return ResponseModel.success(data=None,msg="删除成功")