64 lines
2.4 KiB
Python
64 lines
2.4 KiB
Python
"""Supplier ↔ material link service (v2.5) — the direct edge of the new flow.
|
|
|
|
Linking a material to a supplier derives materials.category_id from the
|
|
supplier's category when the material has none (first-link-wins). The same
|
|
rule is enforced by the DB trigger trg_supplier_materials_derive_category;
|
|
this service applies it in the application layer so the ORM state is
|
|
consistent inside the same transaction.
|
|
"""
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.catalog import Material
|
|
from app.models.procurement import Supplier, SupplierMaterial
|
|
from app.utils.errors import NotFoundError, ValidationError
|
|
|
|
|
|
async def link_supplier_material(
|
|
db: AsyncSession, supplier_id, material_id, is_active: bool = True
|
|
) -> SupplierMaterial:
|
|
"""Upsert a supplier<->material link (idempotent) + derive category (first-link-wins).
|
|
|
|
Caller is responsible for the surrounding transaction.
|
|
"""
|
|
supplier = await db.get(Supplier, supplier_id)
|
|
if supplier is None:
|
|
raise NotFoundError("تأمینکننده یافت نشد", field="supplier_id")
|
|
material = await db.get(Material, material_id)
|
|
if material is None:
|
|
raise NotFoundError("متریال یافت نشد", field="material_id")
|
|
|
|
link = await db.scalar(
|
|
select(SupplierMaterial).where(
|
|
SupplierMaterial.supplier_id == supplier_id,
|
|
SupplierMaterial.material_id == material_id,
|
|
)
|
|
)
|
|
if link is not None:
|
|
if not link.is_active and is_active:
|
|
link.is_active = True
|
|
return link
|
|
|
|
link = SupplierMaterial(supplier_id=supplier_id, material_id=material_id, is_active=is_active)
|
|
db.add(link)
|
|
await db.flush()
|
|
|
|
# First-link-wins: only when the material has no category yet.
|
|
if material.category_id is None:
|
|
material.category_id = supplier.category_id
|
|
return link
|
|
|
|
|
|
async def unlink_supplier_material(db: AsyncSession, supplier_id, material_id) -> None:
|
|
"""Remove the link. The material's category is NEVER auto-cleared (documented rule)."""
|
|
link = await db.scalar(
|
|
select(SupplierMaterial).where(
|
|
SupplierMaterial.supplier_id == supplier_id,
|
|
SupplierMaterial.material_id == material_id,
|
|
)
|
|
)
|
|
if link is None:
|
|
raise ValidationError("این متریال به این تأمینکننده متصل نیست", field="material_id")
|
|
await db.delete(link)
|
|
await db.flush()
|