-
Notifications
You must be signed in to change notification settings - Fork 0
/
database.py
138 lines (110 loc) · 4.21 KB
/
database.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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
from typing import List, Tuple
# -- DATABASE TYPES
User = Tuple[int, str, str, str]
Contact = Tuple[int, str, str, str, int]
# -- DATABASE QUERIES --
CREATE_USERS = """CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT,
email TEXT,
password TEXT
);"""
CREATE_CONTACTS = """CREATE TABLE IF NOT EXISTS contacts (
id INTEGER PRIMARY KEY,
name TEXT,
phone_no TEXT,
email TEXT,
user_id INTEGER,
FOREIGN KEY(user_id) REFERENCES users(id)
);"""
SELECT_ALL_USERS = "SELECT * FROM users;"
SELECT_USER = "SELECT * FROM users WHERE id = ?;"
SELECT_USER_CONTACTS = "SELECT * FROM contacts WHERE user_id = ?;"
SELECT_CONTACT = "SELECT * FROM contacts WHERE id = ?;"
INSERT_USER = "INSERT INTO users (name, email, password) VALUES (?, ?, ?);"
INSERT_CONTACT = "INSERT INTO contacts (name, phone_no, email, user_id) VALUES (?, ?, ?, ?);"
UPDATE_USER = """UPDATE users
SET name = ?,
email = ?,
password = ?
WHERE id = ?;"""
UPDATE_CONTACT = """UPDATE contacts
SET name = ?,
phone_no = ?,
email = ?
WHERE id = ? AND user_id = ?;"""
DELETE_USER = "DELETE FROM users WHERE id = ?;"
DELETE_USER_CONTACTS = "DELETE FROM contacts where user_id = ?;"
DELETE_CONTACT = "DELETE FROM contacts where id = ?;"
# -- DATABASE CONNECTION --
def create_tables(connection):
"""Create database tables"""
with connection:
cursor = connection.cursor()
cursor.execute(CREATE_USERS)
cursor.execute(CREATE_CONTACTS)
# -- USERS --
def get_users(connection) -> List[User]:
"""Returns users from database"""
with connection:
cursor = connection.cursor()
cursor.execute(SELECT_ALL_USERS)
return cursor.fetchall()
def get_user(connection, id: int) -> User:
"""Return a User from database"""
with connection:
cursor = connection.cursor()
cursor.execute(SELECT_USER, (id,))
return cursor.fetchone()
def create_user(connection, name: str, email: str, password: str):
"""Create new user & return id from database"""
with connection:
cursor = connection.cursor()
cursor.execute(INSERT_USER, (name, email, password))
user_id = cursor.lastrowid
return user_id
def update_user(connection, name: str, email: str, password: str, user_id: int):
"""Update specified user in database"""
with connection:
cursor = connection.cursor()
cursor.execute(UPDATE_USER, (name, email, password, user_id))
return cursor.fetchone()
def delete_user(connection, user_id: int):
"""Delete specified user from database"""
with connection:
cursor = connection.cursor()
cursor.execute(DELETE_USER, (user_id,))
cursor.execute(DELETE_USER_CONTACTS, (user_id,))
# -- CONTACTS --
def get_contacts(connection, user_id: int) -> List[Contact]:
"""Get ALL contact for a specified user"""
with connection:
cursor = connection.cursor()
cursor.execute(SELECT_USER_CONTACTS, (user_id,))
return cursor.fetchall()
def get_contact(connection, contact_id) -> Contact:
"""Get contact for user"""
with connection:
cursor = connection.cursor()
cursor.execute(SELECT_CONTACT, (contact_id,))
return cursor.fetchone()
def add_contact(connection, contact_name: str, contact_phone_no: str, contact_email: str, user_id: int):
"""Create a new contact for specified user"""
with connection:
cursor = connection.cursor()
cursor.execute(INSERT_CONTACT, (contact_name,
contact_phone_no, contact_email, user_id))
contact_id = cursor.lastrowid
return contact_id
def update_contact(connection, contact_name: str, contact_phone_no: str, contact_email: str, contact_id: int, user_id: int):
"""Update specified contact in database"""
with connection:
cursor = connection.cursor()
cursor.execute(UPDATE_CONTACT, (contact_name,
contact_phone_no, contact_email, contact_id, user_id))
return cursor.fetchone()
def delete_contact(connection, contact_id: int):
"""Delete specified contact from database"""
with connection:
cursor = connection.cursor()
cursor.execute(DELETE_CONTACT, (contact_id,))