97 lines
2.6 KiB
Python
97 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import aiosqlite
|
|
import structlog
|
|
|
|
from backend.config import settings
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
created_at TEXT DEFAULT (datetime('now'))
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS notes (
|
|
id TEXT PRIMARY KEY,
|
|
user_id INTEGER NOT NULL,
|
|
title TEXT DEFAULT '',
|
|
path TEXT NOT NULL,
|
|
created_at TEXT DEFAULT (datetime('now')),
|
|
updated_at TEXT DEFAULT (datetime('now')),
|
|
deleted_at TEXT,
|
|
FOREIGN KEY(user_id) REFERENCES users(id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS tags (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS note_tags (
|
|
note_id TEXT NOT NULL,
|
|
tag_id INTEGER NOT NULL,
|
|
PRIMARY KEY (note_id, tag_id),
|
|
FOREIGN KEY(note_id) REFERENCES notes(id),
|
|
FOREIGN KEY(tag_id) REFERENCES tags(id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS note_templates (
|
|
note_id TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
PRIMARY KEY (note_id, name),
|
|
FOREIGN KEY(note_id) REFERENCES notes(id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS history (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
note_id TEXT NOT NULL,
|
|
user_id INTEGER NOT NULL,
|
|
action TEXT NOT NULL,
|
|
at TEXT DEFAULT (datetime('now')),
|
|
ip TEXT,
|
|
ua TEXT,
|
|
FOREIGN KEY(note_id) REFERENCES notes(id),
|
|
FOREIGN KEY(user_id) REFERENCES users(id)
|
|
);
|
|
|
|
CREATE VIRTUAL TABLE IF NOT EXISTS notes_fts USING fts5(
|
|
note_id UNINDEXED,
|
|
user_id UNINDEXED,
|
|
title,
|
|
body
|
|
);
|
|
"""
|
|
|
|
|
|
async def init_db() -> None:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
await db.executescript(SCHEMA)
|
|
await db.commit()
|
|
await seed_users()
|
|
logger.info("database_initialized", path=settings.database_path)
|
|
|
|
|
|
async def seed_users() -> None:
|
|
names = list(dict.fromkeys([*settings.user_list, settings.common_space]))
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
for name in names:
|
|
await db.execute("INSERT OR IGNORE INTO users (name) VALUES (?)", (name,))
|
|
await db.commit()
|
|
|
|
|
|
async def get_user_id(db: aiosqlite.Connection, name: str) -> int | None:
|
|
row = await db.execute("SELECT id FROM users WHERE name = ?", (name,))
|
|
row = await row.fetchone()
|
|
return row["id"] if row else None
|
|
|
|
|
|
async def list_users() -> list[dict]:
|
|
async with aiosqlite.connect(settings.database_path) as db:
|
|
db.row_factory = aiosqlite.Row
|
|
rows = await db.execute("SELECT id, name FROM users ORDER BY name")
|
|
return [dict(r) for r in await rows.fetchall()]
|