shinalok
55918d56d1
Completed and abandoned habits were both lumped under a single terminal state, making it impossible to tell "finished successfully" apart from "gave up" when reviewing habit history.
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.habit import Habit, HabitStatus, HabitType
|
|
from app.schemas.habit import HabitCreate, HabitUpdate
|
|
|
|
|
|
def list_habits(
|
|
db: Session, user_id: int, habit_type: HabitType | None = None, status: HabitStatus | None = None
|
|
) -> list[Habit]:
|
|
stmt = select(Habit).where(Habit.user_id == user_id)
|
|
if habit_type is not None:
|
|
stmt = stmt.where(Habit.habit_type == habit_type)
|
|
if status is not None:
|
|
stmt = stmt.where(Habit.status == status)
|
|
stmt = stmt.order_by(Habit.sort_order.is_(None), Habit.sort_order, Habit.created_at)
|
|
return list(db.scalars(stmt))
|
|
|
|
|
|
def list_active_habits_with_reminders(db: Session) -> list[Habit]:
|
|
"""스케줄러 전용: 유저 스코핑 없이 알림 시각이 설정된 전체 active 습관을 반환한다."""
|
|
stmt = select(Habit).where(Habit.status == HabitStatus.ACTIVE, Habit.reminder_time.isnot(None))
|
|
return list(db.scalars(stmt))
|
|
|
|
|
|
def list_active_user_ids(db: Session) -> list[int]:
|
|
"""스케줄러 전용: 유저 스코핑 없이, active 습관을 하나 이상 가진 유저 id 목록을 반환한다(주간/월간 요약 알림 대상)."""
|
|
stmt = (
|
|
select(Habit.user_id)
|
|
.where(Habit.status == HabitStatus.ACTIVE, Habit.user_id.isnot(None))
|
|
.distinct()
|
|
)
|
|
return list(db.scalars(stmt))
|
|
|
|
|
|
def get_habit(db: Session, habit_id: int, user_id: int) -> Habit | None:
|
|
return db.scalar(select(Habit).where(Habit.id == habit_id, Habit.user_id == user_id))
|
|
|
|
|
|
def create_habit(db: Session, user_id: int, data: HabitCreate) -> Habit:
|
|
habit = Habit(
|
|
user_id=user_id,
|
|
name=data.name,
|
|
habit_type=data.habit_type,
|
|
weekdays_mask=data.weekdays_mask,
|
|
condition_text=data.condition_text,
|
|
reminder_time=data.reminder_time,
|
|
status=HabitStatus.ACTIVE,
|
|
)
|
|
db.add(habit)
|
|
db.commit()
|
|
db.refresh(habit)
|
|
return habit
|
|
|
|
|
|
def update_habit(db: Session, habit: Habit, data: HabitUpdate) -> Habit:
|
|
habit.name = data.name
|
|
habit.habit_type = data.habit_type
|
|
habit.weekdays_mask = data.weekdays_mask
|
|
habit.condition_text = data.condition_text
|
|
habit.reminder_time = data.reminder_time
|
|
db.commit()
|
|
db.refresh(habit)
|
|
return habit
|
|
|
|
|
|
def delete_habit(db: Session, habit: Habit) -> None:
|
|
db.delete(habit)
|
|
db.commit()
|
|
|
|
|
|
def complete_habit(db: Session, habit: Habit) -> Habit:
|
|
habit.status = HabitStatus.COMPLETED
|
|
habit.completed_at = datetime.now()
|
|
db.commit()
|
|
db.refresh(habit)
|
|
return habit
|
|
|
|
|
|
def abandon_habit(db: Session, habit: Habit) -> Habit:
|
|
habit.status = HabitStatus.ABANDONED
|
|
habit.abandoned_at = datetime.now()
|
|
db.commit()
|
|
db.refresh(habit)
|
|
return habit
|
|
|
|
|
|
def reactivate_habit(db: Session, habit: Habit) -> Habit:
|
|
habit.status = HabitStatus.ACTIVE
|
|
habit.completed_at = None
|
|
habit.abandoned_at = None
|
|
db.commit()
|
|
db.refresh(habit)
|
|
return habit
|
|
|
|
|
|
def reorder_habits(db: Session, user_id: int, ordered_ids: list[int]) -> None:
|
|
"""ordered_ids에 나온 순서대로 sort_order를 다시 매긴다. 목록에 없는 id나 다른 유저의 habit은 무시한다."""
|
|
habits = db.scalars(
|
|
select(Habit).where(Habit.id.in_(ordered_ids), Habit.user_id == user_id)
|
|
).all()
|
|
habit_map = {h.id: h for h in habits}
|
|
for index, habit_id in enumerate(ordered_ids):
|
|
habit = habit_map.get(habit_id)
|
|
if habit is not None:
|
|
habit.sort_order = index
|
|
db.commit()
|