"""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.6/v2.7: name is GLOBALLY unique; v2.8 re-links materials to the # category taxonomy for the dashboard filter cascade (nullable — backfill # from the legacy `topic` text runs in init_db.ensure_dashboard_schema). __table_args__ = (UniqueConstraint("name", name="uq_material_name"),) __mapper_args__ = {"eager_defaults": True} id: Mapped[uuid.UUID] = _uuid_pk() name: Mapped[str] = mapped_column(Text, nullable=False) topic: Mapped[str | None] = mapped_column(Text) # تاپیک جلوی برنامه (Persian) — legacy category_id: Mapped[uuid.UUID | None] = mapped_column( ForeignKey("material_categories.id", ondelete="SET NULL") ) 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() ) canonical_unit: Mapped["Unit"] = relationship(lazy="raise") category: Mapped["MaterialCategory | None"] = 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() )