feat(handlers): Get data from db Methods
Added all methods from fronted documentation whitch needs
This commit is contained in:
Black_Raven committed
1 parent
0439ccd6d9
commit
b9c1a1c018
2 files changed
+354
-16
No files matched your search
+101
-11
@@ -1,23 +1,113 @@
|
||||
import sqlite3
|
||||
from module.category import Category
|
||||
from typing import List, Dict, Any, Optional
|
||||
from loguru import logger as log
|
||||
|
||||
def GetAllCategories(db_name: str = 'pipe_store.db', parent_id: int = 1) -> list[Category]:
|
||||
def get_all_categories_with_attributes(db_name: str = 'pipe_store.db') -> List[Dict[str, Any]]:
|
||||
conn = sqlite3.connect(db_name)
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
|
||||
try:
|
||||
# Получаем все категории (только основные, без parent_id)
|
||||
cursor.execute("""
|
||||
SELECT id, name, image_path
|
||||
FROM categories
|
||||
WHERE parent_id IS NULL OR parent_id = 1
|
||||
ORDER BY id
|
||||
""")
|
||||
|
||||
categories = []
|
||||
|
||||
cursor.execute("SELECT id, name, description, image_path FROM categories WHERE parent_id = ?", [parent_id])
|
||||
for cat_row in cursor.fetchall():
|
||||
category_id = cat_row['id']
|
||||
|
||||
for row in cursor.fetchall():
|
||||
category = Category(
|
||||
id = row["id"],
|
||||
name = row["name"],
|
||||
description= row["description"],
|
||||
image_path= row["image_path"],
|
||||
# Получаем все фильтры для товаров в этой категории
|
||||
cursor.execute("""
|
||||
SELECT DISTINCT f.id, f.name, f.filter_key
|
||||
FROM filters f
|
||||
WHERE f.filter_key IN (
|
||||
SELECT DISTINCT pf.filter_key
|
||||
FROM product_filters pf
|
||||
JOIN products p ON pf.product_id = p.id
|
||||
WHERE p.category_id = ?
|
||||
)
|
||||
categories.append(category.model_dump())
|
||||
ORDER BY f.id
|
||||
""", (category_id,))
|
||||
|
||||
attributes = []
|
||||
|
||||
for filter_row in cursor.fetchall():
|
||||
filter_id = filter_row['id']
|
||||
filter_key = filter_row['filter_key']
|
||||
filter_name = filter_row['name']
|
||||
|
||||
# Получаем уникальные значения для этого фильтра
|
||||
cursor.execute("""
|
||||
SELECT DISTINCT pf.value, COUNT(*) as count
|
||||
FROM product_filters pf
|
||||
JOIN products p ON pf.product_id = p.id
|
||||
WHERE p.category_id = ? AND pf.filter_key = ?
|
||||
GROUP BY pf.value
|
||||
ORDER BY pf.value
|
||||
""", (category_id, filter_key))
|
||||
|
||||
vals = []
|
||||
for val_row in cursor.fetchall():
|
||||
# Генерируем уникальный ID для значения
|
||||
val_id = hash(f"{category_id}_{filter_key}_{val_row['value']}") % 1000000
|
||||
vals.append({
|
||||
"id": val_id,
|
||||
"val": val_row['value']
|
||||
})
|
||||
|
||||
if vals: # Добавляем только если есть значения
|
||||
attributes.append({
|
||||
"id": filter_id,
|
||||
"name": filter_name,
|
||||
"vals": vals
|
||||
})
|
||||
|
||||
categories.append({
|
||||
"id": category_id,
|
||||
"name": cat_row['name'],
|
||||
"image": cat_row['image_path'] or f"/images/categories/{cat_row['name'].lower().replace(' ', '_')}.jpg",
|
||||
"atributes": attributes
|
||||
})
|
||||
|
||||
conn.close()
|
||||
return categories
|
||||
|
||||
except sqlite3.Error as e:
|
||||
log.error(f"Ошибка при получении категорий: {e}")
|
||||
return []
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_category_by_id(category_id: int, db_name: str = 'pipe_store.db') -> Optional[Dict[str, Any]]:
|
||||
"""Получить категорию по ID"""
|
||||
conn = sqlite3.connect(db_name)
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
|
||||
try:
|
||||
cursor.execute("""
|
||||
SELECT id, name, description, image_path, parent_id
|
||||
FROM categories
|
||||
WHERE id = ?
|
||||
""", (category_id,))
|
||||
|
||||
row = cursor.fetchone()
|
||||
if row:
|
||||
return {
|
||||
"id": row['id'],
|
||||
"name": row['name'],
|
||||
"description": row['description'],
|
||||
"image": row['image_path'],
|
||||
"parent_id": row['parent_id']
|
||||
}
|
||||
return None
|
||||
|
||||
except sqlite3.Error as e:
|
||||
log.error(f"Ошибка при получении категории {category_id}: {e}")
|
||||
return None
|
||||
finally:
|
||||
conn.close()
|
||||
@@ -1,11 +1,259 @@
|
||||
import sqlite3
|
||||
from module.product import Product
|
||||
from typing import List, Dict, Any, Optional, Tuple
|
||||
from loguru import logger as log
|
||||
|
||||
def get_products(db_name: str = 'pipe_store.db', parent_id: int = 1) -> List[Product]:
|
||||
def get_products_with_filters(
|
||||
category_id: int,
|
||||
page: int = 1,
|
||||
limit: int = 9,
|
||||
filters: Dict[str, Any] = None,
|
||||
db_name: str = 'pipe_store.db'
|
||||
) -> Dict[str, Any]:
|
||||
conn = sqlite3.connect(db_name)
|
||||
conn.row_factory = sqlite3.Row
|
||||
transaction = conn.cursor()
|
||||
cursor = conn.cursor()
|
||||
|
||||
products = []
|
||||
try:
|
||||
# Базовый запрос для получения товаров
|
||||
base_query = """
|
||||
SELECT DISTINCT p.id, p.name, p.full_name, p.description,
|
||||
p.diameter, p.wall_thickness, p.length, p.color,
|
||||
p.application_type, p.stiffness_class, p.image_path,
|
||||
p.created_at
|
||||
FROM products p
|
||||
WHERE p.category_id = ?
|
||||
"""
|
||||
|
||||
transaction.execute("SELECT ... FROM products")
|
||||
params = [category_id]
|
||||
|
||||
# Применяем фильтры
|
||||
if filters:
|
||||
for filter_key, filter_value in filters.items():
|
||||
if filter_value:
|
||||
if isinstance(filter_value, list):
|
||||
# Для множественных значений
|
||||
placeholders = ','.join(['?' for _ in filter_value])
|
||||
base_query += f"""
|
||||
AND p.id IN (
|
||||
SELECT DISTINCT pf.product_id
|
||||
FROM product_filters pf
|
||||
WHERE pf.filter_key = ?
|
||||
AND pf.value IN ({placeholders})
|
||||
)
|
||||
"""
|
||||
params.append(filter_key)
|
||||
params.extend(filter_value)
|
||||
else:
|
||||
# Для одиночных значений
|
||||
base_query += """
|
||||
AND p.id IN (
|
||||
SELECT DISTINCT pf.product_id
|
||||
FROM product_filters pf
|
||||
WHERE pf.filter_key = ? AND pf.value = ?
|
||||
)
|
||||
"""
|
||||
params.append(filter_key)
|
||||
params.append(filter_value)
|
||||
|
||||
# Получаем общее количество товаров
|
||||
count_query = f"SELECT COUNT(*) as total FROM ({base_query})"
|
||||
cursor.execute(count_query, params)
|
||||
total = cursor.fetchone()['total']
|
||||
|
||||
# Добавляем пагинацию
|
||||
offset = (page - 1) * limit
|
||||
paginated_query = f"""
|
||||
{base_query}
|
||||
ORDER BY p.id
|
||||
LIMIT ? OFFSET ?
|
||||
"""
|
||||
params.extend([limit, offset])
|
||||
|
||||
cursor.execute(paginated_query, params)
|
||||
|
||||
# Формируем результат
|
||||
items = []
|
||||
for row in cursor.fetchall():
|
||||
# Формируем атрибуты товара
|
||||
attributes = {}
|
||||
|
||||
# Добавляем все возможные атрибуты
|
||||
if row['color']:
|
||||
attributes['color'] = row['color']
|
||||
if row['length']:
|
||||
attributes['size'] = f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" if row['diameter'] else str(row['length'])
|
||||
if row['diameter']:
|
||||
attributes['diameter'] = str(row['diameter'])
|
||||
if row['wall_thickness']:
|
||||
attributes['wall_thickness'] = str(row['wall_thickness'])
|
||||
if row['stiffness_class']:
|
||||
attributes['stiffness_class'] = row['stiffness_class']
|
||||
if row['application_type']:
|
||||
attributes['application_type'] = row['application_type']
|
||||
|
||||
# Определяем количество в упаковке (можно вынести в отдельную таблицу)
|
||||
attributes['amount'] = get_pack_amount(row['id'], conn)
|
||||
|
||||
items.append({
|
||||
"id": row['id'],
|
||||
"name": row['name'],
|
||||
"description": row['description'] or f"Труба ПВХ {row['diameter']}мм",
|
||||
"price": calculate_price(row), # Функция для расчета цены
|
||||
"categoryId": category_id,
|
||||
"images": get_product_images(row['id'], row['image_path'], conn),
|
||||
"attributes": attributes,
|
||||
"inStock": check_stock(row['id'], conn) # Проверка наличия
|
||||
})
|
||||
|
||||
return {
|
||||
"items": items,
|
||||
"total": total,
|
||||
"page": page,
|
||||
"limit": limit,
|
||||
"total_pages": (total + limit - 1) // limit
|
||||
}
|
||||
|
||||
except sqlite3.Error as e:
|
||||
log.error(f"Ошибка при получении товаров: {e}")
|
||||
return {"items": [], "total": 0, "page": page, "limit": limit}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_product_by_id(product_id: int, db_name: str = 'pipe_store.db') -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Получить товар по ID с полной информацией
|
||||
|
||||
Args:
|
||||
product_id: ID товара
|
||||
db_name: Имя базы данных
|
||||
|
||||
Returns:
|
||||
Dict с полной информацией о товаре или None
|
||||
"""
|
||||
conn = sqlite3.connect(db_name)
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
|
||||
try:
|
||||
cursor.execute("""
|
||||
SELECT p.id, p.name, p.full_name, p.description,
|
||||
p.diameter, p.wall_thickness, p.length, p.color,
|
||||
p.application_type, p.stiffness_class, p.image_path,
|
||||
p.created_at, p.category_id
|
||||
FROM products p
|
||||
WHERE p.id = ?
|
||||
""", (product_id,))
|
||||
|
||||
row = cursor.fetchone()
|
||||
if not row:
|
||||
return None
|
||||
|
||||
# Получаем все изображения товара (если есть несколько)
|
||||
images = get_product_images(product_id, row['image_path'], conn)
|
||||
|
||||
# Формируем полное описание
|
||||
full_description = generate_full_description(row)
|
||||
|
||||
return {
|
||||
"id": row['id'],
|
||||
"name": row['name'],
|
||||
"full_name": row['full_name'],
|
||||
"description": full_description,
|
||||
"price": calculate_price(row),
|
||||
"categoryId": row['category_id'],
|
||||
"images": images,
|
||||
"attributes": {
|
||||
"color": row['color'],
|
||||
"diameter": str(row['diameter']) if row['diameter'] else None,
|
||||
"wall_thickness": str(row['wall_thickness']) if row['wall_thickness'] else None,
|
||||
"length": str(row['length']) if row['length'] else None,
|
||||
"stiffness_class": row['stiffness_class'],
|
||||
"application_type": row['application_type'],
|
||||
"size": f"{row['diameter']}x{row['wall_thickness']}x{row['length']}" if all([row['diameter'], row['wall_thickness'], row['length']]) else None
|
||||
},
|
||||
"inStock": check_stock(product_id, conn),
|
||||
"created_at": row['created_at']
|
||||
}
|
||||
|
||||
except sqlite3.Error as e:
|
||||
log.error(f"Ошибка при получении товара {product_id}: {e}")
|
||||
return None
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_product_images(product_id: int, main_image: Optional[str], conn: sqlite3.Connection) -> List[str]:
|
||||
"""
|
||||
Получить список изображений товара
|
||||
(В текущей структуре только одно изображение, но можно расширить)
|
||||
"""
|
||||
images = []
|
||||
if main_image:
|
||||
images.append(main_image)
|
||||
|
||||
# Здесь можно добавить логику для получения дополнительных изображений
|
||||
# Например, из отдельной таблицы product_images
|
||||
|
||||
# Если нет изображения, возвращаем заглушку
|
||||
if not images:
|
||||
images.append("/images/products/default.jpg")
|
||||
|
||||
return images
|
||||
|
||||
def calculate_price(row: sqlite3.Row) -> float:
|
||||
"""
|
||||
Рассчитать цену товара на основе параметров
|
||||
(В реальном проекте цена должна быть в базе данных)
|
||||
"""
|
||||
# Временная логика расчета цены
|
||||
base_price = 500
|
||||
|
||||
if row['diameter']:
|
||||
base_price += (row['diameter'] - 50) * 5
|
||||
|
||||
if row['length']:
|
||||
base_price += row['length'] * 0.1
|
||||
|
||||
if row['stiffness_class'] == 'SN4':
|
||||
base_price *= 1.2
|
||||
|
||||
return round(base_price, 2)
|
||||
|
||||
def check_stock(product_id: int, conn: sqlite3.Connection) -> bool:
|
||||
"""
|
||||
Проверить наличие товара на складе
|
||||
(В реальном проекте должна быть таблица stock)
|
||||
"""
|
||||
# Временная логика - все товары в наличии
|
||||
return True
|
||||
|
||||
def get_pack_amount(product_id: int, conn: sqlite3.Connection) -> str:
|
||||
"""
|
||||
Получить количество в упаковке
|
||||
(В реальном проекте должно быть в базе данных)
|
||||
"""
|
||||
# Временная логика
|
||||
return "4"
|
||||
|
||||
def generate_full_description(row: sqlite3.Row) -> str:
|
||||
"""Сгенерировать полное описание товара"""
|
||||
parts = []
|
||||
|
||||
if row['full_name']:
|
||||
parts.append(row['full_name'])
|
||||
|
||||
if row['description']:
|
||||
parts.append(row['description'])
|
||||
|
||||
if row['application_type']:
|
||||
parts.append(f"Тип применения: {row['application_type']} канализация")
|
||||
|
||||
if row['diameter'] and row['wall_thickness'] and row['length']:
|
||||
parts.append(f"Размеры: {row['diameter']}x{row['wall_thickness']}x{row['length']} мм")
|
||||
|
||||
if row['stiffness_class']:
|
||||
parts.append(f"Класс жесткости: {row['stiffness_class']}")
|
||||
|
||||
if row['color']:
|
||||
parts.append(f"Цвет: {row['color']}")
|
||||
|
||||
return " ".join(parts)
|
||||
Reference in new issue
Block a user