198 lines
8.6 KiB
Python
198 lines
8.6 KiB
Python
"""Material domain services: EAV value coercion/validation + atomic creation."""
|
|
from datetime import date
|
|
from decimal import Decimal, InvalidOperation
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.catalog import (
|
|
AttributeDefinition,
|
|
Brand,
|
|
BrandMaterial,
|
|
Material,
|
|
MaterialAttributeValue,
|
|
MaterialCategory,
|
|
Unit,
|
|
)
|
|
from app.schemas.catalog import MaterialAttributeIn, MaterialCreate, MaterialUpdate
|
|
from app.utils.errors import ConflictError, NotFoundError, ValidationError
|
|
|
|
|
|
def coerce_eav_value(data_type: str, value, field: str = "value"):
|
|
"""Coerce a client value against attribute_definitions.data_type.
|
|
|
|
Returns the Python value to store in the matching typed column.
|
|
Raises 400 VALIDATION_ERROR on mismatch (contract error shape).
|
|
"""
|
|
if data_type == "number":
|
|
try:
|
|
return Decimal(str(value))
|
|
except (ValueError, InvalidOperation, TypeError):
|
|
raise ValidationError("نوع مقدار با data_type سازگار نیست — expected number", field=field)
|
|
if data_type == "integer":
|
|
try:
|
|
if isinstance(value, str):
|
|
if "." in value or "e" in value.lower() or "E" in value:
|
|
raise ValueError
|
|
return int(value)
|
|
if isinstance(value, bool): # bool is an int subclass — reject it explicitly
|
|
raise ValueError
|
|
if isinstance(value, (int, float, Decimal)):
|
|
dec = Decimal(str(value))
|
|
if dec != dec.to_integral_value():
|
|
raise ValueError
|
|
return int(dec)
|
|
raise ValueError
|
|
except (ValueError, InvalidOperation, TypeError):
|
|
raise ValidationError("نوع مقدار با data_type سازگار نیست — expected integer", field=field)
|
|
if data_type in ("string", "enum"):
|
|
if value is None:
|
|
raise ValidationError("مقدار نمی‌تواند خالی باشد", field=field)
|
|
return str(value)
|
|
if data_type == "boolean":
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str):
|
|
lowered = value.strip().lower()
|
|
if lowered in ("true", "1", "yes"):
|
|
return True
|
|
if lowered in ("false", "0", "no"):
|
|
return False
|
|
raise ValidationError("نوع مقدار با data_type سازگار نیست — expected boolean", field=field)
|
|
if data_type == "date":
|
|
if isinstance(value, date):
|
|
return value
|
|
try:
|
|
return date.fromisoformat(str(value))
|
|
except ValueError:
|
|
raise ValidationError("نوع مقدار با data_type سازگار نیست — expected date (YYYY-MM-DD)", field=field)
|
|
raise ValidationError(f"data_type نامعتبر است: {data_type}", field="data_type")
|
|
|
|
|
|
def typed_column_values(data_type: str, coerced):
|
|
"""Map a coerced value to the EAV typed column dict."""
|
|
if data_type in ("number", "integer"):
|
|
return {"value_number": coerced}
|
|
if data_type in ("string", "enum"):
|
|
return {"value_string": coerced}
|
|
if data_type == "boolean":
|
|
return {"value_boolean": coerced}
|
|
if data_type == "date":
|
|
return {"value_date": coerced}
|
|
raise ValidationError(f"data_type نامعتبر است: {data_type}", field="data_type")
|
|
|
|
|
|
async def validate_attributes(
|
|
db: AsyncSession, attributes: list[MaterialAttributeIn]
|
|
) -> list[tuple[AttributeDefinition, dict]]:
|
|
"""Validate an EAV payload; returns [(definition, typed-column-dict)]."""
|
|
if not attributes:
|
|
return []
|
|
def_ids = {a.attribute_definition_id for a in attributes}
|
|
defs = {d.id: d for d in (await db.execute(select(AttributeDefinition).where(AttributeDefinition.id.in_(def_ids)))).scalars()}
|
|
|
|
result: list[tuple[AttributeDefinition, dict]] = []
|
|
for index, attr in enumerate(attributes):
|
|
ad = defs.get(attr.attribute_definition_id)
|
|
if ad is None:
|
|
raise NotFoundError("ویژگی یافت نشد", field=f"attributes[{index}].attribute_definition_id")
|
|
field = f"attributes[{index}].value"
|
|
coerced = coerce_eav_value(ad.data_type, attr.value, field)
|
|
if ad.data_type == "enum":
|
|
allowed = ad.enum_values or []
|
|
if allowed and str(attr.value) not in {str(v) for v in allowed}:
|
|
raise ValidationError("مقدار در گزینه‌های مجاز این ویژگی نیست", field=field)
|
|
result.append((ad, typed_column_values(ad.data_type, coerced)))
|
|
return result
|
|
|
|
|
|
async def create_material(db: AsyncSession, payload: MaterialCreate) -> Material:
|
|
"""Create a material + brand_materials rows + EAV values in ONE transaction.
|
|
|
|
v2.5/2.6: category_id is OPTIONAL — a global material may be created
|
|
without a category (it gets derived automatically from the first linked
|
|
supplier). name is GLOBALLY unique.
|
|
"""
|
|
async with db.begin():
|
|
duplicate = await db.scalar(
|
|
select(Material.id).where(Material.name == payload.name)
|
|
)
|
|
if duplicate is not None:
|
|
raise ConflictError("متریالی با این نام در کل کاتالوگ قبلاً ثبت شده است", field="name")
|
|
|
|
if payload.category_id is not None:
|
|
if await db.get(MaterialCategory, payload.category_id) is None:
|
|
raise NotFoundError("دسته‌بندی یافت نشد", field="category_id")
|
|
if not await db.get(Unit, payload.canonical_unit_id):
|
|
raise NotFoundError("واحد مرجع یافت نشد", field="canonical_unit_id")
|
|
|
|
material = Material(
|
|
category_id=payload.category_id,
|
|
name=payload.name,
|
|
topic=payload.topic,
|
|
canonical_unit_id=payload.canonical_unit_id,
|
|
status=payload.status,
|
|
)
|
|
db.add(material)
|
|
await db.flush()
|
|
|
|
if payload.brand_ids:
|
|
brands = (await db.execute(select(Brand).where(Brand.id.in_(payload.brand_ids)))).scalars().all()
|
|
found = {b.id for b in brands}
|
|
for brand_id in payload.brand_ids:
|
|
if brand_id not in found:
|
|
raise NotFoundError("برند یافت نشد", field="brand_ids")
|
|
db.add(BrandMaterial(brand_id=brand_id, material_id=material.id))
|
|
|
|
validated = await validate_attributes(db, payload.attributes)
|
|
for ad, values in validated:
|
|
db.add(
|
|
MaterialAttributeValue(
|
|
material_id=material.id, attribute_definition_id=ad.id, **values
|
|
)
|
|
)
|
|
return material
|
|
|
|
|
|
async def update_material(db: AsyncSession, material: Material, payload: MaterialUpdate) -> Material:
|
|
"""Partial update (name, canonical_unit_id, status). EAV/brands are NOT edited here.
|
|
|
|
v2.5/2.6: category is DERIVED (from the first linked supplier) — it is
|
|
no longer part of the update payload.
|
|
"""
|
|
async with db.begin():
|
|
if payload.name is not None:
|
|
duplicate = await db.scalar(
|
|
select(Material.id).where(
|
|
Material.name == payload.name,
|
|
Material.id != material.id,
|
|
)
|
|
)
|
|
if duplicate is not None:
|
|
raise ConflictError("متریالی با این نام در کل کاتالوگ قبلاً ثبت شده است", field="name")
|
|
if payload.name is not None:
|
|
material.name = payload.name
|
|
if payload.topic is not None:
|
|
material.topic = payload.topic
|
|
if payload.canonical_unit_id is not None and payload.canonical_unit_id != material.canonical_unit_id:
|
|
if not await db.get(Unit, payload.canonical_unit_id):
|
|
raise NotFoundError("واحد مرجع یافت نشد", field="canonical_unit_id")
|
|
material.canonical_unit_id = payload.canonical_unit_id
|
|
if payload.status is not None:
|
|
material.status = payload.status
|
|
return material
|
|
|
|
|
|
async def ensure_brand_material_belong(
|
|
db: AsyncSession, brand_material_id: UUID | None, material_id: UUID
|
|
) -> None:
|
|
"""App-layer integrity check: brand_material_id must belong to material_id (migration design note)."""
|
|
if brand_material_id is None:
|
|
return
|
|
bm = await db.get(BrandMaterial, brand_material_id)
|
|
if bm is None:
|
|
raise NotFoundError("برند-متریال یافت نشد", field="brand_material_id")
|
|
if bm.material_id != material_id:
|
|
raise ConflictError("brand_material_id متعلق به این متریال نیست", field="brand_material_id")
|