mitlist/be/app/models.py
Mohamad 0cd1b9c84d feat: Implement soft delete functionality and remove cascading deletes
This commit introduces a significant update to the database schema and models by implementing soft delete functionality across multiple tables, including users, groups, lists, items, expenses, and more. Key changes include:

- Addition of `deleted_at` and `is_deleted` columns to facilitate soft deletes.
- Removal of cascading delete behavior from foreign key constraints to prevent accidental data loss.
- Updates to the models to incorporate the new soft delete mixin, ensuring consistent handling of deletions across the application.
- Introduction of a new endpoint for group deletion, requiring owner confirmation to enhance data integrity.

These changes aim to improve data safety and compliance with data protection regulations while maintaining the integrity of related records.
2025-06-25 20:16:28 +02:00

484 lines
24 KiB
Python

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,
func,
text as sa_text,
Text,
Numeric,
CheckConstraint,
Date
)
from sqlalchemy.orm import relationship, declared_attr
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.declarative import declared_attr
from .database import Base
class SoftDeleteMixin:
deleted_at = Column(DateTime(timezone=True), nullable=True, index=True)
is_deleted = Column(Boolean, default=False, nullable=False, index=True)
@declared_attr
def __mapper_args__(cls):
return {
'polymorphic_identity': cls.__name__.lower(),
'passive_deletes': True
}
# --- 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
# Consider renaming to a more generic term like 'DERIVED' or 'ENTITY_DRIVEN'
# if expenses might be derived from other entities in the future.
# 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"
class ChoreHistoryEventTypeEnum(str, enum.Enum):
CREATED = "created"
UPDATED = "updated"
DELETED = "deleted"
COMPLETED = "completed"
REOPENED = "reopened"
ASSIGNED = "assigned"
UNASSIGNED = "unassigned"
REASSIGNED = "reassigned"
SCHEDULE_GENERATED = "schedule_generated"
DUE_DATE_CHANGED = "due_date_changed"
DETAILS_CHANGED = "details_changed"
# --- User Model ---
class User(Base, SoftDeleteMixin):
__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)
is_guest = 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")
created_invites = relationship("Invite", back_populates="creator")
created_lists = relationship("List", foreign_keys="List.created_by_id", back_populates="creator")
added_items = relationship("Item", foreign_keys="Item.added_by_id", back_populates="added_by_user")
completed_items = relationship("Item", foreign_keys="Item.completed_by_id", back_populates="completed_by_user")
expenses_paid = relationship("Expense", foreign_keys="Expense.paid_by_user_id", back_populates="paid_by_user")
expenses_created = relationship("Expense", foreign_keys="Expense.created_by_user_id", back_populates="created_by_user")
expense_splits = relationship("ExpenseSplit", foreign_keys="ExpenseSplit.user_id", back_populates="user")
settlements_made = relationship("Settlement", foreign_keys="Settlement.paid_by_user_id", back_populates="payer")
settlements_received = relationship("Settlement", foreign_keys="Settlement.paid_to_user_id", back_populates="payee")
settlements_created = relationship("Settlement", foreign_keys="Settlement.created_by_user_id", back_populates="created_by_user")
created_chores = relationship("Chore", foreign_keys="[Chore.created_by_id]", back_populates="creator")
assigned_chores = relationship("ChoreAssignment", back_populates="assigned_user")
chore_history_entries = relationship("ChoreHistory", back_populates="changed_by_user")
assignment_history_entries = relationship("ChoreAssignmentHistory", back_populates="changed_by_user")
financial_audit_logs = relationship("FinancialAuditLog", back_populates="user")
time_entries = relationship("TimeEntry", back_populates="user")
categories = relationship("Category", back_populates="user")
class Group(Base, SoftDeleteMixin):
__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)
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')
creator = relationship("User", back_populates="created_groups")
member_associations = relationship("UserGroup", back_populates="group")
invites = relationship("Invite", back_populates="group")
lists = relationship("List", back_populates="group")
expenses = relationship("Expense", foreign_keys="Expense.group_id", back_populates="group")
settlements = relationship("Settlement", foreign_keys="Settlement.group_id", back_populates="group")
chores = relationship("Chore", back_populates="group")
chore_history = relationship("ChoreHistory", back_populates="group")
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"), nullable=False)
group_id = Column(Integer, ForeignKey("groups.id"), 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")
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"), 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")
class List(Base, SoftDeleteMixin):
__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)
group_id = Column(Integer, ForeignKey("groups.id"), nullable=True)
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')
archived_at = Column(DateTime(timezone=True), nullable=True, index=True)
creator = relationship("User", back_populates="created_lists")
group = relationship("Group", back_populates="lists")
items = relationship(
"Item",
back_populates="list",
order_by="Item.position.asc(), Item.created_at.asc()"
)
expenses = relationship("Expense", foreign_keys="Expense.list_id", back_populates="list")
class Item(Base, SoftDeleteMixin):
__tablename__ = "items"
__table_args__ = (
Index('ix_items_list_id_position', 'list_id', 'position'),
)
id = Column(Integer, primary_key=True, index=True)
list_id = Column(Integer, ForeignKey("lists.id"), nullable=False)
name = Column(String, index=True, nullable=False)
quantity = Column(String, nullable=True)
is_complete = Column(Boolean, default=False, nullable=False)
price = Column(Numeric(10, 2), nullable=True)
position = Column(Integer, nullable=False, server_default='0')
category_id = Column(Integer, ForeignKey('categories.id'), nullable=True)
added_by_id = Column(Integer, ForeignKey("users.id"), nullable=False)
completed_by_id = Column(Integer, ForeignKey("users.id"), 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)
version = Column(Integer, nullable=False, default=1, server_default='1')
# --- Relationships ---
list = relationship("List", back_populates="items")
added_by_user = relationship("User", foreign_keys=[added_by_id], back_populates="added_items")
completed_by_user = relationship("User", foreign_keys=[completed_by_id], back_populates="completed_items")
expenses = relationship("Expense", back_populates="item")
category = relationship("Category", back_populates="items")
class Expense(Base, SoftDeleteMixin):
__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)
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')
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")
parent_expense = relationship("Expense", remote_side=[id], back_populates="child_expenses")
child_expenses = relationship("Expense", back_populates="parent_expense")
overall_settlement_status = Column(SAEnum(ExpenseOverallStatusEnum, name="expenseoverallstatusenum", create_type=True), nullable=False, server_default=ExpenseOverallStatusEnum.unpaid.value, default=ExpenseOverallStatusEnum.unpaid)
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
parent_expense_id = Column(Integer, ForeignKey("expenses.id"), nullable=True)
last_occurrence = Column(DateTime(timezone=True), nullable=True)
__table_args__ = (
CheckConstraint('group_id IS NOT NULL OR list_id IS NOT NULL', name='ck_expense_group_or_list'),
)
class ExpenseSplit(Base, SoftDeleteMixin):
__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"), 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)
expense = relationship("Expense", back_populates="splits")
user = relationship("User", foreign_keys=[user_id], back_populates="expense_splits")
settlement_activities = relationship("SettlementActivity", back_populates="split")
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)
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')
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__ = (
CheckConstraint('paid_by_user_id != paid_to_user_id', name='chk_settlement_different_users'),
)
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)
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)
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)
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, SoftDeleteMixin):
__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"), 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)
parent_chore_id = Column(Integer, ForeignKey('chores.id'), nullable=True, index=True)
frequency = Column(SAEnum(ChoreFrequencyEnum, name="chorefrequencyenum", create_type=True), nullable=False)
custom_interval_days = Column(Integer, nullable=True)
next_due_date = Column(Date, nullable=False)
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)
group = relationship("Group", back_populates="chores")
creator = relationship("User", back_populates="created_chores")
assignments = relationship("ChoreAssignment", back_populates="chore")
history = relationship("ChoreHistory", back_populates="chore")
parent_chore = relationship("Chore", remote_side=[id], back_populates="child_chores")
child_chores = relationship("Chore", back_populates="parent_chore")
# --- ChoreAssignment Model ---
class ChoreAssignment(Base, SoftDeleteMixin):
__tablename__ = "chore_assignments"
id = Column(Integer, primary_key=True, index=True)
chore_id = Column(Integer, ForeignKey("chores.id"), nullable=False, index=True)
assigned_to_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
due_date = Column(Date, nullable=False)
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)
chore = relationship("Chore", back_populates="assignments")
assigned_user = relationship("User", back_populates="assigned_chores")
history = relationship("ChoreAssignmentHistory", back_populates="assignment")
time_entries = relationship("TimeEntry", back_populates="assignment")
# === NEW: RecurrencePattern Model ===
class RecurrencePattern(Base, SoftDeleteMixin):
__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)
days_of_week = Column(String, nullable=True)
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)
expenses = relationship("Expense", back_populates="recurrence_pattern")
# === END: RecurrencePattern Model ===
# === NEW: Chore History Models ===
class ChoreHistory(Base):
__tablename__ = "chore_history"
id = Column(Integer, primary_key=True, index=True)
chore_id = Column(Integer, ForeignKey("chores.id"), nullable=True, index=True)
group_id = Column(Integer, ForeignKey("groups.id"), nullable=True, index=True)
event_type = Column(SAEnum(ChoreHistoryEventTypeEnum, name="chorehistoryeventtypeenum", create_type=True), nullable=False)
event_data = Column(JSONB, nullable=True)
changed_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
timestamp = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
chore = relationship("Chore", back_populates="history")
group = relationship("Group", back_populates="chore_history")
changed_by_user = relationship("User", back_populates="chore_history_entries")
class ChoreAssignmentHistory(Base):
__tablename__ = "chore_assignment_history"
id = Column(Integer, primary_key=True, index=True)
assignment_id = Column(Integer, ForeignKey("chore_assignments.id"), nullable=False, index=True)
event_type = Column(SAEnum(ChoreHistoryEventTypeEnum, name="chorehistoryeventtypeenum", create_type=True), nullable=False)
event_data = Column(JSONB, nullable=True)
changed_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
timestamp = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
assignment = relationship("ChoreAssignment", back_populates="history")
changed_by_user = relationship("User", back_populates="assignment_history_entries")
# --- New Models from Roadmap ---
class FinancialAuditLog(Base):
__tablename__ = 'financial_audit_log'
id = Column(Integer, primary_key=True, index=True)
timestamp = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
user_id = Column(Integer, ForeignKey('users.id'), nullable=True)
action_type = Column(String, nullable=False, index=True)
entity_type = Column(String, nullable=False)
entity_id = Column(Integer, nullable=False)
details = Column(JSONB, nullable=True)
user = relationship("User", back_populates="financial_audit_logs")
class Category(Base, SoftDeleteMixin):
__tablename__ = 'categories'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=False, index=True)
user_id = Column(Integer, ForeignKey('users.id'), nullable=True)
group_id = Column(Integer, ForeignKey('groups.id'), nullable=True)
user = relationship("User", back_populates="categories")
items = relationship("Item", back_populates="category")
__table_args__ = (UniqueConstraint('name', 'user_id', 'group_id', name='uq_category_scope'),)
class TimeEntry(Base, SoftDeleteMixin):
__tablename__ = 'time_entries'
id = Column(Integer, primary_key=True, index=True)
chore_assignment_id = Column(Integer, ForeignKey('chore_assignments.id'), nullable=False)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
start_time = Column(DateTime(timezone=True), nullable=False)
end_time = Column(DateTime(timezone=True), nullable=True)
duration_seconds = Column(Integer, nullable=True)
assignment = relationship("ChoreAssignment", back_populates="time_entries")
user = relationship("User", back_populates="time_entries")