feat(rbac): enforce procurement write restrictions and support supervisor role in SSO

This commit is contained in:
2026-09-26 12:32:01 +03:30
parent 730d0bd496
commit 5a9b543040
31 changed files with 480 additions and 139 deletions
+18 -3
View File
@@ -3,6 +3,7 @@ from fastapi import APIRouter, Depends, Query
from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.auth import require_role
from app.db.engine import get_db
from app.models.catalog import MaterialCategory
from app.models.procurement import Quote, Supplier
@@ -12,6 +13,7 @@ from app.utils.pagination import paginate, sort_query
from app.utils.serialization import to_iso
router = APIRouter(prefix="/suppliers", tags=["suppliers"])
_procurement_writer = require_role("admin", "procurement_manager")
def supplier_to_dict(supplier: Supplier, category: MaterialCategory | None = None) -> dict:
@@ -71,7 +73,11 @@ async def list_suppliers(
@router.post("", status_code=201)
async def create_supplier(payload: SupplierCreate, db: AsyncSession = Depends(get_db)):
async def create_supplier(
payload: SupplierCreate,
db: AsyncSession = Depends(get_db),
_auth=Depends(_procurement_writer),
):
category = await _category_or_404(db, payload.category_id)
supplier = Supplier(
name=payload.name,
@@ -96,7 +102,12 @@ async def get_supplier(supplier_id, db: AsyncSession = Depends(get_db)):
@router.patch("/{supplier_id}")
async def update_supplier(supplier_id, payload: SupplierUpdate, db: AsyncSession = Depends(get_db)):
async def update_supplier(
supplier_id,
payload: SupplierUpdate,
db: AsyncSession = Depends(get_db),
_auth=Depends(_procurement_writer),
):
supplier = await db.get(Supplier, supplier_id)
if supplier is None:
raise NotFoundError("تأمین‌کننده یافت نشد")
@@ -120,7 +131,11 @@ async def update_supplier(supplier_id, payload: SupplierUpdate, db: AsyncSession
@router.delete("/{supplier_id}")
async def delete_supplier(supplier_id, db: AsyncSession = Depends(get_db)):
async def delete_supplier(
supplier_id,
db: AsyncSession = Depends(get_db),
_auth=Depends(_procurement_writer),
):
supplier = await db.get(Supplier, supplier_id)
if supplier is None:
raise NotFoundError("تأمین‌کننده یافت نشد")