47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
|
|
import duckdb
|
|
|
|
DB_PATH = "tasks/tasks.duckdb"
|
|
|
|
def init_db():
|
|
|
|
con = duckdb.connect(DB_PATH)
|
|
con.execute("""
|
|
CREATE TABLE IF NOT EXISTS tasks (
|
|
id INTEGER PRIMARY KEY,
|
|
title VARCHAR,
|
|
status VARCHAR DEFAULT 'todo',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
con.execute("CREATE SEQUENCE IF NOT EXISTS seq_task_id START 1")
|
|
con.close()
|
|
|
|
def get_connection():
|
|
return duckdb.connect(DB_PATH)
|
|
|
|
def add_task(title):
|
|
con = get_connection()
|
|
con.execute("INSERT INTO tasks (id, title) VALUES (nextval('seq_task_id'), ?)", [title])
|
|
con.close()
|
|
|
|
def list_tasks(status=None):
|
|
con = get_connection()
|
|
if status:
|
|
res = con.execute("SELECT * FROM tasks WHERE status = ? ORDER BY id", [status]).fetchall()
|
|
else:
|
|
res = con.execute("SELECT * FROM tasks ORDER BY id").fetchall()
|
|
con.close()
|
|
return res
|
|
|
|
def update_task(task_id, status):
|
|
con = get_connection()
|
|
con.execute("UPDATE tasks SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", [status, task_id])
|
|
con.close()
|
|
|
|
def delete_task(task_id):
|
|
con = get_connection()
|
|
con.execute("DELETE FROM tasks WHERE id = ?", [task_id])
|
|
con.close()
|