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
|
from sqlite3 import Cursor
import time
from typing import List
from model.task_tag import TaskTag
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] :
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
def delete(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)
|