104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
"""Standard conversions — global, applies to all categories (P2-32..35)."""
|
|
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 StandardConversion
|
|
from app.schemas.catalog import StandardConversionCreate, StandardConversionUpdate
|
|
from app.utils.errors import ConflictError, NotFoundError
|
|
from app.utils.pagination import paginate
|
|
from app.utils.serialization import to_iso
|
|
|
|
router = APIRouter(prefix="/standard-conversions", tags=["standard-conversions"])
|
|
|
|
|
|
def sc_to_dict(sc: StandardConversion) -> dict:
|
|
return {
|
|
"id": str(sc.id),
|
|
"from_unit": sc.from_unit,
|
|
"to_unit": sc.to_unit,
|
|
"factor": float(sc.factor),
|
|
"updated_at": to_iso(sc.updated_at),
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
async def list_standard_conversions(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
from_unit: str | None = None,
|
|
to_unit: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
stmt = select(StandardConversion)
|
|
if from_unit:
|
|
stmt = stmt.where(StandardConversion.from_unit == from_unit)
|
|
if to_unit:
|
|
stmt = stmt.where(StandardConversion.to_unit == to_unit)
|
|
items, meta = await paginate(stmt, page, page_size, db)
|
|
return {"data": [sc_to_dict(sc) for sc in items], "meta": meta}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_standard_conversion(payload: StandardConversionCreate, db: AsyncSession = Depends(get_db)):
|
|
duplicate = await db.scalar(
|
|
select(StandardConversion.id).where(
|
|
StandardConversion.from_unit == payload.from_unit,
|
|
StandardConversion.to_unit == payload.to_unit,
|
|
)
|
|
)
|
|
if duplicate is not None:
|
|
raise ConflictError("این تبدیل استاندارد قبلاً ثبت شده است")
|
|
sc = StandardConversion(
|
|
from_unit=payload.from_unit, to_unit=payload.to_unit, factor=payload.factor
|
|
)
|
|
db.add(sc)
|
|
await db.flush()
|
|
return {"data": sc_to_dict(sc)}
|
|
|
|
|
|
@router.get("/{sc_id}")
|
|
async def get_standard_conversion(sc_id, db: AsyncSession = Depends(get_db)):
|
|
sc = await db.get(StandardConversion, sc_id)
|
|
if sc is None:
|
|
raise NotFoundError("تبدیل استاندارد یافت نشد")
|
|
return {"data": sc_to_dict(sc)}
|
|
|
|
|
|
@router.patch("/{sc_id}")
|
|
async def update_standard_conversion(sc_id, payload: StandardConversionUpdate, db: AsyncSession = Depends(get_db)):
|
|
sc = await db.get(StandardConversion, sc_id)
|
|
if sc is None:
|
|
raise NotFoundError("تبدیل استاندارد یافت نشد")
|
|
new_from = payload.from_unit if payload.from_unit is not None else sc.from_unit
|
|
new_to = payload.to_unit if payload.to_unit is not None else sc.to_unit
|
|
if payload.from_unit is not None or payload.to_unit is not None:
|
|
duplicate = await db.scalar(
|
|
select(StandardConversion.id).where(
|
|
StandardConversion.from_unit == new_from,
|
|
StandardConversion.to_unit == new_to,
|
|
StandardConversion.id != sc_id,
|
|
)
|
|
)
|
|
if duplicate is not None:
|
|
raise ConflictError("این تبدیل استاندارد قبلاً ثبت شده است")
|
|
if payload.from_unit is not None:
|
|
sc.from_unit = payload.from_unit
|
|
if payload.to_unit is not None:
|
|
sc.to_unit = payload.to_unit
|
|
if payload.factor is not None:
|
|
sc.factor = payload.factor
|
|
await db.flush()
|
|
return {"data": sc_to_dict(sc)}
|
|
|
|
|
|
@router.delete("/{sc_id}")
|
|
async def delete_standard_conversion(sc_id, db: AsyncSession = Depends(get_db)):
|
|
sc = await db.get(StandardConversion, sc_id)
|
|
if sc is None:
|
|
raise NotFoundError("تبدیل استاندارد یافت نشد")
|
|
await db.delete(sc)
|
|
await db.flush()
|
|
return {"data": {"deleted": True}}
|