from sqlite3 import Cursor import time from typing import List from todo.model.task_tag import TaskTag def one_is_used(cursor: Cursor, tag_ids: List[int]) -> bool: if len(tag_ids) >= 1: cursor.execute( "SELECT task_id FROM task_tags WHERE tag_id IN (%s) LIMIT 1" % ",".join("?"*len(tag_ids)), tag_ids) return len(cursor.fetchall()) == 1 else: return False def get(cursor: Cursor) -> List[TaskTag]: cursor.execute("SELECT task_id, tag_id FROM task_tags") return [TaskTag(r[0], r[1]) for r in cursor.fetchall()] def insert_many(cursor: Cursor, task_id: int, tag_ids: List[int]) -> List[TaskTag] : if len(tag_ids) >= 1: now = int(time.time()) task_tags = [TaskTag(task_id = task_id, tag_id = tag) for tag in tag_ids] cursor.executemany( " INSERT INTO task_tags(" " task_id," " tag_id," " created_at" " ) VALUES (?, ?, ?)", [(t.task_id, t.tag_id, now) for t in task_tags]) return task_tags else: return [] def delete_from_tasks(cursor: Cursor, task_ids: List[int]): if len(task_ids) >= 1: cursor.execute( "DELETE FROM task_tags WHERE task_id IN (%s)" % ",".join("?"*len(task_ids)), task_ids) def delete_from_tags(cursor: Cursor, task_id: int, tag_ids: List[int]): if len(tag_ids) >= 1: cursor.execute( "DELETE FROM task_tags WHERE task_id = ? AND tag_id IN (%s)" % ",".join("?"*len(tag_ids)), [task_id] + tag_ids)