|
|
from utils import *
|
|
|
import bcrypt
|
|
|
import uuid
|
|
|
from datetime import date
|
|
|
from psycopg2 import IntegrityError
|
|
|
|
|
|
def hash_password(plain_password: str):
|
|
|
hashed_bytes = bcrypt.hashpw(plain_password.encode('utf-8'), bcrypt.gensalt())
|
|
|
hashed_str = hashed_bytes.decode('utf-8')
|
|
|
return hashed_str
|
|
|
|
|
|
class AuthError(Exception):
|
|
|
pass
|
|
|
|
|
|
def login(email: str, password: str) :
|
|
|
connection = psycopg2.connect(**DB_CONFIG)
|
|
|
cursor = connection.cursor()
|
|
|
try:
|
|
|
cursor.execute("SELECT password,token FROM users WHERE mail = %s", (email,))
|
|
|
result = cursor.fetchone()
|
|
|
|
|
|
if result is None:
|
|
|
raise AuthError("Неверный email или пароль")
|
|
|
|
|
|
print(result)
|
|
|
stored_hash, token = result
|
|
|
if not bcrypt.checkpw(password.encode('utf-8'), stored_hash.encode('utf-8')):
|
|
|
raise AuthError("Неверный email или пароль")
|
|
|
|
|
|
if token is None:
|
|
|
# Опционально: можно сгенерировать новый токен здесь
|
|
|
raise AuthError("У пользователя отсутствует токен")
|
|
|
|
|
|
return token
|
|
|
finally:
|
|
|
cursor.close()
|
|
|
|
|
|
|
|
|
def registration(nickname: str, password: str, email: str) -> str:
|
|
|
# Хэшируем пароль
|
|
|
hashed = hash_password(password)
|
|
|
token = str(uuid.uuid4())
|
|
|
token_expiry_date = date.today().strftime("%Y.%m.%d")
|
|
|
|
|
|
money = "100"
|
|
|
histories_id = "{}"
|
|
|
|
|
|
connection = psycopg2.connect(**DB_CONFIG)
|
|
|
cursor = connection.cursor()
|
|
|
try:
|
|
|
cursor.execute(
|
|
|
"INSERT INTO users (nickname, mail, password, token, token_expiry_date, money, histories_id) "
|
|
|
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
|
|
|
(nickname, email, hashed, token, token_expiry_date, money, histories_id)
|
|
|
)
|
|
|
connection.commit()
|
|
|
print("Пользователь успешно создан")
|
|
|
return token
|
|
|
except IntegrityError as e:
|
|
|
connection.rollback()
|
|
|
if "email_uniq" in str(e) or "users_email_key" in str(e):
|
|
|
print("Ошибка: пользователь с таким email уже существует")
|
|
|
raise AuthError("Пользователь с таким email уже существует") from e
|
|
|
else:
|
|
|
print("Другая ошибка базы данных:", e)
|
|
|
raise
|
|
|
finally:
|
|
|
cursor.close()
|
|
|
|
|
|
def get_products() -> list[dict]:
|
|
|
connection = None
|
|
|
cursor = None
|
|
|
res = None
|
|
|
try:
|
|
|
connection = psycopg2.connect(**DB_CONFIG)
|
|
|
cursor = connection.cursor()
|
|
|
|
|
|
res = fetch_table_as_json(cursor, "shop")
|
|
|
|
|
|
except psycopg2.Error as e:
|
|
|
print("Ошибка при работе с PostgreSQL:", e)
|
|
|
|
|
|
finally:
|
|
|
cursor.close()
|
|
|
connection.close()
|
|
|
|
|
|
return res
|
|
|
|
|
|
def get_product(product_id: int):
|
|
|
connection = None
|
|
|
cursor = None
|
|
|
res = None
|
|
|
try:
|
|
|
connection = psycopg2.connect(**DB_CONFIG)
|
|
|
cursor = connection.cursor()
|
|
|
|
|
|
res = fetch_row_as_json(cursor, "shop", product_id)
|
|
|
|
|
|
except psycopg2.Error as e:
|
|
|
print("Ошибка при работе с PostgreSQL:", e)
|
|
|
|
|
|
finally:
|
|
|
cursor.close()
|
|
|
connection.close()
|
|
|
|
|
|
return res
|