97 lines
2.7 KiB
Python
97 lines
2.7 KiB
Python
import psycopg2
|
|
|
|
try:
|
|
conn = psycopg2.connect("dbname='auth' user='auth' port='5432' host='localhost' password='123'")
|
|
except:
|
|
print("I am unable to connect to the database")
|
|
|
|
def add_account(login, passwd):
|
|
|
|
SQL_INSERT_ACCOUNT = f"""INSERT INTO account
|
|
(login, pswd, is_block, dt_create, "token", "refresh", dt_)
|
|
VALUES('{login}', '{passwd}', false, now(), '', '', now()); """
|
|
|
|
with conn.cursor() as curs:
|
|
try:
|
|
# simple single row system query
|
|
# conn.start()
|
|
# print(dir(conn))
|
|
curs.execute(SQL_INSERT_ACCOUNT)
|
|
conn.commit()
|
|
# returns a single row as a tuple
|
|
#single_row = curs.fetchone()
|
|
|
|
# use an f-string to print the single tuple returned
|
|
#print(f"{single_row}")
|
|
|
|
# simple multi row system query
|
|
# curs.execute("SELECT query, backend_type FROM pg_stat_activity")
|
|
|
|
# a default install should include this query and some backend workers
|
|
# use the * unpack operator to print many_rows which is a Python list
|
|
|
|
# a more robust way of handling errors
|
|
except (Exception, psycopg2.DatabaseError) as error:
|
|
print(error)
|
|
|
|
finally:
|
|
curs.close()
|
|
conn.close()
|
|
|
|
|
|
|
|
def get_list_account(_login):
|
|
SQL_SELECT_ACCOUNT = f"""SELECT login, pswd
|
|
FROM account
|
|
WHERE login like '{_login}'"""
|
|
|
|
with conn.cursor() as curs:
|
|
try:
|
|
|
|
curs.execute(SQL_SELECT_ACCOUNT)
|
|
# returns a single row as a tuple
|
|
single_row = curs.fetchone()
|
|
print(f"{single_row}")
|
|
return single_row
|
|
|
|
except (Exception, psycopg2.DatabaseError) as error:
|
|
print(error)
|
|
|
|
finally:
|
|
curs.close()
|
|
conn.close()
|
|
|
|
|
|
|
|
def delete_role(id):
|
|
id = int(id)
|
|
SQL_DELETE_ROLE = f"DELETE FROM role where ID = {id}"
|
|
with conn.cursor() as curs:
|
|
try:
|
|
curs.execute(SQL_DELETE_ROLE)
|
|
conn.commit()
|
|
except (Exception, psycopg2.DatabaseError) as error:
|
|
print(error)
|
|
finally:
|
|
curs.close()
|
|
|
|
|
|
|
|
def add_role(name):
|
|
SQL_INSERT_ROLE = f"INSERT INTO role (name) VALUES('{name}') RETURNING id"
|
|
|
|
with conn.cursor() as curs:
|
|
try:
|
|
curs.execute(SQL_INSERT_ROLE)
|
|
last_id = curs.fetchone()[0]
|
|
print(f"last_id:{last_id}")
|
|
conn.commit()
|
|
except (Exception, psycopg2.DatabaseError) as error:
|
|
print(error)
|
|
finally:
|
|
curs.close()
|
|
# conn.close()
|
|
return last_id
|
|
|
|
|