src/services.py - develop - 2madeira/nami

src/services.py
203 строки6 KB
Andrey Zhunev
feat(api): improve API endpoints typing and validation
07 фев 2026, 23:13
07 фев 2026, 23:13bbd81c6
from re import L
from fastapi import HTTPException
from sqlalchemy import desc, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models import DayPart, FrequencyType, Habit, HabitLog, HabitRule, User
from src.utils import get_recent_dates


async def create_habit(
    session: AsyncSession, # открываем сессию для операций с бд
    name: str,
    user_id: int,
    frequency_type: FrequencyType,
    day_part: DayPart,
    description: str | None = None,
) -> Habit:
    """Create new habit"""
    # Create rule first
    new_habit_rule = HabitRule(
        frequency_type=frequency_type,
        day_part=day_part
    )

    # Cretate habit and connect with relationship
    new_habit = Habit(
        name=name, 
        description=description,
        user_id=user_id,
        rule=new_habit_rule # передаем объект, так лучше 
    )

    # Add only habit (rule will added with cascade automaticaly)
    # Используй relationship - бест практис.
    session.add(new_habit)
    await session.commit()
    await session.refresh(new_habit)

    return new_habit


async def get_habits(
    session: AsyncSession, 
    user_id: int
) -> list[Habit]:
    """Get habit list by user_id"""
    query = (
        select(Habit)
        .where(or_(Habit.user_id == user_id, Habit.user_id.is_(None)))
    )
    result = await session.execute(query)
    habit_list: list[Habit] = result.scalars().all()

    return habit_list


async def delete_habit(
    session: AsyncSession,
    habit_id: int
) -> None:
    """Delete habit"""
    habit = await session.get(Habit, habit_id)
    if not habit:
        raise HTTPException(status_code=404, detail="Habit not found.")

    await session.delete(habit)
    await session.commit()

    return None

async def update_habit(
    session: AsyncSession,
    habit_id: int,
    updates: dict  # только переданные в PATCH поля 
) -> Habit:
    """
    Allow update fields present only in whitelist:
    - habit.name
    - habit.description
    """
    # whitelist разрешенных полей
    ALLOWED_FIELDS = {'name', 'description'}

    # Проверка пустого запроса
    if not updates:
        raise HTTPException(
            status_code=400,
            detail="At least one field must be provided"
        )

    habit = await session.get(Habit, habit_id)
    if not habit: #почему просто не написать is None?
        raise HTTPException(status_code=404, detail="Habit not found")

    # Проверка сразу всех полей на разрешенность
    invalid_fields = set(updates.keys()) - ALLOWED_FIELDS
    if invalid_fields:
        raise HTTPException(
            status_code=400,
            detail=f"Cannot update fields: {', '.join(invalid_fields)}"
        )

    # Обновляем только разрешенные поля
    for field, value in updates.items():
        setattr(habit, field, value)

    await session.commit()
    await session.refresh(habit)
    return habit


async def check_in_habit(
    session: AsyncSession,
    habit_id: int, #  передаем id нужной привычки
) -> dict:
    """
    Check in habit complete. Check next points: 
    - Done or not today
    - Update streak (current and max)
    - Create new HabitLog
    """
    inspected_habit = await session.get(Habit, habit_id) # находим в бд нужную привычку

    if inspected_habit is None:
        raise HTTPException(status_code=404, detail="Habit not found")

    query = (
        select(HabitLog)
        .where(HabitLog.habit_id == habit_id)
        .order_by(desc(HabitLog.date))
    ) # в логах нужной привычки ищем последнюю запись

    result = await session.execute(query)
    last_log: HabitLog = result.scalars().first()

    dates = get_recent_dates()

    # Проверка выполнения сегодня
    if last_log and last_log.date == dates["today"]:
        raise HTTPException(status_code=400, detail="Already done today!")
    
    # Логика streak
    if last_log and last_log.date == dates["yesterday"]:
        inspected_habit.current_streak += 1
    else:
        inspected_habit.current_streak = 1

    # Нужно проверить max_streak
    if inspected_habit.current_streak > inspected_habit.max_streak:
        inspected_habit.max_streak = inspected_habit.current_streak

    # Создать новую запись в habitlog
    new_log = HabitLog(
        habit_id = inspected_habit.id,
        completed = True,
        date = dates["today"]
    )
    session.add(new_log)

    await session.commit()
    await session.refresh(new_log)

    return {
        "message": "Habit checked succesfully",
        "habit_id": inspected_habit.id,
        "current_streak": inspected_habit.current_streak,
        "max_streak" : inspected_habit.max_streak,
        "log_id": new_log.id,
        "date": new_log.date
    }


async def create_test_user(
    session: AsyncSession,
    telegram_id: int
) -> User:
    """
    Create test user logic.
    
    WARNING: Need to implement telegram_id unique value check.
    """
    user = User(telegram_id=telegram_id)
    session.add(user)

    await session.commit()
    await session.refresh(user)

    return user


async def delete_test_user(
    session: AsyncSession,
    user_id: int
) -> None:
    """Delete test user logic"""
    user = await session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found.")

    await session.delete(user)
    await session.commit()

    return None