81577ac7e8
- 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.
234 lines
8.7 KiB
Python
234 lines
8.7 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.future import select
|
|
from sqlalchemy.orm import selectinload
|
|
from typing import List, Optional
|
|
import logging
|
|
from datetime import date
|
|
|
|
from app.models import Chore, Group, User, ChoreFrequencyEnum, ChoreTypeEnum
|
|
from app.schemas.chore import ChoreCreate, ChoreUpdate
|
|
from app.core.chore_utils import calculate_next_due_date
|
|
from app.crud.group import get_group_by_id, is_user_member
|
|
from app.core.exceptions import ChoreNotFoundError, GroupNotFoundError, PermissionDeniedError, DatabaseIntegrityError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
async def create_chore(
|
|
db: AsyncSession,
|
|
chore_in: ChoreCreate,
|
|
user_id: int,
|
|
group_id: Optional[int] = None
|
|
) -> Chore:
|
|
"""Creates a new chore, either personal or within a specific group."""
|
|
if chore_in.type == ChoreTypeEnum.group:
|
|
if not group_id:
|
|
raise ValueError("group_id is required for group chores")
|
|
# Validate group existence and user membership
|
|
group = await get_group_by_id(db, group_id)
|
|
if not group:
|
|
raise GroupNotFoundError(group_id)
|
|
if not await is_user_member(db, group_id, user_id):
|
|
raise PermissionDeniedError(detail=f"User {user_id} not a member of group {group_id}")
|
|
else: # personal chore
|
|
if group_id:
|
|
raise ValueError("group_id must be None for personal chores")
|
|
|
|
db_chore = Chore(
|
|
**chore_in.model_dump(exclude_unset=True, exclude={'group_id'}),
|
|
group_id=group_id,
|
|
created_by_id=user_id,
|
|
)
|
|
|
|
# Specific check for custom frequency
|
|
if chore_in.frequency == ChoreFrequencyEnum.custom and chore_in.custom_interval_days is None:
|
|
raise ValueError("custom_interval_days must be set for custom frequency chores.")
|
|
|
|
db.add(db_chore)
|
|
try:
|
|
await db.commit()
|
|
await db.refresh(db_chore)
|
|
result = await db.execute(
|
|
select(Chore)
|
|
.where(Chore.id == db_chore.id)
|
|
.options(selectinload(Chore.creator), selectinload(Chore.group))
|
|
)
|
|
return result.scalar_one()
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error(f"Error creating chore: {e}", exc_info=True)
|
|
raise DatabaseIntegrityError(f"Could not create chore. Error: {str(e)}")
|
|
|
|
async def get_chore_by_id(
|
|
db: AsyncSession,
|
|
chore_id: int,
|
|
) -> Optional[Chore]:
|
|
"""Gets a chore by its ID with creator and group info."""
|
|
result = await db.execute(
|
|
select(Chore)
|
|
.where(Chore.id == chore_id)
|
|
.options(selectinload(Chore.creator), selectinload(Chore.group))
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_chore_by_id_and_group(
|
|
db: AsyncSession,
|
|
chore_id: int,
|
|
group_id: int,
|
|
user_id: int
|
|
) -> Optional[Chore]:
|
|
"""Gets a specific group chore by ID, ensuring it belongs to the group and user is a member."""
|
|
if not await is_user_member(db, group_id, user_id):
|
|
raise PermissionDeniedError(detail=f"User {user_id} not a member of group {group_id}")
|
|
|
|
chore = await get_chore_by_id(db, chore_id)
|
|
if chore and chore.group_id == group_id and chore.type == ChoreTypeEnum.group:
|
|
return chore
|
|
return None
|
|
|
|
async def get_personal_chores(
|
|
db: AsyncSession,
|
|
user_id: int
|
|
) -> List[Chore]:
|
|
"""Gets all personal chores for a user."""
|
|
result = await db.execute(
|
|
select(Chore)
|
|
.where(
|
|
Chore.created_by_id == user_id,
|
|
Chore.type == ChoreTypeEnum.personal
|
|
)
|
|
.options(selectinload(Chore.creator), selectinload(Chore.assignments))
|
|
.order_by(Chore.next_due_date, Chore.name)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def get_chores_by_group_id(
|
|
db: AsyncSession,
|
|
group_id: int,
|
|
user_id: int
|
|
) -> List[Chore]:
|
|
"""Gets all chores for a specific group, if the user is a member."""
|
|
if not await is_user_member(db, group_id, user_id):
|
|
raise PermissionDeniedError(detail=f"User {user_id} not a member of group {group_id}")
|
|
|
|
result = await db.execute(
|
|
select(Chore)
|
|
.where(
|
|
Chore.group_id == group_id,
|
|
Chore.type == ChoreTypeEnum.group
|
|
)
|
|
.options(selectinload(Chore.creator), selectinload(Chore.assignments))
|
|
.order_by(Chore.next_due_date, Chore.name)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def update_chore(
|
|
db: AsyncSession,
|
|
chore_id: int,
|
|
chore_in: ChoreUpdate,
|
|
user_id: int,
|
|
group_id: Optional[int] = None
|
|
) -> Optional[Chore]:
|
|
"""Updates a chore's details."""
|
|
db_chore = await get_chore_by_id(db, chore_id)
|
|
if not db_chore:
|
|
raise ChoreNotFoundError(chore_id, group_id)
|
|
|
|
# Check permissions
|
|
if db_chore.type == ChoreTypeEnum.group:
|
|
if not group_id:
|
|
raise ValueError("group_id is required for group chores")
|
|
if not await is_user_member(db, group_id, user_id):
|
|
raise PermissionDeniedError(detail=f"User {user_id} not a member of group {group_id}")
|
|
if db_chore.group_id != group_id:
|
|
raise ChoreNotFoundError(chore_id, group_id)
|
|
else: # personal chore
|
|
if group_id:
|
|
raise ValueError("group_id must be None for personal chores")
|
|
if db_chore.created_by_id != user_id:
|
|
raise PermissionDeniedError(detail="Only the creator can update personal chores")
|
|
|
|
update_data = chore_in.model_dump(exclude_unset=True)
|
|
|
|
# Handle type change
|
|
if 'type' in update_data:
|
|
new_type = update_data['type']
|
|
if new_type == ChoreTypeEnum.group and not group_id:
|
|
raise ValueError("group_id is required for group chores")
|
|
if new_type == ChoreTypeEnum.personal and group_id:
|
|
raise ValueError("group_id must be None for personal chores")
|
|
|
|
# Recalculate next_due_date if needed
|
|
recalculate = False
|
|
if 'frequency' in update_data and update_data['frequency'] != db_chore.frequency:
|
|
recalculate = True
|
|
if 'custom_interval_days' in update_data and update_data['custom_interval_days'] != db_chore.custom_interval_days:
|
|
recalculate = True
|
|
|
|
current_next_due_date_for_calc = db_chore.next_due_date
|
|
if 'next_due_date' in update_data:
|
|
current_next_due_date_for_calc = update_data['next_due_date']
|
|
if not ('frequency' in update_data or 'custom_interval_days' in update_data):
|
|
recalculate = False
|
|
|
|
for field, value in update_data.items():
|
|
setattr(db_chore, field, value)
|
|
|
|
if recalculate:
|
|
db_chore.next_due_date = calculate_next_due_date(
|
|
current_due_date=current_next_due_date_for_calc,
|
|
frequency=db_chore.frequency,
|
|
custom_interval_days=db_chore.custom_interval_days,
|
|
last_completed_date=db_chore.last_completed_at
|
|
)
|
|
|
|
if db_chore.frequency == ChoreFrequencyEnum.custom and db_chore.custom_interval_days is None:
|
|
raise ValueError("custom_interval_days must be set for custom frequency chores.")
|
|
|
|
try:
|
|
await db.commit()
|
|
await db.refresh(db_chore)
|
|
result = await db.execute(
|
|
select(Chore)
|
|
.where(Chore.id == db_chore.id)
|
|
.options(selectinload(Chore.creator), selectinload(Chore.group))
|
|
)
|
|
return result.scalar_one()
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error(f"Error updating chore {chore_id}: {e}", exc_info=True)
|
|
raise DatabaseIntegrityError(f"Could not update chore {chore_id}. Error: {str(e)}")
|
|
|
|
async def delete_chore(
|
|
db: AsyncSession,
|
|
chore_id: int,
|
|
user_id: int,
|
|
group_id: Optional[int] = None
|
|
) -> bool:
|
|
"""Deletes a chore and its assignments, ensuring user has permission."""
|
|
db_chore = await get_chore_by_id(db, chore_id)
|
|
if not db_chore:
|
|
raise ChoreNotFoundError(chore_id, group_id)
|
|
|
|
# Check permissions
|
|
if db_chore.type == ChoreTypeEnum.group:
|
|
if not group_id:
|
|
raise ValueError("group_id is required for group chores")
|
|
if not await is_user_member(db, group_id, user_id):
|
|
raise PermissionDeniedError(detail=f"User {user_id} not a member of group {group_id}")
|
|
if db_chore.group_id != group_id:
|
|
raise ChoreNotFoundError(chore_id, group_id)
|
|
else: # personal chore
|
|
if group_id:
|
|
raise ValueError("group_id must be None for personal chores")
|
|
if db_chore.created_by_id != user_id:
|
|
raise PermissionDeniedError(detail="Only the creator can delete personal chores")
|
|
|
|
await db.delete(db_chore)
|
|
try:
|
|
await db.commit()
|
|
return True
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error(f"Error deleting chore {chore_id}: {e}", exc_info=True)
|
|
raise DatabaseIntegrityError(f"Could not delete chore {chore_id}. Error: {str(e)}")
|