-
Notifications
You must be signed in to change notification settings - Fork 1
/
db.py
44 lines (36 loc) · 1.46 KB
/
db.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
import psycopg2
from utils.config import Config
from datetime import datetime
class Database():
def __init__(self, db):
try:
self.conn = psycopg2.connect( database = db,
user = Config.PS_USERNAME,
password = Config.PS_PASSWORD,
host = Config.PS_HOST,
port = Config.PS_PORT)
self.cursor = self.conn.cursor()
self.db = db
except (Exception, psycopg2.Error) as error :
print ("Error while connecting to PostgreSQL", error)
def query(self, query: str, vars=None) -> psycopg2.extensions.cursor:
try:
if vars:
self.cursor.execute(query, vars)
else:
self.cursor.execute(query)
self.conn.commit()
return self.cursor
except (Exception, psycopg2.Error) as error :
print ("Error while querying PostgreSQL", error)
return self.cursor
# close communication with the PostgreSQL database server
def close(self )-> bool:
try:
self.cursor.close()
self.conn.close()
print ("\"" + self.db + "\" DB is disconnected!")
return True
except (Exception, psycopg2.Error) as error :
print ("Error while querying PostgreSQL", error)
return False