Spaces:
Sleeping
Sleeping
File size: 7,478 Bytes
456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 95005e1 456b2e2 | 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 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | """
USER PAYROLL Models
"""
from sqlalchemy import Column, String, Boolean, Integer, Text, Date, DateTime, Numeric, ForeignKey, CheckConstraint, UniqueConstraint
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import relationship
from datetime import datetime, date
from typing import Optional
from decimal import Decimal
from app.models.base import BaseModel
class UserPayroll(BaseModel):
"""
User Payroll (Worker Compensation & Payment Tracking)
Tracks all work done by users and their compensation.
Used for all workers (casual, hourly, salaried).
Key Features:
- Flexible pay periods (weekly, bi-weekly, monthly)
- Aggregated work summary from timesheets (tickets, days)
- Simple compensation: base_earnings + bonus - deductions
- Payment tracking with reference numbers
- Optimistic locking for concurrent updates
Business Rules:
- One payroll record per user per project per pay period
- period_end_date >= period_start_date
- All amounts must be non-negative
- total_amount = base_earnings + bonus - deductions
- Can only be paid once (is_paid flag)
- Version column prevents race conditions during recalculation
Compensation Calculation:
- base_earnings calculated from project_role compensation_type:
* FIXED_RATE: days_worked × base_rate (or hours × base_rate for hourly)
* PER_UNIT: tickets_closed × per_unit_rate
* COMMISSION: ticket_value × commission_percentage
* FIXED_PLUS_COMMISSION: (days × base_rate) + (ticket_value × commission%)
Generation:
- Automatically generated weekly (Monday morning for previous Mon-Sun)
- Recalculated if timesheets are corrected
- Manual generation supported for ad-hoc periods
"""
__tablename__ = "user_payroll"
# Links
user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="RESTRICT"), nullable=False, index=True)
project_id = Column(UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True)
project_team_id = Column(UUID(as_uuid=True), ForeignKey("project_team.id", ondelete="SET NULL"), nullable=True)
# Pay Period
period_start_date = Column(Date, nullable=False, index=True)
period_end_date = Column(Date, nullable=False, index=True)
# Work Summary (aggregated from timesheets)
tickets_closed = Column(Integer, default=0, nullable=False)
days_worked = Column(Integer, default=0, nullable=False)
# Compensation Breakdown
base_earnings = Column(Numeric(12, 2), default=0, nullable=False) # Calculated from compensation_type
bonus_amount = Column(Numeric(12, 2), default=0, nullable=False) # Performance bonuses (manual)
deductions = Column(Numeric(12, 2), default=0, nullable=False) # Deductions (advances, etc.)
total_amount = Column(Numeric(12, 2), nullable=False) # Total compensation
# Calculation Details
calculation_notes = Column(Text, nullable=True)
# Payment Status
is_paid = Column(Boolean, default=False, nullable=False, index=True)
paid_at = Column(DateTime(timezone=True), nullable=True)
payment_method = Column(String, nullable=True) # 'mobile_money', 'bank_transfer', 'cash'
payment_reference = Column(String, nullable=True) # M-Pesa transaction ID, etc.
paid_by_user_id = Column(UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
# Metadata
additional_metadata = Column(JSONB, default={}, nullable=False)
# Concurrency Control (Optimistic Locking)
version = Column(Integer, default=1, nullable=False, index=True)
# Relationships
user = relationship("User", foreign_keys=[user_id])
project = relationship("Project", back_populates="payrolls")
project_team = relationship("ProjectTeam")
paid_by = relationship("User", foreign_keys=[paid_by_user_id])
timesheets = relationship("Timesheet", back_populates="payroll", foreign_keys="[Timesheet.payroll_id]")
# Constraints
__table_args__ = (
UniqueConstraint('user_id', 'project_id', 'period_start_date', 'period_end_date',
name='uq_user_payroll_period'),
CheckConstraint('period_end_date >= period_start_date', name='chk_valid_period'),
CheckConstraint('tickets_closed >= 0', name='chk_positive_tickets'),
CheckConstraint('days_worked >= 0', name='chk_positive_days'),
CheckConstraint('base_earnings >= 0', name='chk_positive_base_earnings'),
CheckConstraint('bonus_amount >= 0', name='chk_positive_bonus'),
CheckConstraint('deductions >= 0', name='chk_positive_deductions'),
CheckConstraint('total_amount >= 0', name='chk_positive_total'),
)
def __repr__(self):
return f"<UserPayroll user_id={self.user_id} period={self.period_start_date} to {self.period_end_date} total={self.total_amount}>"
@property
def is_current_week(self) -> bool:
"""Check if this payroll is for the current week"""
today = date.today()
return self.period_start_date <= today <= self.period_end_date
@property
def can_recalculate(self) -> bool:
"""Check if payroll can be recalculated (not yet paid)"""
return not self.is_paid
@property
def net_earnings(self) -> Decimal:
"""Calculate net earnings (before deductions)"""
return Decimal(self.base_earnings or 0) + Decimal(self.bonus_amount or 0)
def calculate_total(self) -> Decimal:
"""Calculate total compensation amount"""
net = self.net_earnings
total = net - Decimal(self.deductions or 0)
return max(total, Decimal('0')) # Never negative
def mark_as_paid(
self,
paid_by_user_id: UUID,
payment_method: str,
payment_reference: Optional[str] = None
) -> None:
"""Mark payroll as paid"""
if self.is_paid:
raise ValueError("Payroll already marked as paid")
self.is_paid = True
self.paid_at = datetime.utcnow()
self.paid_by_user_id = paid_by_user_id
self.payment_method = payment_method
self.payment_reference = payment_reference
self.updated_at = datetime.utcnow()
def recalculate_from_data(
self,
tickets_closed: int,
days_worked: int,
base_earnings: Decimal,
bonus: Decimal = Decimal('0'),
deductions: Decimal = Decimal('0'),
calculation_notes: Optional[str] = None
) -> None:
"""Recalculate payroll from fresh data (used when timesheets are corrected)"""
if self.is_paid:
raise ValueError("Cannot recalculate paid payroll")
self.tickets_closed = tickets_closed
self.days_worked = days_worked
self.base_earnings = base_earnings
self.bonus_amount = bonus
self.deductions = deductions
self.total_amount = self.calculate_total()
if calculation_notes:
self.calculation_notes = calculation_notes
# Increment version for optimistic locking
self.version += 1
self.updated_at = datetime.utcnow()
@property
def period_label(self) -> str:
"""Human-readable period label"""
return f"{self.period_start_date.strftime('%b %d')} - {self.period_end_date.strftime('%b %d, %Y')}"
|