262 lines
11 KiB
Python
262 lines
11 KiB
Python
"""SQLAlchemy models for the CATALOG domain — mirrors migration.sql exactly.
|
|
|
|
Column types, nullability, server defaults, FKs, unique constraints and
|
|
indexes all follow the DDL in `data base/migration.sql`. No table is
|
|
created through SQLAlchemy metadata; the schema already exists in Postgres.
|
|
"""
|
|
import uuid
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import (
|
|
Boolean,
|
|
Date,
|
|
DateTime,
|
|
Enum,
|
|
ForeignKey,
|
|
Identity,
|
|
Integer,
|
|
Numeric,
|
|
SmallInteger,
|
|
String,
|
|
Text,
|
|
UniqueConstraint,
|
|
Uuid,
|
|
func,
|
|
text,
|
|
)
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from app.db.base import Base
|
|
|
|
attribute_data_type_enum = Enum(
|
|
"number", "integer", "string", "enum", "boolean", "date",
|
|
name="attribute_data_type",
|
|
native_enum=True,
|
|
)
|
|
|
|
material_status_enum = Enum(
|
|
"active", "discontinued", "draft",
|
|
name="material_status",
|
|
native_enum=True,
|
|
)
|
|
|
|
|
|
def _uuid_pk() -> Mapped[uuid.UUID]:
|
|
"""UUID primary key, server-generated via gen_random_uuid()."""
|
|
return mapped_column(Uuid(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()"))
|
|
|
|
|
|
def _timestamptz() -> Mapped[datetime]:
|
|
return mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
|
|
|
|
|
|
class Unit(Base):
|
|
"""units — the only non-UUID PK in the system (INTEGER IDENTITY)."""
|
|
|
|
__tablename__ = "units"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, Identity(), primary_key=True)
|
|
code: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
|
|
name_fa: Mapped[str | None] = mapped_column(String)
|
|
dimension: Mapped[str | None] = mapped_column(String) # length, mass, pressure, ...
|
|
si_factor: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False, server_default=text("1.0"))
|
|
is_base: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
|
|
|
|
|
|
class Brand(Base):
|
|
__tablename__ = "brands"
|
|
__table_args__ = (UniqueConstraint("name", "country", name="uq_brands_name_country"),)
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
name: Mapped[str] = mapped_column(Text, nullable=False)
|
|
country: Mapped[str | None] = mapped_column(Text)
|
|
created_at: Mapped[datetime] = _timestamptz()
|
|
|
|
|
|
class MaterialCategory(Base):
|
|
__tablename__ = "material_categories"
|
|
# eager_defaults: fetch onupdate=func.now() via RETURNING during flush so
|
|
# updated_at is never expired post-flush (lazy refresh raises
|
|
# MissingGreenlet in async code).
|
|
__mapper_args__ = {"eager_defaults": True}
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
slug: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
|
|
name: Mapped[str] = mapped_column(Text, nullable=False)
|
|
description: Mapped[str | None] = mapped_column(Text)
|
|
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true"))
|
|
created_at: Mapped[datetime] = _timestamptz()
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
|
)
|
|
|
|
|
|
class AttributeDefinition(Base):
|
|
"""Global attribute dictionary (EAV definition side)."""
|
|
|
|
__tablename__ = "attribute_definitions"
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
attribute_key: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
|
|
label: Mapped[str] = mapped_column(Text, nullable=False)
|
|
data_type: Mapped[str] = mapped_column(attribute_data_type_enum, nullable=False)
|
|
unit_id: Mapped[int | None] = mapped_column(ForeignKey("units.id", ondelete="RESTRICT"))
|
|
enum_values: Mapped[list | None] = mapped_column(JSONB) # allowed options for enum types
|
|
created_at: Mapped[datetime] = _timestamptz()
|
|
|
|
|
|
class CategoryAttribute(Base):
|
|
"""Category-attribute junction (scopes global attributes & sets category rules)."""
|
|
|
|
__tablename__ = "category_attributes"
|
|
__table_args__ = (UniqueConstraint("category_id", "attribute_definition_id", name="uq_category_attribute"),)
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
category_id: Mapped[uuid.UUID] = mapped_column(
|
|
ForeignKey("material_categories.id", ondelete="CASCADE"), nullable=False
|
|
)
|
|
attribute_definition_id: Mapped[uuid.UUID] = mapped_column(
|
|
ForeignKey("attribute_definitions.id", ondelete="CASCADE"), nullable=False
|
|
)
|
|
is_required: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
|
|
is_filterable: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
|
|
conditional_rules: Mapped[dict | None] = mapped_column(JSONB)
|
|
sort_order: Mapped[int] = mapped_column(SmallInteger, nullable=False, server_default=text("0"))
|
|
|
|
attribute: Mapped["AttributeDefinition"] = relationship(lazy="raise")
|
|
|
|
|
|
class Material(Base):
|
|
__tablename__ = "materials"
|
|
# v2.5/2.6: name is GLOBALLY unique; category_id is nullable + derived.
|
|
__table_args__ = (UniqueConstraint("name", name="uq_material_name"),)
|
|
__mapper_args__ = {"eager_defaults": True}
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
category_id: Mapped[uuid.UUID | None] = mapped_column(
|
|
ForeignKey("material_categories.id", ondelete="RESTRICT")
|
|
)
|
|
name: Mapped[str] = mapped_column(Text, nullable=False)
|
|
topic: Mapped[str | None] = mapped_column(Text) # تاپیک جلوی برنامه (Persian) — legacy
|
|
canonical_unit_id: Mapped[int] = mapped_column(
|
|
ForeignKey("units.id", ondelete="RESTRICT"), nullable=False
|
|
)
|
|
status: Mapped[str] = mapped_column(material_status_enum, nullable=False, server_default=text("'active'"))
|
|
created_at: Mapped[datetime] = _timestamptz()
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
|
)
|
|
|
|
category: Mapped["MaterialCategory | None"] = relationship(lazy="raise")
|
|
canonical_unit: Mapped["Unit"] = relationship(lazy="raise")
|
|
attribute_values: Mapped[list["MaterialAttributeValue"]] = relationship(
|
|
back_populates="material", lazy="raise", passive_deletes=True
|
|
)
|
|
|
|
|
|
class BrandMaterial(Base):
|
|
"""M:N Brand <-> Material junction."""
|
|
|
|
__tablename__ = "brand_materials"
|
|
__table_args__ = (UniqueConstraint("brand_id", "material_id", name="uq_brand_material"),)
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
brand_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("brands.id", ondelete="CASCADE"), nullable=False)
|
|
material_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("materials.id", ondelete="CASCADE"), nullable=False)
|
|
catalog_code: Mapped[str | None] = mapped_column(Text)
|
|
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true"))
|
|
created_at: Mapped[datetime] = _timestamptz()
|
|
|
|
brand: Mapped["Brand"] = relationship(lazy="raise")
|
|
material: Mapped["Material"] = relationship(lazy="raise")
|
|
|
|
|
|
class MaterialUnit(Base):
|
|
"""Packaging / unit overrides per material (and optionally per brand)."""
|
|
|
|
__tablename__ = "material_units"
|
|
__table_args__ = (
|
|
UniqueConstraint("material_id", "brand_material_id", "unit_id", name="uq_material_unit"),
|
|
)
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
material_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("materials.id", ondelete="CASCADE"), nullable=False)
|
|
brand_material_id: Mapped[uuid.UUID | None] = mapped_column(
|
|
ForeignKey("brand_materials.id", ondelete="CASCADE")
|
|
) # NULL = generic material fallback
|
|
unit_id: Mapped[int] = mapped_column(ForeignKey("units.id", ondelete="RESTRICT"), nullable=False)
|
|
conversion_factor: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False)
|
|
is_default: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
|
|
|
|
|
|
class MaterialAttributeValue(Base):
|
|
"""Typed EAV values for a material."""
|
|
|
|
__tablename__ = "material_attribute_values"
|
|
__table_args__ = (UniqueConstraint("material_id", "attribute_definition_id", name="uq_material_attr_value"),)
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
material_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("materials.id", ondelete="CASCADE"), nullable=False)
|
|
attribute_definition_id: Mapped[uuid.UUID] = mapped_column(
|
|
ForeignKey("attribute_definitions.id", ondelete="RESTRICT"), nullable=False
|
|
)
|
|
value_number: Mapped[Decimal | None] = mapped_column(Numeric(18, 6))
|
|
value_string: Mapped[str | None] = mapped_column(Text)
|
|
value_boolean: Mapped[bool | None] = mapped_column(Boolean)
|
|
value_date: Mapped[date | None] = mapped_column(Date)
|
|
|
|
material: Mapped["Material"] = relationship(back_populates="attribute_values", lazy="raise")
|
|
definition: Mapped["AttributeDefinition"] = relationship(lazy="raise")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Units-dashboard tables (NEW v2 contracts) — created IF NOT EXISTS at startup
|
|
# by app.db.init_db.ensure_dashboard_tables.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class CategoryReferenceUnit(Base):
|
|
__tablename__ = "category_reference_units"
|
|
__mapper_args__ = {"eager_defaults": True}
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
category_id: Mapped[uuid.UUID] = mapped_column(
|
|
ForeignKey("material_categories.id", ondelete="CASCADE"), nullable=False, unique=True
|
|
)
|
|
ref_unit: Mapped[str] = mapped_column(Text, nullable=False)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
|
)
|
|
|
|
|
|
class UnitConversion(Base):
|
|
__tablename__ = "unit_conversions"
|
|
__mapper_args__ = {"eager_defaults": True}
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
category_id: Mapped[uuid.UUID] = mapped_column(
|
|
ForeignKey("material_categories.id", ondelete="CASCADE"), nullable=False
|
|
)
|
|
brand: Mapped[str | None] = mapped_column(Text)
|
|
from_unit: Mapped[str] = mapped_column(Text, nullable=False)
|
|
to_unit: Mapped[str] = mapped_column(Text, nullable=False)
|
|
factor: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
|
)
|
|
|
|
|
|
class StandardConversion(Base):
|
|
__tablename__ = "standard_conversions"
|
|
__table_args__ = (UniqueConstraint("from_unit", "to_unit", name="uq_standard_conversion"),)
|
|
__mapper_args__ = {"eager_defaults": True}
|
|
|
|
id: Mapped[uuid.UUID] = _uuid_pk()
|
|
from_unit: Mapped[str] = mapped_column(Text, nullable=False)
|
|
to_unit: Mapped[str] = mapped_column(Text, nullable=False)
|
|
factor: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False)
|
|
updated_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
|
|
)
|