135 lines
3.5 KiB
Python
135 lines
3.5 KiB
Python
"""
|
|
品牌和供应商业务服务层
|
|
"""
|
|
from typing import List, Optional, Tuple
|
|
from sqlalchemy.orm import Session
|
|
from app.crud.brand_supplier import brand, supplier
|
|
from app.schemas.brand_supplier import (
|
|
BrandCreate,
|
|
BrandUpdate,
|
|
SupplierCreate,
|
|
SupplierUpdate
|
|
)
|
|
from app.core.exceptions import NotFoundException, AlreadyExistsException
|
|
|
|
|
|
class BrandService:
|
|
"""品牌服务类"""
|
|
|
|
def get_brand(self, db: Session, brand_id: int):
|
|
"""获取品牌详情"""
|
|
obj = brand.get(db, brand_id)
|
|
if not obj:
|
|
raise NotFoundException("品牌")
|
|
return obj
|
|
|
|
def get_brands(
|
|
self,
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 20,
|
|
status: Optional[str] = None,
|
|
keyword: Optional[str] = None
|
|
) -> Tuple[List, int]:
|
|
"""获取品牌列表"""
|
|
return brand.get_multi(db, skip, limit, status, keyword)
|
|
|
|
def create_brand(
|
|
self,
|
|
db: Session,
|
|
obj_in: BrandCreate,
|
|
creator_id: Optional[int] = None
|
|
):
|
|
"""创建品牌"""
|
|
try:
|
|
return brand.create(db, obj_in, creator_id)
|
|
except ValueError as e:
|
|
raise AlreadyExistsException("品牌") from e
|
|
|
|
def update_brand(
|
|
self,
|
|
db: Session,
|
|
brand_id: int,
|
|
obj_in: BrandUpdate,
|
|
updater_id: Optional[int] = None
|
|
):
|
|
"""更新品牌"""
|
|
db_obj = brand.get(db, brand_id)
|
|
if not db_obj:
|
|
raise NotFoundException("品牌")
|
|
return brand.update(db, db_obj, obj_in, updater_id)
|
|
|
|
def delete_brand(
|
|
self,
|
|
db: Session,
|
|
brand_id: int,
|
|
deleter_id: Optional[int] = None
|
|
) -> bool:
|
|
"""删除品牌"""
|
|
if not brand.get(db, brand_id):
|
|
raise NotFoundException("品牌")
|
|
return brand.delete(db, brand_id, deleter_id)
|
|
|
|
|
|
class SupplierService:
|
|
"""供应商服务类"""
|
|
|
|
def get_supplier(self, db: Session, supplier_id: int):
|
|
"""获取供应商详情"""
|
|
obj = supplier.get(db, supplier_id)
|
|
if not obj:
|
|
raise NotFoundException("供应商")
|
|
return obj
|
|
|
|
def get_suppliers(
|
|
self,
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 20,
|
|
status: Optional[str] = None,
|
|
keyword: Optional[str] = None
|
|
) -> Tuple[List, int]:
|
|
"""获取供应商列表"""
|
|
return supplier.get_multi(db, skip, limit, status, keyword)
|
|
|
|
def create_supplier(
|
|
self,
|
|
db: Session,
|
|
obj_in: SupplierCreate,
|
|
creator_id: Optional[int] = None
|
|
):
|
|
"""创建供应商"""
|
|
try:
|
|
return supplier.create(db, obj_in, creator_id)
|
|
except ValueError as e:
|
|
raise AlreadyExistsException("供应商") from e
|
|
|
|
def update_supplier(
|
|
self,
|
|
db: Session,
|
|
supplier_id: int,
|
|
obj_in: SupplierUpdate,
|
|
updater_id: Optional[int] = None
|
|
):
|
|
"""更新供应商"""
|
|
db_obj = supplier.get(db, supplier_id)
|
|
if not db_obj:
|
|
raise NotFoundException("供应商")
|
|
return supplier.update(db, db_obj, obj_in, updater_id)
|
|
|
|
def delete_supplier(
|
|
self,
|
|
db: Session,
|
|
supplier_id: int,
|
|
deleter_id: Optional[int] = None
|
|
) -> bool:
|
|
"""删除供应商"""
|
|
if not supplier.get(db, supplier_id):
|
|
raise NotFoundException("供应商")
|
|
return supplier.delete(db, supplier_id, deleter_id)
|
|
|
|
|
|
# 创建全局实例
|
|
brand_service = BrandService()
|
|
supplier_service = SupplierService()
|