orion/docs/implementation/product-suppliers-table.md

Product Suppliers Table - Implementation Plan

Status: Planned

This document outlines the architecture for a dedicated product_suppliers table to support multiple suppliers per product with independent pricing.

---

Problem Statement

Currently, the Product model has:

• supplier - Single supplier code string
• supplier_product_id - Single supplier reference
• cost_cents - Single cost value

This limits vendors to one supplier per product. In reality:

• A vendor may source the same product from multiple suppliers
• Each supplier has different costs, lead times, and availability
• The vendor may want to track cost history and switch suppliers

Proposed Solution

New Table: product_suppliers

class ProductSupplier(Base, TimestampMixin):
    """Supplier pricing for a vendor product.

    Allows multiple suppliers per product with independent costs.
    """
    __tablename__ = "product_suppliers"

    id = Column(Integer, primary_key=True, index=True)
    product_id = Column(Integer, ForeignKey("products.id"), nullable=False, index=True)

    # === SUPPLIER IDENTIFICATION ===
    supplier_code = Column(String(50), nullable=False, index=True)  # 'codeswholesale', 'direct', 'wholesale_partner'
    supplier_name = Column(String(200))  # Human-readable name
    supplier_product_id = Column(String(100))  # Supplier's product reference/SKU
    supplier_product_url = Column(String(500))  # Link to supplier's product page

    # === PRICING (integer cents) ===
    cost_cents = Column(Integer, nullable=False)  # What vendor pays this supplier
    currency = Column(String(3), default="EUR")

    # === AVAILABILITY ===
    is_active = Column(Boolean, default=True)  # Supplier still offers this product
    is_primary = Column(Boolean, default=False)  # Primary supplier for auto-ordering
    stock_status = Column(String(20))  # 'in_stock', 'low_stock', 'out_of_stock', 'discontinued'
    lead_time_days = Column(Integer)  # Days to receive after ordering
    min_order_quantity = Column(Integer, default=1)

    # === TRACKING ===
    last_price_update = Column(DateTime)  # When cost was last verified
    last_order_date = Column(DateTime)  # Last time ordered from this supplier
    total_orders = Column(Integer, default=0)  # Historical order count

    # === NOTES ===
    notes = Column(Text)  # Internal notes about this supplier relationship

    # === RELATIONSHIPS ===
    product = relationship("Product", back_populates="suppliers")

    # === CONSTRAINTS ===
    __table_args__ = (
        UniqueConstraint("product_id", "supplier_code", name="uq_product_supplier"),
        Index("idx_supplier_product_active", "product_id", "is_active"),
        Index("idx_supplier_code_active", "supplier_code", "is_active"),
    )

Product Model Changes

class Product(Base, TimestampMixin):
    # ... existing fields ...

    # === DEPRECATED (keep for migration, remove later) ===
    # supplier = Column(String(50))  # Deprecated: use suppliers relationship
    # supplier_product_id = Column(String)  # Deprecated: use suppliers relationship

    # === COST (denormalized for performance) ===
    # This is the effective cost, calculated from primary supplier or average
    cost_cents = Column(Integer)  # Keep as denormalized field
    cost_calculation_method = Column(String(20), default="primary")  # 'primary', 'lowest', 'average'

    # === RELATIONSHIPS ===
    suppliers = relationship(
        "ProductSupplier",
        back_populates="product",
        cascade="all, delete-orphan",
        order_by="ProductSupplier.is_primary.desc()"
    )

    # === COMPUTED PROPERTIES ===
    @property
    def primary_supplier(self) -> "ProductSupplier | None":
        """Get the primary supplier for this product."""
        for s in self.suppliers:
            if s.is_primary and s.is_active:
                return s
        # Fallback to first active supplier
        for s in self.suppliers:
            if s.is_active:
                return s
        return None

    @property
    def lowest_cost_cents(self) -> int | None:
        """Get lowest cost across all active suppliers."""
        costs = [s.cost_cents for s in self.suppliers if s.is_active and s.cost_cents]
        return min(costs) if costs else None

    @property
    def average_cost_cents(self) -> int | None:
        """Get average cost across all active suppliers."""
        costs = [s.cost_cents for s in self.suppliers if s.is_active and s.cost_cents]
        return int(sum(costs) / len(costs)) if costs else None

    def update_effective_cost(self) -> None:
        """Update denormalized cost_cents based on calculation method."""
        if self.cost_calculation_method == "primary":
            supplier = self.primary_supplier
            self.cost_cents = supplier.cost_cents if supplier else None
        elif self.cost_calculation_method == "lowest":
            self.cost_cents = self.lowest_cost_cents
        elif self.cost_calculation_method == "average":
            self.cost_cents = self.average_cost_cents

---

Implementation Steps

Phase 1: Database Schema

1. Create migration

   • Add product_suppliers table
   • Add cost_calculation_method to products
   • Keep existing supplier, supplier_product_id, cost_cents fields

2. Migrate existing data

   INSERT INTO product_suppliers (product_id, supplier_code, supplier_product_id, cost_cents, is_primary)
   SELECT id, supplier, supplier_product_id, cost_cents, TRUE
   FROM products
   WHERE supplier IS NOT NULL;
   

3. Update models

   • Add ProductSupplier model
   • Add relationship to Product
   • Add computed properties

Phase 2: Service Layer

1. Create ProductSupplierService

   • add_supplier(product_id, supplier_data)
   • update_supplier(supplier_id, data)
   • remove_supplier(supplier_id)
   • set_primary_supplier(product_id, supplier_id)
   • sync_supplier_costs(supplier_code) - Bulk update from supplier API

2. Update VendorProductService

   • Include suppliers in product detail response
   • Update cost calculation on supplier changes

3. Update import services

   • CodesWholesale import creates/updates ProductSupplier records
   • Other supplier imports follow same pattern

Phase 3: API Endpoints

Admin Endpoints

GET    /api/v1/admin/products/{id}/suppliers       # List suppliers
POST   /api/v1/admin/products/{id}/suppliers       # Add supplier
PUT    /api/v1/admin/products/{id}/suppliers/{sid} # Update supplier
DELETE /api/v1/admin/products/{id}/suppliers/{sid} # Remove supplier
POST   /api/v1/admin/products/{id}/suppliers/{sid}/set-primary

Vendor Endpoints

GET    /api/v1/vendor/products/{id}/suppliers
POST   /api/v1/vendor/products/{id}/suppliers
PUT    /api/v1/vendor/products/{id}/suppliers/{sid}
DELETE /api/v1/vendor/products/{id}/suppliers/{sid}

Phase 4: Frontend

1. Product Detail Page

   • Suppliers tab/section showing all suppliers
   • Add/edit supplier modal
   • Set primary supplier button
   • Cost comparison view

2. Bulk Operations

   • Import suppliers from CSV
   • Sync costs from supplier API
   • Update all products from a supplier

Phase 5: Cleanup

1. Remove deprecated fields (after migration period)

   • Drop products.supplier
   • Drop products.supplier_product_id

2. Update documentation

   • API documentation
   • Architecture docs

---

Pydantic Schemas

# models/schema/product_supplier.py

class ProductSupplierBase(BaseModel):
    supplier_code: str
    supplier_name: str | None = None
    supplier_product_id: str | None = None
    supplier_product_url: str | None = None
    cost: float  # Euros (converted to cents internally)
    currency: str = "EUR"
    is_active: bool = True
    is_primary: bool = False
    stock_status: str | None = None
    lead_time_days: int | None = None
    min_order_quantity: int = 1
    notes: str | None = None


class ProductSupplierCreate(ProductSupplierBase):
    pass


class ProductSupplierUpdate(BaseModel):
    supplier_name: str | None = None
    supplier_product_id: str | None = None
    supplier_product_url: str | None = None
    cost: float | None = None
    is_active: bool | None = None
    stock_status: str | None = None
    lead_time_days: int | None = None
    min_order_quantity: int | None = None
    notes: str | None = None


class ProductSupplierResponse(ProductSupplierBase):
    id: int
    product_id: int
    cost_cents: int
    last_price_update: datetime | None = None
    last_order_date: datetime | None = None
    total_orders: int = 0
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True

---

Known Supplier Codes

Code	Name	Type	Notes
codeswholesale	CodesWholesale	API	Digital game keys
direct	Direct/Internal	Manual	Vendor's own inventory
wholesale_partner	Wholesale Partner	Manual	B2B partner
dropship	Dropship Supplier	Manual	Ships directly to customer

---

Benefits

1. Multi-supplier support: Track costs from multiple sources
2. Price comparison: See which supplier offers best price
3. Cost history: Track price changes over time
4. Automatic ordering: Route orders to primary supplier
5. Supplier analytics: See order volume per supplier
6. Graceful transitions: Switch suppliers without losing data

---

Estimated Effort

Phase	Effort	Priority
Phase 1: Database	2-3 hours	High
Phase 2: Services	3-4 hours	High
Phase 3: API	2-3 hours	Medium
Phase 4: Frontend	4-6 hours	Medium
Phase 5: Cleanup	1 hour	Low

Total: ~12-17 hours

---

Plan created: 2025-12-20