136 lines
5.4 KiB
Python
136 lines
5.4 KiB
Python
"""Brand-material M:N junction CRUD. (brand_id, material_id) is UNIQUE."""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.engine import get_db
|
|
from app.models.catalog import Brand, BrandMaterial, Material
|
|
from app.models.procurement import LineItem
|
|
from app.schemas.catalog import BrandMaterialCreate, BrandMaterialUpdate
|
|
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="/brand-materials", tags=["brand-materials"])
|
|
|
|
|
|
def bm_to_dict(bm: BrandMaterial, expand: set[str] = frozenset()) -> dict:
|
|
d = {
|
|
"id": str(bm.id),
|
|
"brand_id": str(bm.brand_id),
|
|
"material_id": str(bm.material_id),
|
|
"catalog_code": bm.catalog_code,
|
|
"is_active": bm.is_active,
|
|
"created_at": to_iso(bm.created_at),
|
|
}
|
|
if "brand" in expand and hasattr(bm, "_brand"):
|
|
d["brand"] = {"id": str(bm._brand.id), "name": bm._brand.name, "country": bm._brand.country}
|
|
if "material" in expand and hasattr(bm, "_material"):
|
|
d["material"] = {"id": str(bm._material.id), "name": bm._material.name}
|
|
return d
|
|
|
|
|
|
@router.get("")
|
|
async def list_brand_materials(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
brand_id=None,
|
|
material_id=None,
|
|
is_active: bool | None = None,
|
|
sort: str | None = None,
|
|
expand: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
stmt = select(BrandMaterial)
|
|
if brand_id:
|
|
stmt = stmt.where(BrandMaterial.brand_id == brand_id)
|
|
if material_id:
|
|
stmt = stmt.where(BrandMaterial.material_id == material_id)
|
|
if is_active is not None:
|
|
stmt = stmt.where(BrandMaterial.is_active == is_active)
|
|
stmt = sort_query(stmt, BrandMaterial, sort)
|
|
items, meta = await paginate(stmt, page, page_size, db)
|
|
|
|
tokens = {t.strip() for t in expand.split(",")} if expand else set()
|
|
if "brand" in tokens or "material" in tokens:
|
|
brand_ids = {bm.brand_id for bm in items}
|
|
material_ids = {bm.material_id for bm in items}
|
|
brands = {}
|
|
materials = {}
|
|
if brand_ids:
|
|
brands = {b.id: b for b in (await db.execute(select(Brand).where(Brand.id.in_(brand_ids)))).scalars()}
|
|
if material_ids:
|
|
materials = {
|
|
m.id: m for m in (await db.execute(select(Material).where(Material.id.in_(material_ids)))).scalars()
|
|
}
|
|
for bm in items:
|
|
bm._brand = brands.get(bm.brand_id)
|
|
bm._material = materials.get(bm.material_id)
|
|
return {"data": [bm_to_dict(bm, tokens) for bm in items], "meta": meta}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_brand_material(payload: BrandMaterialCreate, db: AsyncSession = Depends(get_db)):
|
|
if await db.get(Brand, payload.brand_id) is None:
|
|
raise NotFoundError("برند یافت نشد", field="brand_id")
|
|
if await db.get(Material, payload.material_id) is None:
|
|
raise NotFoundError("متریال یافت نشد", field="material_id")
|
|
duplicate = await db.scalar(
|
|
select(BrandMaterial.id).where(
|
|
BrandMaterial.brand_id == payload.brand_id, BrandMaterial.material_id == payload.material_id
|
|
)
|
|
)
|
|
if duplicate is not None:
|
|
raise ConflictError("این ارتباط برند-متریال قبلاً ثبت شده است")
|
|
bm = BrandMaterial(
|
|
brand_id=payload.brand_id,
|
|
material_id=payload.material_id,
|
|
catalog_code=payload.catalog_code,
|
|
is_active=payload.is_active,
|
|
)
|
|
db.add(bm)
|
|
await db.flush()
|
|
return {"data": bm_to_dict(bm)}
|
|
|
|
|
|
@router.get("/{bm_id}")
|
|
async def get_brand_material(bm_id, expand: str | None = None, db: AsyncSession = Depends(get_db)):
|
|
bm = await db.get(BrandMaterial, bm_id)
|
|
if bm is None:
|
|
raise NotFoundError("ارتباط برند-متریال یافت نشد")
|
|
tokens = {t.strip() for t in expand.split(",")} if expand else set()
|
|
if "brand" in tokens:
|
|
bm._brand = await db.get(Brand, bm.brand_id)
|
|
if "material" in tokens:
|
|
bm._material = await db.get(Material, bm.material_id)
|
|
return {"data": bm_to_dict(bm, tokens)}
|
|
|
|
|
|
@router.patch("/{bm_id}")
|
|
async def update_brand_material(bm_id, payload: BrandMaterialUpdate, db: AsyncSession = Depends(get_db)):
|
|
bm = await db.get(BrandMaterial, bm_id)
|
|
if bm is None:
|
|
raise NotFoundError("ارتباط برند-متریال یافت نشد")
|
|
if payload.catalog_code is not None:
|
|
bm.catalog_code = payload.catalog_code
|
|
if payload.is_active is not None:
|
|
bm.is_active = payload.is_active
|
|
await db.flush()
|
|
return {"data": bm_to_dict(bm)}
|
|
|
|
|
|
@router.delete("/{bm_id}")
|
|
async def delete_brand_material(bm_id, db: AsyncSession = Depends(get_db)):
|
|
bm = await db.get(BrandMaterial, bm_id)
|
|
if bm is None:
|
|
raise NotFoundError("ارتباط برند-متریال یافت نشد")
|
|
# Recommendation (P2-16): block if line items reference it (SET NULL would lose context).
|
|
referenced = await db.scalar(
|
|
select(func.count()).select_from(LineItem).where(LineItem.brand_material_id == bm_id)
|
|
)
|
|
if referenced:
|
|
raise ConflictError("این ارتباط توسط خطوط آیتم استفاده شده و قابل حذف نیست")
|
|
await db.delete(bm)
|
|
await db.flush()
|
|
return {"data": {"id": str(bm_id), "deleted": True}}
|