Cheng Zhou
72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.material_equipment import MaterialEquipment
|
|
from app.schemas.material_equipment import MaterialEquipmentCreate, MaterialEquipmentUpdate
|
|
|
|
|
|
async def create_equipment(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
equipment_in: MaterialEquipmentCreate,
|
|
created_by: uuid.UUID | None,
|
|
) -> MaterialEquipment:
|
|
item = MaterialEquipment(
|
|
study_id=study_id,
|
|
name=equipment_in.name,
|
|
spec_model=equipment_in.spec_model,
|
|
unit=equipment_in.unit,
|
|
brand=equipment_in.brand,
|
|
origin=equipment_in.origin,
|
|
production_permit_file_name=equipment_in.production_permit_file_name,
|
|
tech_index_file_name=equipment_in.tech_index_file_name,
|
|
need_calibration=equipment_in.need_calibration,
|
|
calibration_cycle_days=equipment_in.calibration_cycle_days,
|
|
created_by=created_by,
|
|
)
|
|
db.add(item)
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
return item
|
|
|
|
|
|
async def get_equipment(db: AsyncSession, equipment_id: uuid.UUID) -> MaterialEquipment | None:
|
|
result = await db.execute(select(MaterialEquipment).where(MaterialEquipment.id == equipment_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_equipments(
|
|
db: AsyncSession,
|
|
study_id: uuid.UUID,
|
|
name: str | None = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Sequence[MaterialEquipment]:
|
|
stmt = select(MaterialEquipment).where(MaterialEquipment.study_id == study_id)
|
|
if name:
|
|
stmt = stmt.where(MaterialEquipment.name.ilike(f"%{name}%"))
|
|
stmt = stmt.order_by(MaterialEquipment.created_at.desc()).offset(skip).limit(limit)
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_equipment(
|
|
db: AsyncSession,
|
|
equipment: MaterialEquipment,
|
|
equipment_in: MaterialEquipmentUpdate,
|
|
) -> MaterialEquipment:
|
|
update_data = equipment_in.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(equipment, key, value)
|
|
await db.commit()
|
|
await db.refresh(equipment)
|
|
return equipment
|
|
|
|
|
|
async def delete_equipment(db: AsyncSession, equipment: MaterialEquipment) -> None:
|
|
await db.delete(equipment)
|
|
await db.commit()
|