v0.2.1 add Logging
This commit is contained in:
Björn Nehlsen
@@ -0,0 +1,237 @@
|
||||
import json
|
||||
import sqlite3
|
||||
import threading
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
_ALLOWED_LEVELS = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
|
||||
|
||||
|
||||
class LogStore:
|
||||
def __init__(self, db_path: Path, max_rows: int = 50000) -> None:
|
||||
self.db_path = db_path
|
||||
self.max_rows = max(1000, int(max_rows))
|
||||
self._lock = threading.Lock()
|
||||
self._writes_since_prune = 0
|
||||
|
||||
def _connect(self) -> sqlite3.Connection:
|
||||
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
def ensure_schema(self) -> None:
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS logs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
created_at TEXT NOT NULL DEFAULT (STRFTIME('%Y-%m-%d %H:%M:%f', 'now', 'localtime')),
|
||||
level TEXT NOT NULL,
|
||||
category TEXT NOT NULL,
|
||||
event TEXT NOT NULL,
|
||||
message TEXT NOT NULL,
|
||||
username TEXT,
|
||||
backup_id INTEGER,
|
||||
request_path TEXT,
|
||||
request_method TEXT,
|
||||
status_code INTEGER,
|
||||
details_json TEXT
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_created_at ON logs(created_at DESC)")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_level ON logs(level)")
|
||||
conn.execute("CREATE INDEX IF NOT EXISTS idx_logs_category ON logs(category)")
|
||||
conn.commit()
|
||||
|
||||
def write(
|
||||
self,
|
||||
*,
|
||||
level: str,
|
||||
category: str,
|
||||
event: str,
|
||||
message: str,
|
||||
username: str | None = None,
|
||||
backup_id: int | None = None,
|
||||
request_path: str | None = None,
|
||||
request_method: str | None = None,
|
||||
status_code: int | None = None,
|
||||
details: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
lvl = (level or "INFO").upper()
|
||||
if lvl not in _ALLOWED_LEVELS:
|
||||
lvl = "INFO"
|
||||
|
||||
safe_category = (category or "app").strip()[:50] or "app"
|
||||
safe_event = (event or "event").strip()[:80] or "event"
|
||||
safe_message = (message or "").strip()[:3000] or "-"
|
||||
safe_username = (username or "").strip()[:120] or None
|
||||
safe_path = (request_path or "").strip()[:500] or None
|
||||
safe_method = (request_method or "").strip().upper()[:16] or None
|
||||
details_json = json.dumps(details or {}, ensure_ascii=False)
|
||||
|
||||
with self._lock:
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO logs (
|
||||
level,
|
||||
category,
|
||||
event,
|
||||
message,
|
||||
username,
|
||||
backup_id,
|
||||
request_path,
|
||||
request_method,
|
||||
status_code,
|
||||
details_json
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(
|
||||
lvl,
|
||||
safe_category,
|
||||
safe_event,
|
||||
safe_message,
|
||||
safe_username,
|
||||
backup_id,
|
||||
safe_path,
|
||||
safe_method,
|
||||
status_code,
|
||||
details_json,
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
self._writes_since_prune += 1
|
||||
if self._writes_since_prune >= 100:
|
||||
self._prune_if_needed()
|
||||
self._writes_since_prune = 0
|
||||
|
||||
def _prune_if_needed(self) -> None:
|
||||
with self._connect() as conn:
|
||||
row = conn.execute("SELECT COUNT(*) AS cnt FROM logs").fetchone()
|
||||
count = int(row["cnt"] if row is not None else 0)
|
||||
if count <= self.max_rows:
|
||||
return
|
||||
|
||||
to_delete = count - self.max_rows
|
||||
conn.execute(
|
||||
"""
|
||||
DELETE FROM logs
|
||||
WHERE id IN (
|
||||
SELECT id FROM logs
|
||||
ORDER BY created_at ASC, id ASC
|
||||
LIMIT ?
|
||||
)
|
||||
""",
|
||||
(to_delete,),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
def query_logs(
|
||||
self,
|
||||
*,
|
||||
level: str | None = None,
|
||||
category: str | None = None,
|
||||
search: str | None = None,
|
||||
limit: int = 200,
|
||||
) -> list[dict[str, Any]]:
|
||||
safe_limit = min(max(int(limit), 1), 1000)
|
||||
|
||||
query = (
|
||||
"SELECT id, created_at, level, category, event, message, username, "
|
||||
"backup_id, request_path, request_method, status_code, details_json "
|
||||
"FROM logs WHERE 1=1"
|
||||
)
|
||||
params: list[Any] = []
|
||||
|
||||
if level:
|
||||
query += " AND level = ?"
|
||||
params.append(level.upper())
|
||||
if category:
|
||||
query += " AND category = ?"
|
||||
params.append(category)
|
||||
if search:
|
||||
query += " AND (message LIKE ? OR event LIKE ? OR username LIKE ? OR details_json LIKE ?)"
|
||||
like = f"%{search.strip()}%"
|
||||
params.extend([like, like, like, like])
|
||||
|
||||
query += " ORDER BY created_at DESC, id DESC LIMIT ?"
|
||||
params.append(safe_limit)
|
||||
|
||||
with self._connect() as conn:
|
||||
rows = conn.execute(query, params).fetchall()
|
||||
|
||||
result: list[dict[str, Any]] = []
|
||||
for row in rows:
|
||||
details_raw = row["details_json"] or "{}"
|
||||
try:
|
||||
details = json.loads(details_raw)
|
||||
except json.JSONDecodeError:
|
||||
details = {"raw": details_raw}
|
||||
result.append(
|
||||
{
|
||||
"id": row["id"],
|
||||
"created_at": row["created_at"],
|
||||
"level": row["level"],
|
||||
"category": row["category"],
|
||||
"event": row["event"],
|
||||
"message": row["message"],
|
||||
"username": row["username"],
|
||||
"backup_id": row["backup_id"],
|
||||
"request_path": row["request_path"],
|
||||
"request_method": row["request_method"],
|
||||
"status_code": row["status_code"],
|
||||
"details": details,
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
def list_categories(self) -> list[str]:
|
||||
with self._connect() as conn:
|
||||
rows = conn.execute("SELECT DISTINCT category FROM logs ORDER BY category ASC").fetchall()
|
||||
return [str(row["category"]) for row in rows if row["category"]]
|
||||
|
||||
def count_by_level_last_hours(self, hours: int = 24) -> dict[str, int]:
|
||||
safe_hours = min(max(int(hours), 1), 24 * 30)
|
||||
with self._connect() as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT level, COUNT(*) AS cnt
|
||||
FROM logs
|
||||
WHERE created_at >= DATETIME('now', 'localtime', ?)
|
||||
GROUP BY level
|
||||
""",
|
||||
(f"-{safe_hours} hours",),
|
||||
).fetchall()
|
||||
|
||||
result = {"DEBUG": 0, "INFO": 0, "WARNING": 0, "ERROR": 0, "CRITICAL": 0}
|
||||
for row in rows:
|
||||
level = str(row["level"])
|
||||
if level in result:
|
||||
result[level] = int(row["cnt"])
|
||||
return result
|
||||
|
||||
def clear(self) -> int:
|
||||
with self._connect() as conn:
|
||||
row = conn.execute("SELECT COUNT(*) AS cnt FROM logs").fetchone()
|
||||
count = int(row["cnt"] if row is not None else 0)
|
||||
conn.execute("DELETE FROM logs")
|
||||
conn.commit()
|
||||
return count
|
||||
|
||||
@staticmethod
|
||||
def compact_details(details: dict[str, Any], max_len: int = 800) -> str:
|
||||
if not details:
|
||||
return ""
|
||||
raw = json.dumps(details, ensure_ascii=False, separators=(",", ":"))
|
||||
if len(raw) <= max_len:
|
||||
return raw
|
||||
return f"{raw[:max_len]}..."
|
||||
|
||||
@staticmethod
|
||||
def now_iso() -> str:
|
||||
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
Reference in New Issue
Block a user