164 lines
5.6 KiB
Python
164 lines
5.6 KiB
Python
"""
|
|
资产相关的Pydantic Schema
|
|
"""
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime, date
|
|
from decimal import Decimal
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
# ===== 资产Schema =====
|
|
|
|
class AssetBase(BaseModel):
|
|
"""资产基础Schema"""
|
|
asset_name: str = Field(..., min_length=1, max_length=200, description="资产名称")
|
|
device_type_id: int = Field(..., gt=0, description="设备类型ID")
|
|
brand_id: Optional[int] = Field(None, gt=0, description="品牌ID")
|
|
model: Optional[str] = Field(None, max_length=200, description="规格型号")
|
|
serial_number: Optional[str] = Field(None, max_length=200, description="序列号")
|
|
supplier_id: Optional[int] = Field(None, gt=0, description="供应商ID")
|
|
purchase_date: Optional[date] = Field(None, description="采购日期")
|
|
purchase_price: Optional[Decimal] = Field(None, ge=0, description="采购价格")
|
|
warranty_period: Optional[int] = Field(None, ge=0, description="保修期(月)")
|
|
organization_id: int = Field(..., gt=0, description="所属网点ID")
|
|
location: Optional[str] = Field(None, max_length=500, description="存放位置")
|
|
remark: Optional[str] = Field(None, description="备注")
|
|
|
|
|
|
class AssetCreate(AssetBase):
|
|
"""创建资产Schema"""
|
|
dynamic_attributes: Dict[str, Any] = Field(default_factory=dict, description="动态字段值")
|
|
|
|
|
|
class AssetUpdate(BaseModel):
|
|
"""更新资产Schema"""
|
|
asset_name: Optional[str] = Field(None, min_length=1, max_length=200)
|
|
brand_id: Optional[int] = Field(None, gt=0)
|
|
model: Optional[str] = Field(None, max_length=200)
|
|
serial_number: Optional[str] = Field(None, max_length=200)
|
|
supplier_id: Optional[int] = Field(None, gt=0)
|
|
purchase_date: Optional[date] = None
|
|
purchase_price: Optional[Decimal] = Field(None, ge=0)
|
|
warranty_period: Optional[int] = Field(None, ge=0)
|
|
warranty_expire_date: Optional[date] = None
|
|
organization_id: Optional[int] = Field(None, gt=0)
|
|
location: Optional[str] = Field(None, max_length=500)
|
|
dynamic_attributes: Optional[Dict[str, Any]] = None
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class AssetInDB(BaseModel):
|
|
"""数据库中的资产Schema"""
|
|
id: int
|
|
asset_code: str
|
|
asset_name: str
|
|
device_type_id: int
|
|
brand_id: Optional[int]
|
|
model: Optional[str]
|
|
serial_number: Optional[str]
|
|
supplier_id: Optional[int]
|
|
purchase_date: Optional[date]
|
|
purchase_price: Optional[Decimal]
|
|
warranty_period: Optional[int]
|
|
warranty_expire_date: Optional[date]
|
|
organization_id: int
|
|
location: Optional[str]
|
|
status: str
|
|
dynamic_attributes: Dict[str, Any]
|
|
qr_code_url: Optional[str]
|
|
remark: Optional[str]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class AssetResponse(AssetInDB):
|
|
"""资产响应Schema"""
|
|
pass
|
|
|
|
|
|
class AssetWithRelations(AssetResponse):
|
|
"""带关联信息的资产响应Schema"""
|
|
device_type: Optional[Dict[str, Any]] = None
|
|
brand: Optional[Dict[str, Any]] = None
|
|
supplier: Optional[Dict[str, Any]] = None
|
|
organization: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
# ===== 资产状态历史Schema =====
|
|
|
|
class AssetStatusHistoryBase(BaseModel):
|
|
"""资产状态历史基础Schema"""
|
|
old_status: Optional[str] = Field(None, description="原状态")
|
|
new_status: str = Field(..., description="新状态")
|
|
operation_type: str = Field(..., description="操作类型")
|
|
remark: Optional[str] = Field(None, description="备注")
|
|
|
|
|
|
class AssetStatusHistoryInDB(BaseModel):
|
|
"""数据库中的资产状态历史Schema"""
|
|
id: int
|
|
asset_id: int
|
|
old_status: Optional[str]
|
|
new_status: str
|
|
operation_type: str
|
|
operator_id: int
|
|
operator_name: Optional[str]
|
|
organization_id: Optional[int]
|
|
remark: Optional[str]
|
|
extra_data: Optional[Dict[str, Any]]
|
|
created_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class AssetStatusHistoryResponse(AssetStatusHistoryInDB):
|
|
"""资产状态历史响应Schema"""
|
|
pass
|
|
|
|
|
|
# ===== 批量操作Schema =====
|
|
|
|
class AssetBatchImport(BaseModel):
|
|
"""批量导入Schema"""
|
|
file_path: str = Field(..., description="Excel文件路径")
|
|
|
|
|
|
class AssetBatchImportResult(BaseModel):
|
|
"""批量导入结果Schema"""
|
|
total: int = Field(..., description="总数")
|
|
success: int = Field(..., description="成功数")
|
|
failed: int = Field(..., description="失败数")
|
|
errors: List[Dict[str, Any]] = Field(default_factory=list, description="错误列表")
|
|
|
|
|
|
class AssetBatchDelete(BaseModel):
|
|
"""批量删除Schema"""
|
|
asset_ids: List[int] = Field(..., min_items=1, description="资产ID列表")
|
|
|
|
|
|
# ===== 查询参数Schema =====
|
|
|
|
class AssetQueryParams(BaseModel):
|
|
"""资产查询参数"""
|
|
keyword: Optional[str] = Field(None, description="搜索关键词")
|
|
device_type_id: Optional[int] = Field(None, gt=0, description="设备类型ID")
|
|
organization_id: Optional[int] = Field(None, gt=0, description="网点ID")
|
|
status: Optional[str] = Field(None, description="状态")
|
|
purchase_date_start: Optional[date] = Field(None, description="采购日期开始")
|
|
purchase_date_end: Optional[date] = Field(None, description="采购日期结束")
|
|
page: int = Field(default=1, ge=1, description="页码")
|
|
page_size: int = Field(default=20, ge=1, le=100, description="每页数量")
|
|
|
|
|
|
# ===== 状态转换Schema =====
|
|
|
|
class AssetStatusTransition(BaseModel):
|
|
"""资产状态转换Schema"""
|
|
new_status: str = Field(..., description="目标状态")
|
|
remark: Optional[str] = Field(None, description="备注")
|
|
extra_data: Optional[Dict[str, Any]] = Field(None, description="额外数据")
|