Files
orion/app/modules/inventory/models/inventory_transaction.py
Samir Boulahtit f20266167d
Some checks failed
CI / ruff (push) Failing after 7s
CI / pytest (push) Failing after 1s
CI / architecture (push) Failing after 9s
CI / dependency-scanning (push) Successful in 27s
CI / audit (push) Successful in 8s
CI / docs (push) Has been skipped
fix(lint): auto-fix ruff violations and tune lint rules
- Auto-fixed 4,496 lint issues (import sorting, modern syntax, etc.)
- Added ignore rules for patterns intentional in this codebase:
  E402 (late imports), E712 (SQLAlchemy filters), B904 (raise from),
  SIM108/SIM105/SIM117 (readability preferences)
- Added per-file ignores for tests and scripts
- Excluded broken scripts/rename_terminology.py (has curly quotes)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-12 23:10:42 +01:00

173 lines
5.4 KiB
Python

# app/modules/inventory/models/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,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy import (
Enum as SQLEnum,
)
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
store_id = Column(Integer, ForeignKey("stores.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
store = relationship("Store")
product = relationship("Product")
inventory = relationship("Inventory")
order = relationship("Order")
# Indexes for common queries
__table_args__ = (
Index("idx_inv_tx_store_product", "store_id", "product_id"),
Index("idx_inv_tx_store_created", "store_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,
store_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:
store_id: Store 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(
store_id=store_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,
)