34 lines
906 B
Python
34 lines
906 B
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import sqlite3
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
from app.core.config import settings
|
||
|
|
|
||
|
|
|
||
|
|
def get_connection(db_path: Optional[Path] = None) -> sqlite3.Connection:
|
||
|
|
path = db_path or settings.alarm_db_path
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
connection = sqlite3.connect(path)
|
||
|
|
connection.row_factory = sqlite3.Row
|
||
|
|
return connection
|
||
|
|
|
||
|
|
|
||
|
|
def init_db() -> None:
|
||
|
|
with get_connection() as connection:
|
||
|
|
connection.execute(
|
||
|
|
"""
|
||
|
|
CREATE TABLE IF NOT EXISTS alarm_event (
|
||
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
|
|
alarm_type TEXT NOT NULL,
|
||
|
|
time TEXT NOT NULL,
|
||
|
|
no TEXT,
|
||
|
|
type TEXT,
|
||
|
|
content TEXT NOT NULL,
|
||
|
|
level TEXT NOT NULL
|
||
|
|
)
|
||
|
|
"""
|
||
|
|
)
|
||
|
|
connection.commit()
|