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
110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
"""Brands CRUD. (name, country) is UNIQUE (uq_brands_name_country)."""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, 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 Brand, BrandMaterial
|
|
from app.schemas.catalog import BrandCreate, BrandUpdate
|
|
from app.utils.errors import ConflictError, NotFoundError
|
|
from app.utils.pagination import paginate, sort_query
|
|
from app.utils.serialization import to_iso
|
|
|
|
router = APIRouter(prefix="/brands", tags=["brands"])
|
|
_procurement_writer = require_procurement_writer()
|
|
|
|
|
|
def brand_to_dict(brand: Brand) -> dict:
|
|
return {
|
|
"id": str(brand.id),
|
|
"name": brand.name,
|
|
"country": brand.country,
|
|
"created_at": to_iso(brand.created_at),
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
async def list_brands(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
q: str | None = None,
|
|
country: str | None = None,
|
|
sort: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
stmt = select(Brand)
|
|
if q:
|
|
stmt = stmt.where(Brand.name.ilike(f"%{q}%"))
|
|
if country:
|
|
stmt = stmt.where(Brand.country == country)
|
|
stmt = sort_query(stmt, Brand, sort)
|
|
items, meta = await paginate(stmt, page, page_size, db)
|
|
return {"data": [brand_to_dict(b) for b in items], "meta": meta}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_brand(
|
|
payload: BrandCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_auth=Depends(_procurement_writer),
|
|
):
|
|
rows = (await db.execute(select(Brand).where(Brand.name == payload.name))).scalars().all()
|
|
if any(b.country == payload.country for b in rows):
|
|
raise ConflictError("برندی با این نام و کشور قبلاً ثبت شده است")
|
|
brand = Brand(name=payload.name, country=payload.country)
|
|
db.add(brand)
|
|
await db.flush()
|
|
return {"data": brand_to_dict(brand)}
|
|
|
|
|
|
@router.get("/{brand_id}")
|
|
async def get_brand(brand_id, db: AsyncSession = Depends(get_db)):
|
|
brand = await db.get(Brand, brand_id)
|
|
if brand is None:
|
|
raise NotFoundError("برند یافت نشد")
|
|
return {"data": brand_to_dict(brand)}
|
|
|
|
|
|
@router.patch("/{brand_id}")
|
|
async def update_brand(
|
|
brand_id,
|
|
payload: BrandUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_auth=Depends(_procurement_writer),
|
|
):
|
|
brand = await db.get(Brand, brand_id)
|
|
if brand is None:
|
|
raise NotFoundError("برند یافت نشد")
|
|
new_name = payload.name if payload.name is not None else brand.name
|
|
new_country = payload.country if payload.country is not None else brand.country
|
|
rows = (await db.execute(select(Brand).where(Brand.name == new_name, Brand.id != brand_id))).scalars().all()
|
|
if any(b.country == new_country for b in rows):
|
|
raise ConflictError("برندی با این نام و کشور قبلاً ثبت شده است")
|
|
if payload.name is not None:
|
|
brand.name = payload.name
|
|
if payload.country is not None:
|
|
brand.country = payload.country
|
|
await db.flush()
|
|
return {"data": brand_to_dict(brand)}
|
|
|
|
|
|
@router.delete("/{brand_id}")
|
|
async def delete_brand(
|
|
brand_id,
|
|
db: AsyncSession = Depends(get_db),
|
|
_auth=Depends(_procurement_writer),
|
|
):
|
|
brand = await db.get(Brand, brand_id)
|
|
if brand is None:
|
|
raise NotFoundError("برند یافت نشد")
|
|
# Recommendation (b) from P2-4: block instead of silently cascading associations.
|
|
linked = await db.scalar(
|
|
select(func.count()).select_from(BrandMaterial).where(BrandMaterial.brand_id == brand_id)
|
|
)
|
|
if linked:
|
|
raise ConflictError(f"این برند با {linked} متریال مرتبط است و قابل حذف نیست")
|
|
await db.delete(brand)
|
|
await db.flush()
|
|
return {"data": {"id": str(brand_id), "deleted": True}}
|