emcp/backend/app/repositories/alarm_repo.py

from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional

from app.db.sqlite import get_connection
from app.schemas.platform import AlarmEvent


class AlarmRepository:
    def save_alarm(self, alarm: AlarmEvent) -> int:
        with get_connection() as connection:
            cursor = connection.execute(
                """
                INSERT INTO alarm_event (alarm_type, time, no, type, content, level)
                VALUES (?, ?, ?, ?, ?, ?)
                """,
                (
                    alarm.alarm_type,
                    alarm.time.isoformat(sep=" ", timespec="seconds"),
                    alarm.no,
                    alarm.type,
                    alarm.content,
                    alarm.level,
                ),
            )
            connection.commit()
            return int(cursor.lastrowid)

    def list_alarms(
        self,
        page: int = 1,
        size: int = 20,
        no: str = "",
        alarm_type: str = "",
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None,
    ) -> List[Dict[str, Any]]:
        offset = (page - 1) * size
        conditions = []
        params: List[Any] = []

        if no:
            conditions.append("no = ?")
            params.append(no)
        if alarm_type:
            conditions.append("type = ?")
            params.append(alarm_type)
        if start_time is not None:
            conditions.append("time >= ?")
            params.append(start_time.isoformat(sep=" ", timespec="seconds"))
        if end_time is not None:
            conditions.append("time <= ?")
            params.append(end_time.isoformat(sep=" ", timespec="seconds"))

        where_clause = ""
        if conditions:
            where_clause = "WHERE " + " AND ".join(conditions)

        with get_connection() as connection:
            cursor = connection.execute(
                f"""
                SELECT id, alarm_type, time, no, type, content, level
                FROM alarm_event
                {where_clause}
                ORDER BY id DESC
                LIMIT ? OFFSET ?
                """,
                tuple(params + [size, offset]),
            )
            return [dict(row) for row in cursor.fetchall()]