-
Notifications
You must be signed in to change notification settings - Fork 1
/
db_io.py
72 lines (60 loc) · 2.03 KB
/
db_io.py
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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import sqlite3
from sqlite3 import Error, Connection
from typing import Union
def clear_requirements(conn):
"""
Completely clears tables requirements and media_files
:param conn:
:return:
"""
select_sql = """ drop table if exists media_files"""
rows2 = select(conn, select_sql, ())
select_sql = """create table media_files (file_id integer primary key, path text not null, type text not null, provided integer not null)"""
rows2 = select(conn, select_sql, ())
select_sql = """ drop table if exists requirements"""
rows2 = select(conn, select_sql, ())
select_sql = """create table requirements (req_id integer primary key, map_id integer not null, file_id integer not null, foreign key (map_id) references maps (map_id), foreign key (file_id) references media_files(file_id))"""
rows2 = select(conn, select_sql, ())
def create_connection(db_file):
""" create a database connection to the SQLite database
specified by db_file
:param db_file: database file
:return: Connection object or None
"""
conn = None
try:
conn = sqlite3.connect(db_file)
return conn
except Error as e:
print(e)
return conn
def select(conn, select_sql, param):
"""
Query all rows in the tasks table
:param conn: the Connection object
:return:
"""
try:
cur = conn.cursor()
cur.execute(select_sql, param)
rows = cur.fetchall()
# for row in rows:
# print(row)
return rows
except Error as e:
print(e)
return
def find_map_name(keyword: str, conn: Connection) -> Union[bool, str]:
"""
returns first map with exact match of path or name and keyword
:param keyword:
:param conn:
:return:
"""
select_sql = """ select map_path from maps where map_path = ? or map_name = ?"""
rows = select(conn, select_sql, (keyword,)*2)
if rows:
rows = [a for b in rows for a in b]
return True, rows[0]
else:
return False, None