72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
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 (event_time, line_code, event_type, event_code, event_value, content)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
alarm.event_time.isoformat(sep=" ", timespec="seconds"),
|
|
alarm.line_code,
|
|
alarm.event_type,
|
|
alarm.event_code,
|
|
alarm.event_value,
|
|
alarm.content,
|
|
),
|
|
)
|
|
connection.commit()
|
|
return int(cursor.lastrowid)
|
|
|
|
def list_alarms(
|
|
self,
|
|
page: int = 1,
|
|
size: int = 20,
|
|
line_code: Optional[int] = None,
|
|
event_type: Optional[int] = None,
|
|
start_time: Optional[datetime] = None,
|
|
end_time: Optional[datetime] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
offset = (page - 1) * size
|
|
conditions = []
|
|
params: List[Any] = []
|
|
|
|
if line_code is not None:
|
|
conditions.append("line_code = ?")
|
|
params.append(line_code)
|
|
if event_type is not None:
|
|
conditions.append("event_type = ?")
|
|
params.append(event_type)
|
|
if start_time is not None:
|
|
conditions.append("event_time >= ?")
|
|
params.append(start_time.isoformat(sep=" ", timespec="seconds"))
|
|
if end_time is not None:
|
|
conditions.append("event_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, event_time, line_code, event_type, event_code, event_value, content
|
|
FROM alarm_event
|
|
{where_clause}
|
|
ORDER BY id DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
tuple(params + [size, offset]),
|
|
)
|
|
return [dict(row) for row in cursor.fetchall()]
|