134 lines
4.0 KiB
Python
134 lines
4.0 KiB
Python
import sqlite3
|
|
import os
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent.parent
|
|
DB_PATH = Path(os.getenv('LINKLOG_DATABASE_PATH', BASE_DIR / 'data' / 'linklog.db'))
|
|
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
AVATARS_DIR = DB_PATH.parent / 'avatars'
|
|
AVATARS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return sha256(password.encode('utf-8')).hexdigest()
|
|
|
|
MIGRATIONS = [
|
|
(1, '''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id TEXT PRIMARY KEY,
|
|
username TEXT NOT NULL UNIQUE,
|
|
email TEXT NOT NULL UNIQUE,
|
|
password_hash TEXT NOT NULL,
|
|
avatar_url TEXT,
|
|
bio TEXT,
|
|
is_admin INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS tokens (
|
|
id TEXT PRIMARY KEY,
|
|
user_id TEXT NOT NULL,
|
|
token_hash TEXT NOT NULL UNIQUE,
|
|
token_type TEXT NOT NULL DEFAULT 'access',
|
|
expires_at TEXT NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
revoked INTEGER NOT NULL DEFAULT 0,
|
|
FOREIGN KEY(user_id) REFERENCES users(id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS links (
|
|
id TEXT PRIMARY KEY,
|
|
user_id TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
url TEXT NOT NULL,
|
|
comment TEXT,
|
|
timestamp TEXT NOT NULL,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
is_public INTEGER NOT NULL DEFAULT 1,
|
|
FOREIGN KEY(user_id) REFERENCES users(id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS plugins (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL UNIQUE,
|
|
version TEXT NOT NULL,
|
|
enabled INTEGER NOT NULL DEFAULT 1,
|
|
config TEXT,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS user_plugin_config (
|
|
id TEXT PRIMARY KEY,
|
|
user_id TEXT NOT NULL,
|
|
plugin_name TEXT NOT NULL,
|
|
config TEXT,
|
|
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(user_id, plugin_name),
|
|
FOREIGN KEY(user_id) REFERENCES users(id)
|
|
);
|
|
'''),
|
|
]
|
|
|
|
|
|
def get_connection() -> sqlite3.Connection:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute('PRAGMA foreign_keys = ON')
|
|
return conn
|
|
|
|
|
|
def get_schema_version(conn: sqlite3.Connection) -> int:
|
|
return conn.execute('PRAGMA user_version').fetchone()[0]
|
|
|
|
|
|
def apply_migrations(conn: sqlite3.Connection) -> None:
|
|
current_version = get_schema_version(conn)
|
|
for version, sql in MIGRATIONS:
|
|
if version <= current_version:
|
|
continue
|
|
conn.executescript(sql)
|
|
conn.execute(f'PRAGMA user_version = {version}')
|
|
conn.commit()
|
|
|
|
|
|
def init_db() -> None:
|
|
with get_connection() as conn:
|
|
apply_migrations(conn)
|
|
alice_hash = hash_password('secret123')
|
|
bob_hash = hash_password('secret123')
|
|
conn.execute(
|
|
'''
|
|
INSERT OR IGNORE INTO users (id, username, email, password_hash, is_admin)
|
|
VALUES (?, ?, ?, ?, 1)
|
|
''',
|
|
('user-1', 'alice', 'alice@example.com', alice_hash)
|
|
)
|
|
conn.execute("UPDATE users SET is_admin = 1 WHERE id = 'user-1'")
|
|
conn.execute(
|
|
'''
|
|
INSERT OR IGNORE INTO users (id, username, email, password_hash, is_admin)
|
|
VALUES (?, ?, ?, ?, 0)
|
|
''',
|
|
('user-2', 'bob', 'bob@example.com', bob_hash)
|
|
)
|
|
conn.execute(
|
|
'''
|
|
INSERT OR IGNORE INTO plugins (id, name, version, enabled, config)
|
|
VALUES (?, ?, ?, 1, ?)
|
|
''',
|
|
('plugin-1', 'default_frontend', '1.0.0', '{"route": "/"}')
|
|
)
|
|
conn.execute(
|
|
'''
|
|
INSERT OR IGNORE INTO plugins (id, name, version, enabled, config)
|
|
VALUES (?, ?, ?, 1, ?)
|
|
''',
|
|
('plugin-2', 'mastodon', '1.0.0', '{"enabled": true, "instance": "mastodon.social"}')
|
|
)
|
|
conn.commit()
|