Milad
4a3c226518
- new tables user_panels (panel: procurement/inventory/pm + level full/limited)
and user_project_members (M:N membership, separate from pm responsibility)
- UserContext carries panels + member_project_ids; /users/me and the users
list include them
- admin endpoints: PUT /users/{id}/panels, PUT /users/{id}/panel-projects
- inventory scope: panel inventory 'full' -> manager scope (all projects, NO
stock change); 'limited' -> supervisor scope (own memberships, stock change
allowed); legacy roles kept as migration fallback; admin untouched
- procurement writes: _procurement_writer is now panel-aware (admin, legacy
procurement_manager, or procurement 'full') across all 18 write routers;
quote CREATE additionally allows 'limited' users restricted to their member
projects (validated server-side)
- frontend: panel-aware HomeRoute (inventory -> warehouse panel, procurement
limited -> minimal RequesterHome), wizard project options restricted for
limited users, change-inventory button/route gated to inventory 'limited',
organization panel gains the panels & access-levels management section
119 lines
4.1 KiB
Python
119 lines
4.1 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.auth import require_procurement_writer, require_role
|
|
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"])
|
|
_procurement_writer = require_procurement_writer()
|
|
|
|
|
|
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),
|
|
_auth=Depends(_procurement_writer),
|
|
):
|
|
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),
|
|
_auth=Depends(_procurement_writer),
|
|
):
|
|
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),
|
|
_auth=Depends(_procurement_writer),
|
|
):
|
|
sc = await db.get(StandardConversion, sc_id)
|
|
if sc is None:
|
|
raise NotFoundError("تبدیل استاندارد یافت نشد")
|
|
await db.delete(sc)
|
|
await db.flush()
|
|
return {"data": {"deleted": True}}
|