1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
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)
|
