Spaces:
Sleeping
Sleeping
File size: 5,906 Bytes
db7b74e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
"""
Inventory Transfer Models - Agent-to-Agent Equipment Transfers
WORKFLOW:
1. Agent A initiates transfer to Agent B
2. Optional: PM/Dispatcher approves (if requires_approval=true)
3. Agent B accepts transfer
4. System creates new assignment for Agent B
5. Original assignment marked as transferred
6. Transfer marked as completed
USE CASES:
- Equipment shortage: Agent with extra equipment helps another agent
- Job reassignment: Transfer equipment when job is reassigned
- Emergency: Urgent need for equipment, faster than going to hub
"""
from sqlalchemy import Column, String, Boolean, Integer, Text, DateTime, Numeric, ForeignKey, CheckConstraint, Index
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import relationship
from datetime import datetime
from decimal import Decimal
from app.models.base import BaseModel
class InventoryTransfer(BaseModel):
"""
Agent-to-agent inventory transfers
Tracks peer-to-peer equipment transfers between field agents.
Supports approval workflow and acceptance by recipient.
"""
__tablename__ = "inventory_transfers"
# Source and destination
from_assignment_id = Column(UUID(as_uuid=True), ForeignKey("inventory_assignments.id", ondelete="RESTRICT"), nullable=False)
from_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False)
to_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False)
to_assignment_id = Column(UUID(as_uuid=True), ForeignKey("inventory_assignments.id", ondelete="SET NULL"))
# Transfer details
transfer_reason = Column(Text, nullable=False)
quantity = Column(Numeric(10, 2), nullable=False, default=1)
unit_identifier = Column(Text, nullable=False)
# Approval workflow
status = Column(String, nullable=False, default="pending") # TransferStatus enum
requires_approval = Column(Boolean, nullable=False, default=True)
approved_by_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"))
approved_at = Column(DateTime(timezone=True))
approval_notes = Column(Text)
# Rejection
rejected_by_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"))
rejected_at = Column(DateTime(timezone=True))
rejection_reason = Column(Text)
# Acceptance by recipient
accepted_by_to_user_at = Column(DateTime(timezone=True))
# Completion
completed_at = Column(DateTime(timezone=True))
# Location tracking
transfer_latitude = Column(Numeric(10, 7))
transfer_longitude = Column(Numeric(10, 7))
transfer_location_name = Column(Text)
location_verified = Column(Boolean, default=False)
# Metadata
notes = Column(Text)
additional_metadata = Column(JSONB, default={})
# Audit
initiated_by_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False)
# Relationships
from_assignment = relationship("InventoryAssignment", foreign_keys=[from_assignment_id],
backref="outgoing_transfers")
to_assignment = relationship("InventoryAssignment", foreign_keys=[to_assignment_id],
backref="incoming_transfers")
from_user = relationship("User", foreign_keys=[from_user_id])
to_user = relationship("User", foreign_keys=[to_user_id])
approved_by = relationship("User", foreign_keys=[approved_by_user_id])
rejected_by = relationship("User", foreign_keys=[rejected_by_user_id])
initiated_by = relationship("User", foreign_keys=[initiated_by_user_id])
# Table constraints
__table_args__ = (
CheckConstraint("quantity > 0", name="chk_positive_transfer_quantity"),
CheckConstraint("from_user_id != to_user_id", name="chk_different_users"),
Index("idx_inventory_transfers_from_assignment", "from_assignment_id",
postgresql_where=(Column("deleted_at").is_(None))),
Index("idx_inventory_transfers_to_assignment", "to_assignment_id",
postgresql_where=(Column("to_assignment_id").isnot(None)) & (Column("deleted_at").is_(None))),
Index("idx_inventory_transfers_from_user", "from_user_id", "status", "created_at",
postgresql_where=(Column("deleted_at").is_(None))),
Index("idx_inventory_transfers_to_user", "to_user_id", "status", "created_at",
postgresql_where=(Column("deleted_at").is_(None))),
Index("idx_inventory_transfers_status", "status", "created_at",
postgresql_where=(Column("deleted_at").is_(None))),
Index("idx_inventory_transfers_pending", "to_user_id", "status",
postgresql_where=(Column("status") == "pending") & (Column("deleted_at").is_(None))),
Index("idx_inventory_transfers_metadata_gin", "additional_metadata", postgresql_using="gin"),
)
@property
def is_pending(self) -> bool:
"""Check if transfer is pending approval or acceptance"""
return self.status == "pending"
@property
def is_completed(self) -> bool:
"""Check if transfer is completed"""
return self.status == "completed"
@property
def can_approve(self) -> bool:
"""Check if transfer can be approved"""
return self.status == "pending" and self.requires_approval and not self.approved_at
@property
def can_accept(self) -> bool:
"""Check if transfer can be accepted by recipient"""
if self.requires_approval:
return self.status == "approved" and not self.accepted_by_to_user_at
return self.status == "pending" and not self.accepted_by_to_user_at
def __repr__(self):
return f"<InventoryTransfer(id={self.id}, from={self.from_user_id}, to={self.to_user_id}, status={self.status})>"
|