Files
flowdeck/app/db.py
T
bruno 5a124d1328
CI / test (push) Failing after 5s
CI / lint (push) Failing after 4s
v0.3.0: OAuth2, CSRF, rate limiting, issue CRUD, webhooks, card detail, checklists
- OAuth2 Gitea (login/callback/logout) with fallback admin mode
- Session management with signed cookies (itsdangerous, 7 days)
- CSRF protection middleware on all POST/PUT/PATCH/DELETE
- Rate limiting in-memory (60 req/min per IP)
- Issue creation from FlowDeck (POST /api/issues/{owner}/{repo})
- Inline issue editing (PATCH /api/issues/{owner}/{repo}/{issue_id})
- Card detail modal (double-click): description, labels, comments
- Checklists with toggleable items (CRUD endpoints + UI)
- Priority + due date editable on cards
- Webhook receiver (POST /api/webhook) with HMAC-SHA256
- Auto-register/status webhooks for repos
- GiteaClient: create_issue, update_issue, get_comments, webhooks, collaborators
- New DB tables: users, user_tokens, checklists, checklist_items
- CSS: modal, forms, checklists, comments, PR section
2026-07-08 09:23:57 -04:00

117 lines
4.4 KiB
Python

"""FlowDeck — Database layer (SQLite with WAL mode)."""
from __future__ import annotations
import sqlite3
from contextlib import contextmanager
from pathlib import Path
def get_db_path() -> Path:
from app.config import settings
return settings.db_path
def init_db():
"""Create tables if they don't exist."""
db_path = get_db_path()
db_path.parent.mkdir(parents=True, exist_ok=True)
with get_conn() as conn:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.executescript("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
login TEXT NOT NULL UNIQUE,
full_name TEXT NOT NULL DEFAULT '',
email TEXT NOT NULL DEFAULT '',
avatar_url TEXT NOT NULL DEFAULT '',
is_admin BOOLEAN NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS user_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
gitea_user_id INTEGER NOT NULL UNIQUE,
gitea_token TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS boards (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_owner TEXT NOT NULL,
project_name TEXT NOT NULL,
columns_json TEXT NOT NULL DEFAULT '["Backlog","À faire","En cours","Révision","Terminé"]',
wip_limits_json TEXT DEFAULT '{}',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(project_owner, project_name)
);
CREATE TABLE IF NOT EXISTS cards (
id INTEGER PRIMARY KEY AUTOINCREMENT,
board_id INTEGER NOT NULL REFERENCES boards(id),
gitea_issue_id INTEGER,
column_name TEXT NOT NULL DEFAULT 'Backlog',
position INTEGER NOT NULL DEFAULT 0,
priority TEXT DEFAULT 'Medium',
due_date TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_owner TEXT NOT NULL,
project_name TEXT NOT NULL,
title TEXT NOT NULL DEFAULT 'Notes',
content TEXT NOT NULL DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(project_owner, project_name, title)
);
CREATE TABLE IF NOT EXISTS col_mapping (
id INTEGER PRIMARY KEY AUTOINCREMENT,
board_id INTEGER NOT NULL REFERENCES boards(id),
column_name TEXT NOT NULL,
gitea_label TEXT NOT NULL,
close_issue BOOLEAN NOT NULL DEFAULT 0,
UNIQUE(board_id, column_name)
);
CREATE TABLE IF NOT EXISTS checklists (
id INTEGER PRIMARY KEY AUTOINCREMENT,
board_id INTEGER NOT NULL,
gitea_issue_id INTEGER NOT NULL,
title TEXT NOT NULL DEFAULT 'Checklist',
position INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(board_id, gitea_issue_id, title)
);
CREATE TABLE IF NOT EXISTS checklist_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
checklist_id INTEGER NOT NULL REFERENCES checklists(id) ON DELETE CASCADE,
content TEXT NOT NULL DEFAULT '',
checked BOOLEAN NOT NULL DEFAULT 0,
position INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
""")
conn.commit()
@contextmanager
def get_conn():
"""Yield a SQLite connection (WAL mode, foreign keys on)."""
db_path = get_db_path()
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
try:
yield conn
finally:
conn.close()