42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any, Dict, List
|
||
|
|
|
||
|
|
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) -> List[Dict[str, Any]]:
|
||
|
|
offset = (page - 1) * size
|
||
|
|
with get_connection() as connection:
|
||
|
|
cursor = connection.execute(
|
||
|
|
"""
|
||
|
|
SELECT id, alarm_type, time, no, type, content, level
|
||
|
|
FROM alarm_event
|
||
|
|
ORDER BY id DESC
|
||
|
|
LIMIT ? OFFSET ?
|
||
|
|
""",
|
||
|
|
(size, offset),
|
||
|
|
)
|
||
|
|
return [dict(row) for row in cursor.fetchall()]
|