mitlist/be/app/schemas/expense.py
Mohamad.Elsena 5018ce02f7 feat: Implement recurring expenses feature with scheduling and management
- Added support for recurring expenses, allowing users to define recurrence patterns (daily, weekly, monthly, yearly) for expenses.
- Introduced `RecurrencePattern` model to manage recurrence details and linked it to the `Expense` model.
- Implemented background job scheduling using APScheduler to automatically generate new expenses based on defined patterns.
- Updated expense creation logic to handle recurring expenses, including validation and database interactions.
- Enhanced frontend components to allow users to create and manage recurring expenses through forms and lists.
- Updated documentation to reflect new features and usage guidelines for recurring expenses.
2025-05-22 16:37:14 +02:00

180 lines
7.4 KiB
Python

# app/schemas/expense.py
from pydantic import BaseModel, ConfigDict, validator, Field
from typing import List, Optional, Dict, Any
from decimal import Decimal
from datetime import datetime
# Assuming SplitTypeEnum is accessible here, e.g., from app.models or app.core.enums
# For now, let's redefine it or import it if models.py is parsable by Pydantic directly
# If it's from app.models, you might need to make app.models.SplitTypeEnum Pydantic-compatible or map it.
# For simplicity during schema definition, I'll redefine a string enum here.
# In a real setup, ensure this aligns with the SQLAlchemy enum in models.py.
from app.models import SplitTypeEnum, ExpenseSplitStatusEnum, ExpenseOverallStatusEnum # Try importing directly
from app.schemas.user import UserPublic # For user details in responses
from app.schemas.settlement_activity import SettlementActivityPublic # For settlement activities
# --- ExpenseSplit Schemas ---
class ExpenseSplitBase(BaseModel):
user_id: int
owed_amount: Decimal
share_percentage: Optional[Decimal] = None
share_units: Optional[int] = None
class ExpenseSplitCreate(ExpenseSplitBase):
pass # All fields from base are needed for creation
class ExpenseSplitPublic(ExpenseSplitBase):
id: int
expense_id: int
user: Optional[UserPublic] = None # If we want to nest user details
created_at: datetime
updated_at: datetime
status: ExpenseSplitStatusEnum # New field
paid_at: Optional[datetime] = None # New field
settlement_activities: List[SettlementActivityPublic] = [] # New field
model_config = ConfigDict(from_attributes=True)
# --- Expense Schemas ---
class RecurrencePatternBase(BaseModel):
type: str = Field(..., description="Type of recurrence: daily, weekly, monthly, yearly")
interval: int = Field(..., description="Interval of recurrence (e.g., every X days/weeks/months/years)")
days_of_week: Optional[List[int]] = Field(None, description="Days of week for weekly recurrence (0-6, Sunday-Saturday)")
end_date: Optional[datetime] = Field(None, description="Optional end date for the recurrence")
max_occurrences: Optional[int] = Field(None, description="Optional maximum number of occurrences")
class RecurrencePatternCreate(RecurrencePatternBase):
pass
class RecurrencePatternUpdate(RecurrencePatternBase):
pass
class RecurrencePatternInDB(RecurrencePatternBase):
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class ExpenseBase(BaseModel):
description: str
total_amount: Decimal
currency: Optional[str] = "USD"
expense_date: Optional[datetime] = None
split_type: SplitTypeEnum
list_id: Optional[int] = None
group_id: Optional[int] = None # Should be present if list_id is not, and vice-versa
item_id: Optional[int] = None
paid_by_user_id: int
is_recurring: bool = Field(False, description="Whether this is a recurring expense")
recurrence_pattern: Optional[RecurrencePatternCreate] = Field(None, description="Recurrence pattern for recurring expenses")
class ExpenseCreate(ExpenseBase):
# For EQUAL split, splits are generated. For others, they might be provided.
# This logic will be in the CRUD: if split_type is EXACT_AMOUNTS, PERCENTAGE, SHARES,
# then 'splits_in' should be provided.
splits_in: Optional[List[ExpenseSplitCreate]] = None
@validator('total_amount')
def total_amount_must_be_positive(cls, v):
if v <= Decimal('0'):
raise ValueError('Total amount must be positive')
return v
# Basic validation: if list_id is None, group_id must be provided.
# More complex cross-field validation might be needed.
@validator('group_id', always=True)
def check_list_or_group_id(cls, v, values):
if values.get('list_id') is None and v is None:
raise ValueError('Either list_id or group_id must be provided for an expense')
return v
@validator('recurrence_pattern')
def validate_recurrence_pattern(cls, v, values):
if values.get('is_recurring') and not v:
raise ValueError('Recurrence pattern is required for recurring expenses')
if not values.get('is_recurring') and v:
raise ValueError('Recurrence pattern should not be provided for non-recurring expenses')
return v
class ExpenseUpdate(BaseModel):
description: Optional[str] = None
total_amount: Optional[Decimal] = None
currency: Optional[str] = None
expense_date: Optional[datetime] = None
split_type: Optional[SplitTypeEnum] = None
list_id: Optional[int] = None
group_id: Optional[int] = None
item_id: Optional[int] = None
# paid_by_user_id is usually not updatable directly to maintain integrity.
# Updating splits would be a more complex operation, potentially a separate endpoint or careful logic.
version: int # For optimistic locking
is_recurring: Optional[bool] = None
recurrence_pattern: Optional[RecurrencePatternUpdate] = None
next_occurrence: Optional[datetime] = None
class ExpensePublic(ExpenseBase):
id: int
created_at: datetime
updated_at: datetime
version: int
created_by_user_id: int
splits: List[ExpenseSplitPublic] = []
paid_by_user: Optional[UserPublic] = None # If nesting user details
overall_settlement_status: ExpenseOverallStatusEnum # New field
# list: Optional[ListPublic] # If nesting list details
# group: Optional[GroupPublic] # If nesting group details
# item: Optional[ItemPublic] # If nesting item details
is_recurring: bool
next_occurrence: Optional[datetime]
last_occurrence: Optional[datetime]
recurrence_pattern: Optional[RecurrencePatternInDB]
parent_expense_id: Optional[int]
generated_expenses: List['ExpensePublic'] = []
model_config = ConfigDict(from_attributes=True)
# --- Settlement Schemas ---
class SettlementBase(BaseModel):
group_id: int
paid_by_user_id: int
paid_to_user_id: int
amount: Decimal
settlement_date: Optional[datetime] = None
description: Optional[str] = None
class SettlementCreate(SettlementBase):
@validator('amount')
def amount_must_be_positive(cls, v):
if v <= Decimal('0'):
raise ValueError('Settlement amount must be positive')
return v
@validator('paid_to_user_id')
def payer_and_payee_must_be_different(cls, v, values):
if 'paid_by_user_id' in values and v == values['paid_by_user_id']:
raise ValueError('Payer and payee cannot be the same user')
return v
class SettlementUpdate(BaseModel):
amount: Optional[Decimal] = None
settlement_date: Optional[datetime] = None
description: Optional[str] = None
# group_id, paid_by_user_id, paid_to_user_id are typically not updatable.
version: int # For optimistic locking
class SettlementPublic(SettlementBase):
id: int
created_at: datetime
updated_at: datetime
version: int
created_by_user_id: int
# payer: Optional[UserPublic] # If we want to include payer details
# payee: Optional[UserPublic] # If we want to include payee details
# group: Optional[GroupPublic] # If we want to include group details
model_config = ConfigDict(from_attributes=True)
# Placeholder for nested schemas (e.g., UserPublic) if needed
# from app.schemas.user import UserPublic
# from app.schemas.list import ListPublic
# from app.schemas.group import GroupPublic
# from app.schemas.item import ItemPublic