MMD
0ea0322967
Reverting the production revert so the pm branch can merge cleanly on top of the restored RBAC base. Conflicts against newer integration work resolved by combining: - App.tsx: v7's bot-draft wizard bypass + login returnUrl kept alongside the supervisor gating (no-project block screen, panel scoping) - LoginPage: returnUrl navigation kept, supervisor routing added - migration.sql: bot_drafts table kept, unified-projects ALTERs restored
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_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_role("admin", "procurement_manager")
|
|
|
|
|
|
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}}
|