L'électron rare
f55093d6fe
ESP-IDF CI / Host Tests (Unity) (push) Successful in 1m8s
CI / firmware-native (push) Successful in 2m57s
Rust Protection Tests / Cargo test (host) (push) Failing after 3m21s
ESP-IDF CI / ESP-IDF Build (v5.4) (push) Failing after 6m55s
ESP-IDF CI / Memory Budget Gate (push) Has been skipped
qa-cicd-environments / qa-kxkm-s3-build (push) Successful in 8m53s
qa-cicd-environments / qa-sim-host (push) Successful in 2m2s
qa-cicd-environments / qa-kxkm-s3-memory-budget (push) Successful in 11m17s
Context: the project archive (KXKM_Batterie_Parallelator-main) had no git history locally; a fresh repository is needed to host it on git.saillant.cc (electron/KXKM_Batterie_Parallelator). Approach: initialize a new repo on branch main, stage the archive content, and harden .gitignore before the first commit. Changes: - Import the full project tree: firmware/, firmware-idf/, firmware-rs/, iosApp/, kxkm-bmu-app/, kxkm-api/, hardware/, docs/, specs/, scripts/, models/, tests/ - Keep project dotfiles tracked despite the trailing '.*' ignore rule: .github/, .claude/, .superpowers/, .gitattributes, .markdownlint.json - Extend .gitignore: firmware/src/credentials.h (local secrets, template kept), kxkm-bmu-app/**/build/ (66 MB compiled iOS framework), .remember/ (session data) Impact: the project can now be maintained on the self-hosted Gitea forge with a clean, secret-free initial history.
110 lines
2.8 KiB
Python
110 lines
2.8 KiB
Python
"""SQLite audit trail storage using aiosqlite."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import aiosqlite
|
|
|
|
from .config import settings
|
|
from .models import AuditEvent
|
|
|
|
_db: aiosqlite.Connection | None = None
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS audit_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
timestamp TEXT NOT NULL,
|
|
user_id TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
target INTEGER,
|
|
detail TEXT
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_events(timestamp);
|
|
CREATE INDEX IF NOT EXISTS idx_audit_user ON audit_events(user_id);
|
|
CREATE INDEX IF NOT EXISTS idx_audit_action ON audit_events(action);
|
|
"""
|
|
|
|
|
|
async def init_db() -> None:
|
|
"""Open SQLite connection and create schema."""
|
|
global _db
|
|
_db = await aiosqlite.connect(settings.sqlite_path)
|
|
_db.row_factory = aiosqlite.Row
|
|
await _db.executescript(_SCHEMA)
|
|
await _db.commit()
|
|
|
|
|
|
async def close_db() -> None:
|
|
"""Close SQLite connection."""
|
|
global _db
|
|
if _db:
|
|
await _db.close()
|
|
_db = None
|
|
|
|
|
|
async def insert_audit_events(events: list[AuditEvent]) -> int:
|
|
"""Insert batch of audit events. Returns count inserted."""
|
|
if not _db or not events:
|
|
return 0
|
|
rows = [
|
|
(e.timestamp, e.user_id, e.action, e.target, e.detail)
|
|
for e in events
|
|
]
|
|
await _db.executemany(
|
|
"INSERT INTO audit_events (timestamp, user_id, action, target, detail) "
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
rows,
|
|
)
|
|
await _db.commit()
|
|
return len(rows)
|
|
|
|
|
|
async def query_audit_events(
|
|
from_time: str | None = None,
|
|
to_time: str | None = None,
|
|
user: str | None = None,
|
|
action: str | None = None,
|
|
limit: int = 200,
|
|
) -> tuple[list[AuditEvent], int]:
|
|
"""Query audit events with optional filters. Returns (events, total_count)."""
|
|
if not _db:
|
|
return [], 0
|
|
|
|
conditions: list[str] = []
|
|
params: list[str | int] = []
|
|
|
|
if from_time:
|
|
conditions.append("timestamp >= ?")
|
|
params.append(from_time)
|
|
if to_time:
|
|
conditions.append("timestamp <= ?")
|
|
params.append(to_time)
|
|
if user:
|
|
conditions.append("user_id = ?")
|
|
params.append(user)
|
|
if action:
|
|
conditions.append("action = ?")
|
|
params.append(action)
|
|
|
|
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
|
|
|
# Total count
|
|
count_row = await _db.execute_fetchall(
|
|
f"SELECT COUNT(*) FROM audit_events {where}", params
|
|
)
|
|
total = count_row[0][0] if count_row else 0
|
|
|
|
# Fetch rows
|
|
rows = await _db.execute_fetchall(
|
|
f"SELECT timestamp, user_id, action, target, detail "
|
|
f"FROM audit_events {where} ORDER BY timestamp DESC LIMIT ?",
|
|
params + [limit],
|
|
)
|
|
|
|
events = [
|
|
AuditEvent(
|
|
timestamp=r[0], user_id=r[1], action=r[2], target=r[3], detail=r[4]
|
|
)
|
|
for r in rows
|
|
]
|
|
return events, total
|