147 lines
5.7 KiB
Python
147 lines
5.7 KiB
Python
"""Unit conversions — category/brand-scoped (units.html dashboard, P2-28..31).
|
||
|
||
Model: {category_id, brand?, from_unit, to_unit, factor}
|
||
1 from_unit = factor × to_unit
|
||
brand empty = all brands of the category; brand set = that brand only.
|
||
"""
|
||
from fastapi import APIRouter, Depends, Query
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.db.engine import get_db
|
||
from app.models.catalog import MaterialCategory, UnitConversion
|
||
from app.schemas.catalog import UnitConversionCreate, UnitConversionUpdate
|
||
from app.services.expand_service import category_brief
|
||
from app.utils.errors import ConflictError, NotFoundError
|
||
from app.utils.pagination import paginate
|
||
from app.utils.serialization import to_iso
|
||
|
||
router = APIRouter(prefix="/unit-conversions", tags=["unit-conversions"])
|
||
|
||
|
||
def uc_to_dict(uc: UnitConversion, expand: set[str] = frozenset()) -> dict:
|
||
d = {
|
||
"id": str(uc.id),
|
||
"category_id": str(uc.category_id),
|
||
"brand": uc.brand,
|
||
"from_unit": uc.from_unit,
|
||
"to_unit": uc.to_unit,
|
||
"factor": float(uc.factor),
|
||
"updated_at": to_iso(uc.updated_at),
|
||
}
|
||
if "category" in expand and hasattr(uc, "_category"):
|
||
d["category"] = category_brief(uc._category)
|
||
return d
|
||
|
||
|
||
def _duplicate_condition(uc_or_none, category_id, brand, from_unit, to_unit):
|
||
"""Uniqueness treats NULL brand as a concrete value (matches the partial unique index)."""
|
||
where = (
|
||
UnitConversion.category_id == category_id,
|
||
UnitConversion.from_unit == from_unit,
|
||
UnitConversion.to_unit == to_unit,
|
||
)
|
||
# PG «IS» فقط با NULL معتبر است؛ برای برند مقداردار از «==» استفاده میشود
|
||
where = where + (UnitConversion.brand == brand if brand is not None else UnitConversion.brand.is_(None),)
|
||
if uc_or_none is not None:
|
||
where = where + (UnitConversion.id != uc_or_none.id,)
|
||
return where
|
||
|
||
|
||
async def _check_duplicate(db: AsyncSession, exclude_id, category_id, brand, from_unit, to_unit) -> None:
|
||
duplicate = await db.scalar(
|
||
select(UnitConversion.id).where(*_duplicate_condition(exclude_id, category_id, brand, from_unit, to_unit))
|
||
)
|
||
if duplicate is not None:
|
||
raise ConflictError("این تبدیل قبلاً ثبت شده است")
|
||
|
||
|
||
@router.get("")
|
||
async def list_unit_conversions(
|
||
page: int = Query(1, ge=1),
|
||
page_size: int = Query(20, ge=1, le=100),
|
||
category_id=None,
|
||
brand: str | None = None,
|
||
from_unit: str | None = None,
|
||
to_unit: str | None = None,
|
||
expand: str | None = None,
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
stmt = select(UnitConversion)
|
||
if category_id:
|
||
stmt = stmt.where(UnitConversion.category_id == category_id)
|
||
if brand:
|
||
stmt = stmt.where(UnitConversion.brand == brand)
|
||
if from_unit:
|
||
stmt = stmt.where(UnitConversion.from_unit == from_unit)
|
||
if to_unit:
|
||
stmt = stmt.where(UnitConversion.to_unit == to_unit)
|
||
items, meta = await paginate(stmt, page, page_size, db)
|
||
tokens = {t.strip() for t in expand.split(",")} if expand else set()
|
||
if "category" in tokens:
|
||
cat_ids = {uc.category_id for uc in items}
|
||
cats = {}
|
||
if cat_ids:
|
||
cats = {c.id: c for c in (await db.execute(select(MaterialCategory).where(MaterialCategory.id.in_(cat_ids)))).scalars()}
|
||
for uc in items:
|
||
uc._category = cats.get(uc.category_id)
|
||
return {"data": [uc_to_dict(uc, tokens) for uc in items], "meta": meta}
|
||
|
||
|
||
@router.post("", status_code=201)
|
||
async def create_unit_conversion(payload: UnitConversionCreate, db: AsyncSession = Depends(get_db)):
|
||
category = await db.get(MaterialCategory, payload.category_id)
|
||
if category is None:
|
||
raise NotFoundError("دستهبندی یافت نشد", field="category_id")
|
||
await _check_duplicate(db, None, payload.category_id, payload.brand, payload.from_unit, payload.to_unit)
|
||
uc = UnitConversion(
|
||
category_id=payload.category_id,
|
||
brand=payload.brand,
|
||
from_unit=payload.from_unit,
|
||
to_unit=payload.to_unit,
|
||
factor=payload.factor,
|
||
)
|
||
db.add(uc)
|
||
await db.flush()
|
||
return {"data": uc_to_dict(uc)}
|
||
|
||
|
||
@router.get("/{uc_id}")
|
||
async def get_unit_conversion(uc_id, db: AsyncSession = Depends(get_db)):
|
||
uc = await db.get(UnitConversion, uc_id)
|
||
if uc is None:
|
||
raise NotFoundError("تبدیل واحد یافت نشد")
|
||
return {"data": uc_to_dict(uc)}
|
||
|
||
|
||
@router.patch("/{uc_id}")
|
||
async def update_unit_conversion(uc_id, payload: UnitConversionUpdate, db: AsyncSession = Depends(get_db)):
|
||
uc = await db.get(UnitConversion, uc_id)
|
||
if uc is None:
|
||
raise NotFoundError("تبدیل واحد یافت نشد")
|
||
new_brand = payload.brand if payload.brand is not None else uc.brand
|
||
new_from = payload.from_unit if payload.from_unit is not None else uc.from_unit
|
||
new_to = payload.to_unit if payload.to_unit is not None else uc.to_unit
|
||
if payload.brand is not None or payload.from_unit is not None or payload.to_unit is not None:
|
||
await _check_duplicate(db, uc.id, uc.category_id, new_brand, new_from, new_to)
|
||
if payload.brand is not None:
|
||
uc.brand = payload.brand
|
||
if payload.from_unit is not None:
|
||
uc.from_unit = payload.from_unit
|
||
if payload.to_unit is not None:
|
||
uc.to_unit = payload.to_unit
|
||
if payload.factor is not None:
|
||
uc.factor = payload.factor
|
||
await db.flush()
|
||
return {"data": uc_to_dict(uc)}
|
||
|
||
|
||
@router.delete("/{uc_id}")
|
||
async def delete_unit_conversion(uc_id, db: AsyncSession = Depends(get_db)):
|
||
uc = await db.get(UnitConversion, uc_id)
|
||
if uc is None:
|
||
raise NotFoundError("تبدیل واحد یافت نشد")
|
||
await db.delete(uc)
|
||
await db.flush()
|
||
return {"data": {"deleted": True}}
|