377 lines
No EOL
10 KiB
Python
377 lines
No EOL
10 KiB
Python
import psycopg2 as pg
|
||
from .objects import User, Rights
|
||
|
||
DB_CONFIG = {
|
||
"host": "127.0.0.1",
|
||
"port": 5432,
|
||
"dbname": "example", # TODO: Заменить на название из задания
|
||
"user": "example_user",
|
||
"password": "example_password123" # TODO: Заменить на реальный пароль
|
||
}
|
||
|
||
def get_connection():
|
||
"""Получить подключение к базе данных"""
|
||
return pg.connect(**DB_CONFIG)
|
||
|
||
|
||
def do_request(autocommit=False):
|
||
"""
|
||
Декоратор для выполнения SQL запросов с автоматическим управлением соединением
|
||
|
||
Args:
|
||
autocommit: Если True, автоматически фиксирует транзакцию
|
||
"""
|
||
|
||
def upper_wrapper(func):
|
||
def wrapper(*args, **kwargs):
|
||
conn = get_connection()
|
||
cursor = conn.cursor()
|
||
|
||
try:
|
||
kwargs['cursor'] = cursor
|
||
result = func(*args, **kwargs)
|
||
|
||
if autocommit:
|
||
conn.commit()
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f"Error: Database request failed: {e}")
|
||
conn.rollback()
|
||
return None
|
||
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
return wrapper
|
||
|
||
return upper_wrapper
|
||
|
||
|
||
@do_request()
|
||
def auth(login: str, password: str, *, cursor) -> User | None:
|
||
"""
|
||
Аутентификация пользователя
|
||
|
||
Args:
|
||
login: Имя пользователя
|
||
password: Пароль
|
||
|
||
Returns:
|
||
User объект или None если аутентификация не удалась
|
||
"""
|
||
cursor.execute("""
|
||
SELECT id, name, rights
|
||
FROM users
|
||
WHERE name = %s AND password = %s;
|
||
""", (login, password))
|
||
|
||
user = cursor.fetchone()
|
||
|
||
if not user:
|
||
print("Warning: Authentication failed - user not found")
|
||
return None
|
||
|
||
# Маппинг строки роли на Enum
|
||
rights_mapping = {
|
||
"admin": Rights.ADMIN,
|
||
"manager": Rights.MANAGER,
|
||
"client": Rights.CLIENT,
|
||
"customer": Rights.CLIENT, # Альтернативное название
|
||
"guest": Rights.GUEST,
|
||
}
|
||
|
||
rights = rights_mapping.get(user[2])
|
||
|
||
if not rights:
|
||
print(f"Warning: Unknown user role: {user[2]}")
|
||
return None
|
||
|
||
return User(
|
||
id=user[0],
|
||
name=user[1],
|
||
rights=rights
|
||
)
|
||
|
||
|
||
# TODO: Ниже функции для работы с предметной областью
|
||
# Заменить на ваши функции (например: get_products, add_product, etc.)
|
||
|
||
@do_request()
|
||
def get_items(*, cursor):
|
||
"""
|
||
Получить список доступных элементов
|
||
|
||
Например: get_products(), get_available_rooms(), etc.
|
||
"""
|
||
# Пример запроса - заменить на реальный
|
||
cursor.execute("""
|
||
SELECT *
|
||
FROM items
|
||
WHERE status = 'available';
|
||
""")
|
||
|
||
items = cursor.fetchall()
|
||
return items if items else None
|
||
|
||
|
||
@do_request(autocommit=True)
|
||
def create_request(item_id: int, quantity: int, user: User,
|
||
date_from: str = None, date_to: str = None,
|
||
notes: str = None, *, cursor):
|
||
"""
|
||
Создать новую заявку/заказ
|
||
|
||
Args:
|
||
item_id: ID товара/услуги
|
||
quantity: Количество
|
||
user: Объект пользователя
|
||
date_from: Дата начала (YYYY-MM-DD) или None
|
||
date_to: Дата окончания (YYYY-MM-DD) или None
|
||
notes: Дополнительные заметки
|
||
cursor: Курсор БД (автоматически через декоратор)
|
||
|
||
Returns:
|
||
True если успешно, False если ошибка
|
||
"""
|
||
try:
|
||
# 1. Проверяем существование товара и его доступность
|
||
cursor.execute("""
|
||
SELECT id, name, price, quantity, status
|
||
FROM items
|
||
WHERE id = %s;
|
||
""", (item_id,))
|
||
|
||
item = cursor.fetchone()
|
||
|
||
if not item:
|
||
print(f"Error: Item with id={item_id} not found")
|
||
return False
|
||
|
||
item_db_id, item_name, item_price, item_quantity, item_status = item
|
||
|
||
# 2. Проверяем статус товара
|
||
if item_status not in ['available', 'reserved']:
|
||
print(f"Error: Item '{item_name}' is not available (status: {item_status})")
|
||
return False
|
||
|
||
# 3. Проверяем количество (если товар физический)
|
||
if item_quantity < quantity:
|
||
print(f"Error: Insufficient quantity. Available: {item_quantity}, requested: {quantity}")
|
||
return False
|
||
|
||
# 4. Вычисляем total_price
|
||
total_price = float(item_price) * quantity
|
||
|
||
# 5. Создаём заявку
|
||
cursor.execute("""
|
||
INSERT INTO requests (user_id, item_id, quantity, date_from, date_to,
|
||
status, total_price, notes)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id;
|
||
""", (user.id, item_id, quantity, date_from, date_to,
|
||
'pending', total_price, notes))
|
||
|
||
request_id = cursor.fetchone()[0]
|
||
|
||
print(f"Success: Request #{request_id} created for user {user.name}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"Error creating request: {e}")
|
||
return False
|
||
|
||
|
||
@do_request()
|
||
def get_user_requests(user_id: int, *, cursor):
|
||
"""
|
||
Получить все заявки пользователя
|
||
|
||
Args:
|
||
user_id: ID пользователя
|
||
|
||
Returns:
|
||
List of tuples или None
|
||
"""
|
||
cursor.execute("""
|
||
SELECT
|
||
r.id,
|
||
r.created_at,
|
||
i.name AS item_name,
|
||
r.quantity,
|
||
r.total_price,
|
||
r.date_from,
|
||
r.date_to,
|
||
r.status,
|
||
r.notes
|
||
FROM requests r
|
||
JOIN items i ON r.item_id = i.id
|
||
WHERE r.user_id = %s
|
||
ORDER BY r.created_at DESC;
|
||
""", (user_id,))
|
||
|
||
return cursor.fetchall()
|
||
|
||
|
||
@do_request(autocommit=True)
|
||
def update_request_status(request_id: int, new_status: str, *, cursor):
|
||
"""
|
||
Обновить статус заявки (для менеджера/администратора)
|
||
|
||
Args:
|
||
request_id: ID заявки
|
||
new_status: Новый статус ('pending', 'approved', 'rejected', 'completed', 'cancelled')
|
||
|
||
Returns:
|
||
True если успешно, False если ошибка
|
||
"""
|
||
valid_statuses = ['pending', 'approved', 'rejected', 'completed', 'cancelled']
|
||
|
||
if new_status not in valid_statuses:
|
||
print(f"Error: Invalid status '{new_status}'. Must be one of: {valid_statuses}")
|
||
return False
|
||
|
||
try:
|
||
cursor.execute("""
|
||
UPDATE requests
|
||
SET status = %s, updated_at = CURRENT_TIMESTAMP
|
||
WHERE id = %s;
|
||
""", (new_status, request_id))
|
||
|
||
if cursor.rowcount == 0:
|
||
print(f"Error: Request #{request_id} not found")
|
||
return False
|
||
|
||
print(f"Success: Request #{request_id} status updated to '{new_status}'")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"Error updating request status: {e}")
|
||
return False
|
||
|
||
|
||
@do_request()
|
||
def get_all_requests(status_filter: str = None, *, cursor):
|
||
"""
|
||
Получить все заявки (для администратора/менеджера)
|
||
|
||
Args:
|
||
status_filter: Фильтр по статусу (опционально)
|
||
|
||
Returns:
|
||
List of tuples
|
||
"""
|
||
if status_filter:
|
||
cursor.execute("""
|
||
SELECT
|
||
r.id,
|
||
r.created_at,
|
||
u.name AS user_name,
|
||
u.email AS user_email,
|
||
i.name AS item_name,
|
||
r.quantity,
|
||
r.total_price,
|
||
r.date_from,
|
||
r.date_to,
|
||
r.status
|
||
FROM requests r
|
||
JOIN users u ON r.user_id = u.id
|
||
JOIN items i ON r.item_id = i.id
|
||
WHERE r.status = %s
|
||
ORDER BY r.created_at DESC;
|
||
""", (status_filter,))
|
||
else:
|
||
cursor.execute("""
|
||
SELECT
|
||
r.id,
|
||
r.created_at,
|
||
u.name AS user_name,
|
||
u.email AS user_email,
|
||
i.name AS item_name,
|
||
r.quantity,
|
||
r.total_price,
|
||
r.date_from,
|
||
r.date_to,
|
||
r.status
|
||
FROM requests r
|
||
JOIN users u ON r.user_id = u.id
|
||
JOIN items i ON r.item_id = i.id
|
||
ORDER BY r.created_at DESC;
|
||
""")
|
||
|
||
return cursor.fetchall()
|
||
|
||
|
||
@do_request()
|
||
def search_items(search_term: str, *, cursor):
|
||
"""
|
||
Поиск товаров по названию или описанию
|
||
|
||
Args:
|
||
search_term: Поисковый запрос
|
||
|
||
Returns:
|
||
List of tuples
|
||
"""
|
||
cursor.execute("""
|
||
SELECT
|
||
i.id,
|
||
i.name,
|
||
i.description,
|
||
i.price,
|
||
i.quantity,
|
||
i.status,
|
||
c.name AS category_name
|
||
FROM items i
|
||
LEFT JOIN categories c ON i.category_id = c.id
|
||
WHERE i.name ILIKE %s OR i.description ILIKE %s
|
||
ORDER BY i.name;
|
||
""", (f'%{search_term}%', f'%{search_term}%'))
|
||
|
||
return cursor.fetchall()
|
||
|
||
|
||
@do_request()
|
||
def get_items_by_category(category_id: int, *, cursor):
|
||
"""
|
||
Получить товары по категории
|
||
|
||
Args:
|
||
category_id: ID категории
|
||
|
||
Returns:
|
||
List of tuples
|
||
"""
|
||
cursor.execute("""
|
||
SELECT
|
||
i.id,
|
||
i.name,
|
||
i.description,
|
||
i.price,
|
||
i.quantity,
|
||
i.status
|
||
FROM items i
|
||
WHERE i.category_id = %s AND i.status = 'available'
|
||
ORDER BY i.name;
|
||
""", (category_id,))
|
||
|
||
return cursor.fetchall()
|
||
|
||
|
||
@do_request()
|
||
def get_all_categories(*, cursor):
|
||
"""
|
||
Получить все категории
|
||
|
||
Returns:
|
||
List of tuples (id, name)
|
||
"""
|
||
cursor.execute("""
|
||
SELECT id, name, description
|
||
FROM categories
|
||
ORDER BY name;
|
||
""")
|
||
|
||
return cursor.fetchall() |