feat: add inventory transaction audit trail (Phase 2)

Adds complete audit trail for all stock movements:
- InventoryTransaction model with transaction types (reserve, fulfill,
  release, adjust, set, import, return)
- Alembic migration for inventory_transactions table
- Transaction logging in order_inventory_service for all order operations
- Captures quantity snapshots, order references, and timestamps

Each inventory operation now creates a transaction record for
accountability and debugging.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-01 18:12:09 +01:00
parent 871f52da80
commit 049e3319c3
5 changed files with 468 additions and 7 deletions

View File

@@ -21,6 +21,7 @@ from .customer import Customer, CustomerAddress
from .email import EmailCategory, EmailLog, EmailStatus, EmailTemplate
from .feature import Feature, FeatureCategory, FeatureCode, FeatureUILocation
from .inventory import Inventory
from .inventory_transaction import InventoryTransaction, TransactionType
from .invoice import (
Invoice,
InvoiceStatus,
@@ -127,6 +128,8 @@ __all__ = [
"MarketplaceImportError",
# Inventory
"Inventory",
"InventoryTransaction",
"TransactionType",
# Invoicing
"Invoice",
"InvoiceStatus",

View File

@@ -0,0 +1,170 @@
# models/database/inventory_transaction.py
"""
Inventory Transaction Model - Audit trail for all stock movements.
This model tracks every change to inventory quantities, providing:
- Complete audit trail for compliance and debugging
- Order-linked transactions for traceability
- Support for different transaction types (reserve, fulfill, adjust, etc.)
All stock movements should create a transaction record.
"""
from datetime import UTC, datetime
from enum import Enum
from sqlalchemy import (
Column,
DateTime,
Enum as SQLEnum,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
class TransactionType(str, Enum):
"""Types of inventory transactions."""
# Order-related
RESERVE = "reserve" # Stock reserved for order
FULFILL = "fulfill" # Reserved stock consumed (shipped)
RELEASE = "release" # Reserved stock released (cancelled)
# Manual adjustments
ADJUST = "adjust" # Manual adjustment (+/-)
SET = "set" # Set to exact quantity
# Imports
IMPORT = "import" # Initial import/sync
# Returns
RETURN = "return" # Stock returned from customer
class InventoryTransaction(Base):
"""
Audit log for inventory movements.
Every change to inventory quantity creates a transaction record,
enabling complete traceability of stock levels over time.
"""
__tablename__ = "inventory_transactions"
id = Column(Integer, primary_key=True, index=True)
# Core references
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=False, index=True)
product_id = Column(Integer, ForeignKey("products.id"), nullable=False, index=True)
inventory_id = Column(
Integer, ForeignKey("inventory.id"), nullable=True, index=True
)
# Transaction details
transaction_type = Column(
SQLEnum(TransactionType), nullable=False, index=True
)
quantity_change = Column(Integer, nullable=False) # Positive = add, negative = remove
# Quantities after transaction (snapshot)
quantity_after = Column(Integer, nullable=False)
reserved_after = Column(Integer, nullable=False, default=0)
# Location context
location = Column(String, nullable=True)
warehouse = Column(String, nullable=True)
# Order reference (for order-related transactions)
order_id = Column(Integer, ForeignKey("orders.id"), nullable=True, index=True)
order_number = Column(String, nullable=True)
# Audit fields
reason = Column(Text, nullable=True) # Human-readable reason
created_by = Column(String, nullable=True) # User/system that created
created_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(UTC),
nullable=False,
index=True,
)
# Relationships
vendor = relationship("Vendor")
product = relationship("Product")
inventory = relationship("Inventory")
order = relationship("Order")
# Indexes for common queries
__table_args__ = (
Index("idx_inv_tx_vendor_product", "vendor_id", "product_id"),
Index("idx_inv_tx_vendor_created", "vendor_id", "created_at"),
Index("idx_inv_tx_order", "order_id"),
Index("idx_inv_tx_type_created", "transaction_type", "created_at"),
)
def __repr__(self) -> str:
return (
f"<InventoryTransaction {self.id}: "
f"{self.transaction_type.value} {self.quantity_change:+d} "
f"for product {self.product_id}>"
)
@classmethod
def create_transaction(
cls,
vendor_id: int,
product_id: int,
transaction_type: TransactionType,
quantity_change: int,
quantity_after: int,
reserved_after: int = 0,
inventory_id: int | None = None,
location: str | None = None,
warehouse: str | None = None,
order_id: int | None = None,
order_number: str | None = None,
reason: str | None = None,
created_by: str | None = None,
) -> "InventoryTransaction":
"""
Factory method to create a transaction record.
Args:
vendor_id: Vendor ID
product_id: Product ID
transaction_type: Type of transaction
quantity_change: Change in quantity (positive = add, negative = remove)
quantity_after: Total quantity after this transaction
reserved_after: Reserved quantity after this transaction
inventory_id: Optional inventory record ID
location: Optional location
warehouse: Optional warehouse
order_id: Optional order ID (for order-related transactions)
order_number: Optional order number for display
reason: Optional human-readable reason
created_by: Optional user/system identifier
Returns:
InventoryTransaction instance (not yet added to session)
"""
return cls(
vendor_id=vendor_id,
product_id=product_id,
inventory_id=inventory_id,
transaction_type=transaction_type,
quantity_change=quantity_change,
quantity_after=quantity_after,
reserved_after=reserved_after,
location=location,
warehouse=warehouse,
order_id=order_id,
order_number=order_number,
reason=reason,
created_by=created_by,
)