mitlist/be/app/models.py
mohamad 81577ac7e8 feat: Add Recurrence Pattern and Update Expense Schema
- Introduced a new `RecurrencePattern` model to manage recurrence details for expenses, allowing for daily, weekly, monthly, and yearly patterns.
- Updated the `Expense` model to include fields for recurrence management, such as `is_recurring`, `recurrence_pattern_id`, and `next_occurrence`.
- Modified the database schema to reflect these changes, including alterations to existing columns and the removal of obsolete fields.
- Enhanced the expense creation logic to accommodate recurring expenses and updated related CRUD operations accordingly.
- Implemented necessary migrations to ensure database integrity and support for the new features.
2025-05-23 21:01:49 +02:00

418 lines
21 KiB
Python

# app/models.py
import enum
import secrets
from datetime import datetime, timedelta, timezone
from sqlalchemy import (
Column,
Integer,
String,
DateTime,
ForeignKey,
Boolean,
Enum as SAEnum,
UniqueConstraint,
Index,
DDL,
event,
delete,
func,
text as sa_text,
Text, # <-- Add Text for description
Numeric, # <-- Add Numeric for price
CheckConstraint,
Date # Added Date for Chore model
)
from sqlalchemy.orm import relationship, backref
from .database import Base
# --- Enums ---
class UserRoleEnum(enum.Enum):
owner = "owner"
member = "member"
class SplitTypeEnum(enum.Enum):
EQUAL = "EQUAL" # Split equally among all involved users
EXACT_AMOUNTS = "EXACT_AMOUNTS" # Specific amounts for each user (defined in ExpenseSplit)
PERCENTAGE = "PERCENTAGE" # Percentage for each user (defined in ExpenseSplit)
SHARES = "SHARES" # Proportional to shares/units (defined in ExpenseSplit)
ITEM_BASED = "ITEM_BASED" # If an expense is derived directly from item prices and who added them
# Add more types as needed, e.g., UNPAID (for tracking debts not part of a formal expense)
class ExpenseSplitStatusEnum(enum.Enum):
unpaid = "unpaid"
partially_paid = "partially_paid"
paid = "paid"
class ExpenseOverallStatusEnum(enum.Enum):
unpaid = "unpaid"
partially_paid = "partially_paid"
paid = "paid"
class RecurrenceTypeEnum(enum.Enum):
DAILY = "DAILY"
WEEKLY = "WEEKLY"
MONTHLY = "MONTHLY"
YEARLY = "YEARLY"
# Add more types as needed
# Define ChoreFrequencyEnum
class ChoreFrequencyEnum(enum.Enum):
one_time = "one_time"
daily = "daily"
weekly = "weekly"
monthly = "monthly"
custom = "custom"
class ChoreTypeEnum(enum.Enum):
personal = "personal"
group = "group"
# --- User Model ---
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
name = Column(String, index=True, nullable=True)
is_active = Column(Boolean, default=True, nullable=False)
is_superuser = Column(Boolean, default=False, nullable=False)
is_verified = Column(Boolean, default=False, nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
# --- Relationships ---
created_groups = relationship("Group", back_populates="creator")
group_associations = relationship("UserGroup", back_populates="user", cascade="all, delete-orphan")
created_invites = relationship("Invite", back_populates="creator")
# --- NEW Relationships for Lists/Items ---
created_lists = relationship("List", foreign_keys="List.created_by_id", back_populates="creator") # Link List.created_by_id -> User
added_items = relationship("Item", foreign_keys="Item.added_by_id", back_populates="added_by_user") # Link Item.added_by_id -> User
completed_items = relationship("Item", foreign_keys="Item.completed_by_id", back_populates="completed_by_user") # Link Item.completed_by_id -> User
# --- End NEW Relationships ---
# --- Relationships for Cost Splitting ---
expenses_paid = relationship("Expense", foreign_keys="Expense.paid_by_user_id", back_populates="paid_by_user", cascade="all, delete-orphan")
expenses_created = relationship("Expense", foreign_keys="Expense.created_by_user_id", back_populates="created_by_user", cascade="all, delete-orphan")
expense_splits = relationship("ExpenseSplit", foreign_keys="ExpenseSplit.user_id", back_populates="user", cascade="all, delete-orphan")
settlements_made = relationship("Settlement", foreign_keys="Settlement.paid_by_user_id", back_populates="payer", cascade="all, delete-orphan")
settlements_received = relationship("Settlement", foreign_keys="Settlement.paid_to_user_id", back_populates="payee", cascade="all, delete-orphan")
settlements_created = relationship("Settlement", foreign_keys="Settlement.created_by_user_id", back_populates="created_by_user", cascade="all, delete-orphan")
# --- End Relationships for Cost Splitting ---
# --- Relationships for Chores ---
created_chores = relationship("Chore", foreign_keys="[Chore.created_by_id]", back_populates="creator")
assigned_chores = relationship("ChoreAssignment", back_populates="assigned_user", cascade="all, delete-orphan")
# --- End Relationships for Chores ---
# --- Group Model ---
class Group(Base):
__tablename__ = "groups"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
created_by_id = Column(Integer, ForeignKey("users.id"), nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
# --- Relationships ---
creator = relationship("User", back_populates="created_groups")
member_associations = relationship("UserGroup", back_populates="group", cascade="all, delete-orphan")
invites = relationship("Invite", back_populates="group", cascade="all, delete-orphan")
# --- NEW Relationship for Lists ---
lists = relationship("List", back_populates="group", cascade="all, delete-orphan") # Link List.group_id -> Group
# --- End NEW Relationship ---
# --- Relationships for Cost Splitting ---
expenses = relationship("Expense", foreign_keys="Expense.group_id", back_populates="group", cascade="all, delete-orphan")
settlements = relationship("Settlement", foreign_keys="Settlement.group_id", back_populates="group", cascade="all, delete-orphan")
# --- End Relationships for Cost Splitting ---
# --- Relationship for Chores ---
chores = relationship("Chore", back_populates="group", cascade="all, delete-orphan")
# --- End Relationship for Chores ---
# --- UserGroup Association Model ---
class UserGroup(Base):
__tablename__ = "user_groups"
__table_args__ = (UniqueConstraint('user_id', 'group_id', name='uq_user_group'),)
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
group_id = Column(Integer, ForeignKey("groups.id", ondelete="CASCADE"), nullable=False)
role = Column(SAEnum(UserRoleEnum, name="userroleenum", create_type=True), nullable=False, default=UserRoleEnum.member)
joined_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
user = relationship("User", back_populates="group_associations")
group = relationship("Group", back_populates="member_associations")
# --- Invite Model ---
class Invite(Base):
__tablename__ = "invites"
__table_args__ = (
Index('ix_invites_active_code', 'code', unique=True, postgresql_where=sa_text('is_active = true')),
)
id = Column(Integer, primary_key=True, index=True)
code = Column(String, unique=False, index=True, nullable=False, default=lambda: secrets.token_urlsafe(16))
group_id = Column(Integer, ForeignKey("groups.id", ondelete="CASCADE"), nullable=False)
created_by_id = Column(Integer, ForeignKey("users.id"), nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
expires_at = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc) + timedelta(days=7))
is_active = Column(Boolean, default=True, nullable=False)
group = relationship("Group", back_populates="invites")
creator = relationship("User", back_populates="created_invites")
# === NEW: List Model ===
class List(Base):
__tablename__ = "lists"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
description = Column(Text, nullable=True)
created_by_id = Column(Integer, ForeignKey("users.id"), nullable=False) # Who created this list
group_id = Column(Integer, ForeignKey("groups.id"), nullable=True) # Which group it belongs to (NULL if personal)
is_complete = Column(Boolean, default=False, nullable=False)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
version = Column(Integer, nullable=False, default=1, server_default='1')
# --- Relationships ---
creator = relationship("User", back_populates="created_lists") # Link to User.created_lists
group = relationship("Group", back_populates="lists") # Link to Group.lists
items = relationship("Item", back_populates="list", cascade="all, delete-orphan", order_by="Item.created_at") # Link to Item.list, cascade deletes
# --- Relationships for Cost Splitting ---
expenses = relationship("Expense", foreign_keys="Expense.list_id", back_populates="list", cascade="all, delete-orphan")
# --- End Relationships for Cost Splitting ---
# === NEW: Item Model ===
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
list_id = Column(Integer, ForeignKey("lists.id", ondelete="CASCADE"), nullable=False) # Belongs to which list
name = Column(String, index=True, nullable=False)
quantity = Column(String, nullable=True) # Flexible quantity (e.g., "1", "2 lbs", "a bunch")
is_complete = Column(Boolean, default=False, nullable=False)
price = Column(Numeric(10, 2), nullable=True) # For cost splitting later (e.g., 12345678.99)
added_by_id = Column(Integer, ForeignKey("users.id"), nullable=False) # Who added this item
completed_by_id = Column(Integer, ForeignKey("users.id"), nullable=True) # Who marked it complete
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
version = Column(Integer, nullable=False, default=1, server_default='1')
# --- Relationships ---
list = relationship("List", back_populates="items") # Link to List.items
added_by_user = relationship("User", foreign_keys=[added_by_id], back_populates="added_items") # Link to User.added_items
completed_by_user = relationship("User", foreign_keys=[completed_by_id], back_populates="completed_items") # Link to User.completed_items
# --- Relationships for Cost Splitting ---
# If an item directly results in an expense, or an expense can be tied to an item.
expenses = relationship("Expense", back_populates="item") # An item might have multiple associated expenses
# --- End Relationships for Cost Splitting ---
# === NEW Models for Advanced Cost Splitting ===
class Expense(Base):
__tablename__ = "expenses"
id = Column(Integer, primary_key=True, index=True)
description = Column(String, nullable=False)
total_amount = Column(Numeric(10, 2), nullable=False)
currency = Column(String, nullable=False, default="USD")
expense_date = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
split_type = Column(SAEnum(SplitTypeEnum, name="splittypeenum", create_type=True), nullable=False)
# Foreign Keys
list_id = Column(Integer, ForeignKey("lists.id"), nullable=True, index=True)
group_id = Column(Integer, ForeignKey("groups.id"), nullable=True, index=True)
item_id = Column(Integer, ForeignKey("items.id"), nullable=True)
paid_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
created_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
version = Column(Integer, nullable=False, default=1, server_default='1')
# Relationships
paid_by_user = relationship("User", foreign_keys=[paid_by_user_id], back_populates="expenses_paid")
created_by_user = relationship("User", foreign_keys=[created_by_user_id], back_populates="expenses_created")
list = relationship("List", foreign_keys=[list_id], back_populates="expenses")
group = relationship("Group", foreign_keys=[group_id], back_populates="expenses")
item = relationship("Item", foreign_keys=[item_id], back_populates="expenses")
splits = relationship("ExpenseSplit", back_populates="expense", cascade="all, delete-orphan")
overall_settlement_status = Column(SAEnum(ExpenseOverallStatusEnum, name="expenseoverallstatusenum", create_type=True), nullable=False, server_default=ExpenseOverallStatusEnum.unpaid.value, default=ExpenseOverallStatusEnum.unpaid)
# --- Recurrence fields ---
is_recurring = Column(Boolean, default=False, nullable=False)
recurrence_pattern_id = Column(Integer, ForeignKey("recurrence_patterns.id"), nullable=True)
recurrence_pattern = relationship("RecurrencePattern", back_populates="expenses", uselist=False) # One-to-one
next_occurrence = Column(DateTime(timezone=True), nullable=True) # For recurring expenses
__table_args__ = (
# Ensure at least one context is provided
CheckConstraint('(item_id IS NOT NULL) OR (list_id IS NOT NULL) OR (group_id IS NOT NULL)', name='chk_expense_context'),
)
class ExpenseSplit(Base):
__tablename__ = "expense_splits"
__table_args__ = (
UniqueConstraint('expense_id', 'user_id', name='uq_expense_user_split'),
Index('ix_expense_splits_user_id', 'user_id'), # For looking up user's splits
)
id = Column(Integer, primary_key=True, index=True)
expense_id = Column(Integer, ForeignKey("expenses.id", ondelete="CASCADE"), nullable=False)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
owed_amount = Column(Numeric(10, 2), nullable=False)
share_percentage = Column(Numeric(5, 2), nullable=True)
share_units = Column(Integer, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
# Relationships
expense = relationship("Expense", back_populates="splits")
user = relationship("User", foreign_keys=[user_id], back_populates="expense_splits")
settlement_activities = relationship("SettlementActivity", back_populates="split", cascade="all, delete-orphan")
# New fields for tracking payment status
status = Column(SAEnum(ExpenseSplitStatusEnum, name="expensesplitstatusenum", create_type=True), nullable=False, server_default=ExpenseSplitStatusEnum.unpaid.value, default=ExpenseSplitStatusEnum.unpaid)
paid_at = Column(DateTime(timezone=True), nullable=True) # Timestamp when the split was fully paid
class Settlement(Base):
__tablename__ = "settlements"
id = Column(Integer, primary_key=True, index=True)
group_id = Column(Integer, ForeignKey("groups.id"), nullable=False, index=True)
paid_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
paid_to_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
amount = Column(Numeric(10, 2), nullable=False)
settlement_date = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
description = Column(Text, nullable=True)
created_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
version = Column(Integer, nullable=False, default=1, server_default='1')
# Relationships
group = relationship("Group", foreign_keys=[group_id], back_populates="settlements")
payer = relationship("User", foreign_keys=[paid_by_user_id], back_populates="settlements_made")
payee = relationship("User", foreign_keys=[paid_to_user_id], back_populates="settlements_received")
created_by_user = relationship("User", foreign_keys=[created_by_user_id], back_populates="settlements_created")
__table_args__ = (
# Ensure payer and payee are different users
CheckConstraint('paid_by_user_id != paid_to_user_id', name='chk_settlement_different_users'),
)
# Potential future: PaymentMethod model, etc.
class SettlementActivity(Base):
__tablename__ = "settlement_activities"
id = Column(Integer, primary_key=True, index=True)
expense_split_id = Column(Integer, ForeignKey("expense_splits.id"), nullable=False, index=True)
paid_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True) # User who made this part of the payment
paid_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
amount_paid = Column(Numeric(10, 2), nullable=False)
created_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True) # User who recorded this activity
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
# --- Relationships ---
split = relationship("ExpenseSplit", back_populates="settlement_activities")
payer = relationship("User", foreign_keys=[paid_by_user_id], backref="made_settlement_activities")
creator = relationship("User", foreign_keys=[created_by_user_id], backref="created_settlement_activities")
__table_args__ = (
Index('ix_settlement_activity_expense_split_id', 'expense_split_id'),
Index('ix_settlement_activity_paid_by_user_id', 'paid_by_user_id'),
Index('ix_settlement_activity_created_by_user_id', 'created_by_user_id'),
)
# --- Chore Model ---
class Chore(Base):
__tablename__ = "chores"
id = Column(Integer, primary_key=True, index=True)
type = Column(SAEnum(ChoreTypeEnum, name="choretypeenum", create_type=True), nullable=False)
group_id = Column(Integer, ForeignKey("groups.id", ondelete="CASCADE"), nullable=True, index=True)
name = Column(String, nullable=False, index=True)
description = Column(Text, nullable=True)
created_by_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
frequency = Column(SAEnum(ChoreFrequencyEnum, name="chorefrequencyenum", create_type=True), nullable=False)
custom_interval_days = Column(Integer, nullable=True) # Only if frequency is 'custom'
next_due_date = Column(Date, nullable=False) # Changed to Date
last_completed_at = Column(DateTime(timezone=True), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
# --- Relationships ---
group = relationship("Group", back_populates="chores")
creator = relationship("User", back_populates="created_chores")
assignments = relationship("ChoreAssignment", back_populates="chore", cascade="all, delete-orphan")
# --- ChoreAssignment Model ---
class ChoreAssignment(Base):
__tablename__ = "chore_assignments"
id = Column(Integer, primary_key=True, index=True)
chore_id = Column(Integer, ForeignKey("chores.id", ondelete="CASCADE"), nullable=False, index=True)
assigned_to_user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
due_date = Column(Date, nullable=False) # Specific due date for this instance, changed to Date
is_complete = Column(Boolean, default=False, nullable=False)
completed_at = Column(DateTime(timezone=True), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
# --- Relationships ---
chore = relationship("Chore", back_populates="assignments")
assigned_user = relationship("User", back_populates="assigned_chores")
# === NEW: RecurrencePattern Model ===
class RecurrencePattern(Base):
__tablename__ = "recurrence_patterns"
id = Column(Integer, primary_key=True, index=True)
type = Column(SAEnum(RecurrenceTypeEnum, name="recurrencetypeenum", create_type=True), nullable=False)
interval = Column(Integer, default=1, nullable=False) # e.g., every 1 day, every 2 weeks
days_of_week = Column(String, nullable=True) # For weekly recurrences, e.g., "MON,TUE,FRI"
# day_of_month = Column(Integer, nullable=True) # For monthly on a specific day
# week_of_month = Column(Integer, nullable=True) # For monthly on a specific week (e.g., 2nd week)
# month_of_year = Column(Integer, nullable=True) # For yearly recurrences
end_date = Column(DateTime(timezone=True), nullable=True)
max_occurrences = Column(Integer, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
# Relationship back to Expenses that use this pattern (could be one-to-many if patterns are shared)
# However, the current CRUD implies one RecurrencePattern per Expense if recurring.
# If a pattern can be shared, this would be a one-to-many (RecurrencePattern to many Expenses).
# For now, assuming one-to-one as implied by current Expense.recurrence_pattern relationship setup.
expenses = relationship("Expense", back_populates="recurrence_pattern")
# === END: RecurrencePattern Model ===