Skip to content

Commit 327bdd8

Browse files
cdvxcdvx
committed
refactor database connection classes
1 parent 5827b7c commit 327bdd8

4 files changed

Lines changed: 144 additions & 143 deletions

File tree

‎app/utils/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
11
from .utils import ValidationError
2-
from .connect import conn, DbManager
2+
from .db import conn, DbManager
33

‎app/utils/connect.py‎

Lines changed: 0 additions & 141 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,5 @@
1-
import os
21

32
import psycopg2
4-
from config import Config
5-
from .sql_queries import SQL
6-
7-
8-
DATABASE_URL = Config.POSTGREST_DATABASE_URI
9-
DB_NAME = Config.DB_NAME
10-
113

124
class DatabaseConnection:
135
"""Class to manage database connection"""
@@ -28,136 +20,3 @@ def __init__(self, url=None):
2820
print("Cannot connect to database", e)
2921
else: #pragma: no cover
3022
print('Could not connect to database\nReason: Database url not provided!')
31-
32-
class DbMethods:
33-
"""Class with database actions"""
34-
35-
def create_Users_table(self):
36-
DbManager.excute_sql(SQL['create_users_table'], tablename='users')
37-
38-
def create_Questions_table(self):
39-
DbManager.excute_sql(SQL['create_questions_table'], tablename='questions')
40-
41-
def create_Answers_table(self):
42-
DbManager.excute_sql(SQL['create_answers_table'], tablename='answers')
43-
44-
def insert_new_record(self, tablename, data):
45-
if tablename == 'users':
46-
DbManager.excute_sql(
47-
SQL['insert_into_users'],
48-
tablename='users',
49-
values=(
50-
data['username'],
51-
data['email'],
52-
data['password'],
53-
data['user_id'],
54-
))
55-
56-
elif tablename == 'questions':
57-
DbManager.excute_sql(
58-
SQL['insert_into_questions'],
59-
tablename='questions',
60-
values=(
61-
data['topic'],
62-
data['body'],
63-
data['author'],
64-
data['questionId']
65-
))
66-
67-
elif tablename == 'answers':
68-
DbManager.excute_sql(
69-
SQL['insert_into_answers'],
70-
tablename='answers',
71-
values=(
72-
data['Qn_Id'],
73-
data['body'],
74-
data['answerId'],
75-
data['author'],
76-
data['prefered']
77-
))
78-
79-
def query_all(self, tablename):
80-
return DbManager.get(SQL['select'], tablename=tablename)
81-
82-
83-
def get_user(self, username):
84-
return DbManager.get(SQL['fetch_user'], value=username)
85-
86-
def get_question(self, question_id):
87-
return DbManager.get(SQL['get_question'], value=question_id)
88-
89-
def update_question(self, new_topic, new_body, questionId):
90-
DbManager.excute_sql(SQL['update_question'], values=(new_topic, new_body, questionId))
91-
92-
def update_answer(self, answerId):
93-
DbManager.excute_sql(SQL['update_question'], values=(True, str(answerId)))
94-
95-
def delete_entry(self, tablename, id_value):
96-
if tablename == 'questions':
97-
DbManager.excute_sql(SQL['delete_question'], values=(id_value, ))
98-
elif tablename == 'answers':
99-
DbManager.excute_sql(SQL['delete_answer'], values=(id_value, ))
100-
elif tablename == 'users':
101-
DbManager.excute_sql(SQL['delete_user'], values=(id_value, ))
102-
103-
def drop_table(self, tablename):
104-
DbManager.excute_sql(SQL['drop'].format(tablename), drop=True)
105-
106-
def create_all(self):
107-
self.create_Answers_table()
108-
self.create_Users_table()
109-
self.create_Questions_table()
110-
111-
def drop_all(self):
112-
for table in self.tablenames:
113-
self.drop_table(table)
114-
115-
class DbManager(DatabaseConnection, DbMethods):
116-
"""Class ro manage database actions"""
117-
118-
def __init__(self, env, local=None):
119-
super(DbManager, self).__init__(
120-
url=DbManager.create_db_url(env, local=local)
121-
)
122-
self.tablenames = ['users', 'questions', 'answers']
123-
124-
@classmethod
125-
def create_db_url(cls, env, local=None):
126-
test_db = Config.TEST_DB
127-
db_name = Config.DB_NAME
128-
if env in ['testing', 'development']:
129-
return f"postgresql://localhost/{db_name}" if local \
130-
else f"postgresql://localhost/{test_db}"
131-
return Config.POSTGREST_DATABASE_URI
132-
133-
134-
@classmethod
135-
def excute_sql(cls, sql, tablename=None, values=None, drop=None):
136-
try:
137-
if tablename and drop is None:
138-
cls.instance.tablename = tablename
139-
cls.instance.cursor.execute(sql, values) if values else cls.instance.cursor.execute(sql)
140-
except (Exception, psycopg2.DatabaseError) as error: #pragma: no cover
141-
print(error)
142-
143-
@classmethod
144-
def get(cls, sql, tablename=None, value=None):
145-
try:
146-
if tablename:
147-
cls.instance.cursor.execute(sql.format(tablename))
148-
results = cls.instance.cursor.fetchall()
149-
return list(results)
150-
elif value:
151-
cls.instance.cursor.execute(sql, (value,))
152-
result = cls.instance.cursor.fetchone()
153-
return result if result else None
154-
155-
except (Exception, psycopg2.DatabaseError) as error: #pragma: no cover
156-
print(error)
157-
158-
159-
160-
conn = DbManager(os.getenv('APP_SETTINGS'), local=True)
161-
162-
conn.create_all()
163-
# conn.drop_all

‎app/utils/db.py‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
import os
2+
3+
import psycopg2
4+
from config import Config
5+
from .sql_queries import SQL
6+
from .connect import DatabaseConnection
7+
8+
9+
DATABASE_URL = Config.POSTGREST_DATABASE_URI
10+
DB_NAME = Config.DB_NAME
11+
12+
class DbActions(DatabaseConnection):
13+
"""Class with database actions"""
14+
15+
def create_Users_table(self):
16+
DbManager.excute_sql(SQL['create_users_table'], tablename='users')
17+
18+
def create_Questions_table(self):
19+
DbManager.excute_sql(SQL['create_questions_table'], tablename='questions')
20+
21+
def create_Answers_table(self):
22+
DbManager.excute_sql(SQL['create_answers_table'], tablename='answers')
23+
24+
def insert_new_record(self, tablename, data):
25+
if tablename == 'users':
26+
DbManager.excute_sql(
27+
SQL['insert_into_users'],
28+
tablename='users',
29+
values=(
30+
data['username'],
31+
data['email'],
32+
data['password'],
33+
data['user_id'],
34+
))
35+
36+
elif tablename == 'questions':
37+
DbManager.excute_sql(
38+
SQL['insert_into_questions'],
39+
tablename='questions',
40+
values=(
41+
data['topic'],
42+
data['body'],
43+
data['author'],
44+
data['questionId']
45+
))
46+
47+
elif tablename == 'answers':
48+
DbManager.excute_sql(
49+
SQL['insert_into_answers'],
50+
tablename='answers',
51+
values=(
52+
data['Qn_Id'],
53+
data['body'],
54+
data['answerId'],
55+
data['author'],
56+
data['prefered']
57+
))
58+
59+
def query_all(self, tablename):
60+
return DbManager.get(SQL['select'], tablename=tablename)
61+
62+
63+
def get_user(self, username):
64+
return DbManager.get(SQL['fetch_user'], value=username)
65+
66+
def get_question(self, question_id):
67+
return DbManager.get(SQL['get_question'], value=question_id)
68+
69+
def update_question(self, new_topic, new_body, questionId):
70+
DbManager.excute_sql(SQL['update_question'], values=(new_topic, new_body, questionId))
71+
72+
def update_answer(self, answerId):
73+
DbManager.excute_sql(SQL['update_question'], values=(True, str(answerId)))
74+
75+
def delete_entry(self, tablename, id_value):
76+
if tablename == 'questions':
77+
DbManager.excute_sql(SQL['delete_question'], values=(id_value, ))
78+
elif tablename == 'answers':
79+
DbManager.excute_sql(SQL['delete_answer'], values=(id_value, ))
80+
elif tablename == 'users':
81+
DbManager.excute_sql(SQL['delete_user'], values=(id_value, ))
82+
83+
def drop_table(self, tablename):
84+
DbManager.excute_sql(SQL['drop'].format(tablename), drop=True)
85+
86+
def create_all(self):
87+
self.create_Answers_table()
88+
self.create_Users_table()
89+
self.create_Questions_table()
90+
91+
def drop_all(self):
92+
for table in self.tablenames:
93+
self.drop_table(table)
94+
95+
class DbManager(DbActions):
96+
"""Class to manage database actions"""
97+
98+
def __init__(self, env, local=None):
99+
super(DbManager, self).__init__(
100+
url=DbManager.create_db_url(env, local=local)
101+
)
102+
self.tablenames = ['users', 'questions', 'answers']
103+
104+
@classmethod
105+
def create_db_url(cls, env, local=None):
106+
test_db = Config.TEST_DB
107+
db_name = Config.DB_NAME
108+
if env in ['testing', 'development']:
109+
return f"postgresql://localhost/{db_name}" if local \
110+
else f"postgresql://localhost/{test_db}"
111+
return Config.POSTGREST_DATABASE_URI
112+
113+
114+
@classmethod
115+
def excute_sql(cls, sql, tablename=None, values=None, drop=None):
116+
try:
117+
if tablename and drop is None:
118+
cls.instance.tablename = tablename
119+
cls.instance.cursor.execute(sql, values) if values else cls.instance.cursor.execute(sql)
120+
except (Exception, psycopg2.DatabaseError) as error: #pragma: no cover
121+
print(error)
122+
123+
@classmethod
124+
def get(cls, sql, tablename=None, value=None):
125+
try:
126+
if tablename:
127+
cls.instance.cursor.execute(sql.format(tablename))
128+
results = cls.instance.cursor.fetchall()
129+
return list(results)
130+
elif value:
131+
cls.instance.cursor.execute(sql, (value,))
132+
result = cls.instance.cursor.fetchone()
133+
return result if result else None
134+
135+
except (Exception, psycopg2.DatabaseError) as error: #pragma: no cover
136+
print(error)
137+
138+
139+
140+
conn = DbManager(os.getenv('APP_SETTINGS'), local=True)
141+
142+
conn.create_all()
143+
# conn.drop_all

‎tests/test_route.py‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,6 @@ def test_user_can_get_questions(client, insert_new_records):
8080

8181
def test_user_can_get_question(client, insert_new_records):
8282
questionsList = conn.query_all('questions')
83-
print(questionsList)
8483
if questionsList :
8584
res = client.get(f'/api/v1/questions/{questionsList[0][4]}')
8685
assert 'questionId' in res.json

0 commit comments

Comments
 (0)