121 lines
4.6 KiB
Python
121 lines
4.6 KiB
Python
"""Units CRUD. units.id is the ONLY integer PK in the system (IDENTITY)."""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.engine import get_db
|
|
from app.models.catalog import AttributeDefinition, Material, MaterialUnit, Unit
|
|
from app.models.procurement import LineItem, LineItemSpec
|
|
from app.schemas.catalog import UnitCreate, UnitUpdate
|
|
from app.utils.errors import ConflictError, NotFoundError
|
|
from app.utils.pagination import paginate, sort_query
|
|
|
|
router = APIRouter(prefix="/units", tags=["units"])
|
|
|
|
|
|
def unit_full_to_dict(unit: Unit) -> dict:
|
|
"""Full unit shape (P0-1): includes dimension, si_factor, is_base."""
|
|
return {
|
|
"id": unit.id,
|
|
"code": unit.code,
|
|
"name_fa": unit.name_fa,
|
|
"dimension": unit.dimension,
|
|
"si_factor": float(unit.si_factor),
|
|
"is_base": unit.is_base,
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
async def list_units(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
q: str | None = None,
|
|
dimension: str | None = None,
|
|
sort: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
stmt = select(Unit)
|
|
if q:
|
|
stmt = stmt.where(or_(Unit.code.ilike(f"%{q}%"), Unit.name_fa.ilike(f"%{q}%")))
|
|
if dimension:
|
|
stmt = stmt.where(Unit.dimension == dimension)
|
|
stmt = sort_query(stmt, Unit, sort, default="id")
|
|
items, meta = await paginate(stmt, page, page_size, db)
|
|
return {"data": [unit_full_to_dict(u) for u in items], "meta": meta}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_unit(payload: UnitCreate, db: AsyncSession = Depends(get_db)):
|
|
duplicate = await db.scalar(select(Unit.id).where(Unit.code == payload.code))
|
|
if duplicate is not None:
|
|
raise ConflictError("واحدی با این کد قبلاً ثبت شده است", field="code")
|
|
unit = Unit(
|
|
code=payload.code,
|
|
name_fa=payload.name_fa,
|
|
dimension=payload.dimension,
|
|
si_factor=payload.si_factor,
|
|
is_base=payload.is_base,
|
|
)
|
|
db.add(unit)
|
|
await db.flush()
|
|
return {"data": unit_full_to_dict(unit)}
|
|
|
|
|
|
@router.get("/{unit_id}")
|
|
async def get_unit(unit_id: int, db: AsyncSession = Depends(get_db)):
|
|
unit = await db.get(Unit, unit_id)
|
|
if unit is None:
|
|
raise NotFoundError("واحد یافت نشد")
|
|
return {"data": unit_full_to_dict(unit)}
|
|
|
|
|
|
@router.patch("/{unit_id}")
|
|
async def update_unit(unit_id: int, payload: UnitUpdate, db: AsyncSession = Depends(get_db)):
|
|
unit = await db.get(Unit, unit_id)
|
|
if unit is None:
|
|
raise NotFoundError("واحد یافت نشد")
|
|
if payload.code is not None and payload.code != unit.code:
|
|
duplicate = await db.scalar(select(Unit.id).where(Unit.code == payload.code, Unit.id != unit_id))
|
|
if duplicate is not None:
|
|
raise ConflictError("واحدی با این کد قبلاً ثبت شده است", field="code")
|
|
unit.code = payload.code
|
|
if payload.name_fa is not None:
|
|
unit.name_fa = payload.name_fa
|
|
if payload.dimension is not None:
|
|
unit.dimension = payload.dimension
|
|
if payload.si_factor is not None:
|
|
unit.si_factor = payload.si_factor
|
|
if payload.is_base is not None:
|
|
unit.is_base = payload.is_base
|
|
await db.flush()
|
|
return {"data": unit_full_to_dict(unit)}
|
|
|
|
|
|
@router.delete("/{unit_id}")
|
|
async def delete_unit(unit_id: int, db: AsyncSession = Depends(get_db)):
|
|
unit = await db.get(Unit, unit_id)
|
|
if unit is None:
|
|
raise NotFoundError("واحد یافت نشد")
|
|
|
|
refs: dict[str, int] = {
|
|
"attribute_definitions": await db.scalar(
|
|
select(func.count()).select_from(AttributeDefinition).where(AttributeDefinition.unit_id == unit_id)
|
|
),
|
|
"materials": await db.scalar(
|
|
select(func.count()).select_from(Material).where(Material.canonical_unit_id == unit_id)
|
|
),
|
|
"material_units": await db.scalar(
|
|
select(func.count()).select_from(MaterialUnit).where(MaterialUnit.unit_id == unit_id)
|
|
),
|
|
"line_items": await db.scalar(select(func.count()).select_from(LineItem).where(LineItem.unit_id == unit_id)),
|
|
"line_item_specs": await db.scalar(
|
|
select(func.count()).select_from(LineItemSpec).where(LineItemSpec.unit_id_at_time == unit_id)
|
|
),
|
|
}
|
|
total = sum(refs.values())
|
|
if total:
|
|
raise ConflictError(f"این واحد در {total} رکورد دیگر استفاده شده و قابل حذف نیست")
|
|
await db.delete(unit)
|
|
await db.flush()
|
|
return {"data": {"id": unit_id, "deleted": True}}
|