end view

app/api/api.py

@@ -48,7 +48,7 @@ def validate_token(token: str) -> bool:
         cursor.close()
         connection.close()
 
-def login(email: str, password: str) :
+def login(email: str, password: str) -> str:
     connection = psycopg2.connect(**DB_CONFIG)
     cursor = connection.cursor()
     try:
@@ -74,7 +74,7 @@ def login(email: str, password: str) :
     finally:
         cursor.close()
 
-def logout(token):
+def logout(token) -> None:
     try:
         # Обновляем token_expiry_date на текущую дату (как в регистрации)
         today = date.today()
@@ -588,3 +588,148 @@ def add_money(token: str, money: str) -> None:
     finally:
         cursor.close()
         connection.close()
+
+
+def get_user(token: str) -> dict:
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        # Проверяем, авторизован ли пользователь (опционально, но рекомендуется)
+        if not check_token(cursor, token):
+            raise AuthError("Пользователь не авторизован или токен истёк")
+
+        # Получаем данные пользователя
+        cursor.execute("""
+            SELECT id, nickname, email, token, token_expiry_date, money, histories_id
+            FROM users 
+            WHERE token = %s
+        """, (token,))
+
+        row = cursor.fetchone()
+        if row is None:
+            raise ValueError("Пользователь не найден")
+
+        # Формируем словарь с именами столбцов
+        columns = [desc[0] for desc in cursor.description]
+        user_data = dict(zip(columns, row))
+
+        return user_data
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        raise
+    except ValueError as e:
+        raise
+    finally:
+        cursor.close()
+        connection.close()
+
+def get_products_by_id(products_id: list[int]) -> list[dict]:
+    if not products_id:
+        return []
+
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        # Считаем количество каждого product_id
+        product_counts = Counter(products_id)
+
+        # Получаем продукты из shop по id
+        cursor.execute("""
+            SELECT id, name, cost, count, reserved, picture_url, description, type
+            FROM shop
+            WHERE id = ANY(%s)
+        """, (products_id,))
+
+        rows = cursor.fetchall()
+        columns = [desc[0] for desc in cursor.description]
+
+        result = []
+        for row in rows:
+            product_dict = dict(zip(columns, row))
+            # Добавляем user_count — сколько раз этот товар встречается в корзине
+            product_dict["user_count"] = product_counts.get(product_dict["id"], 0)
+            result.append(product_dict)
+
+        return result
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        raise
+    finally:
+        cursor.close()
+        connection.close()
+
+def get_products_id(token: str, table_name: str) -> list[int]:
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        if not check_token(cursor, token):
+            raise AuthError("Пользователь не авторизован")
+
+        user_id: int = get_user_id(cursor, token)
+
+        cursor.execute("SELECT products_id FROM %s WHERE user_id = %s", (table_name, user_id,))
+        res = cursor.fetchone()
+
+        if res is None or res[0] is None:
+            return []
+
+        products_id = res[0]  # Это list[int], например: [1, 1, 3]
+
+        return products_id
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        raise
+    finally:
+        cursor.close()
+        connection.close()
+
+
+def get_histories_with_products(token: str) -> list[dict]:
+    connection = psycopg2.connect(**DB_CONFIG)
+    cursor = connection.cursor()
+    try:
+        if not check_token(cursor, token):
+            raise AuthError("Пользователь не авторизован")
+
+        user_id: int = get_user_id(cursor, token)
+
+        # Получаем histories_id пользователя
+        cursor.execute("SELECT histories_id FROM users WHERE id = %s", (user_id,))
+        res = cursor.fetchone()
+
+        if res is None or res[0] is None or len(res[0]) == 0:
+            return []
+
+        histories_id = res[0]
+
+        # Получаем истории
+        cursor.execute("""
+            SELECT id, user_id, products_id, date, products_cost
+            FROM history
+            WHERE id = ANY(%s::BIGINT[])
+        """, (histories_id,))
+
+        rows = cursor.fetchall()
+        columns = [desc[0] for desc in cursor.description]
+
+        result = []
+        for row in rows:
+            history_dict = dict(zip(columns, row))
+
+            # Добавляем поле "products" — список товаров с user_count
+            products_id = history_dict["products_id"]
+            history_dict["products"] = get_products_by_id(products_id)  # type: ignore
+
+            result.append(history_dict)
+
+        return result
+
+    except psycopg2.Error as e:
+        print("Ошибка при работе с PostgreSQL:", e)
+        raise
+    finally:
+        cursor.close()
+        connection.close()

app/server/urls.py

@@ -1,14 +1,27 @@
 #from django.contrib import admin
 from django.urls import path
 
-from .views import (get_shop, login, logout, register, unregister, get_basket, get_history)
+from .views import (
+    shop,
+    user,
+    login,
+    logout,
+    register,
+    unregister,
+    basket,
+    history,
+)
 
 urlpatterns = [
-    path("shop/", get_shop, name="get_shop"),
+    # страницы
+    path("shop/", shop, name="shop"),
+    path("user/", user, name="user"),
+    path("basket/", basket, name="basket"),
+    path("history/", history, name="history"),
+
+    # вспомогательные
     path("login/", login, name="login"),
     path("logout/", logout, name="logout"),
     path("register/", register, name="register"),
     path("unregister/", unregister, name="unregister"),
-    path("basket/", get_basket, name="get_basket"),
-    path("history/", get_history, name="get_history"),
 ]

app/server/views.py

@@ -6,7 +6,7 @@ from app.api import api
 from app.utils import decimal_to_float
 
 @csrf_exempt
-async def get_shop(request):
+async def shop(request):
     try:
         products = None
         if request.method == 'GET':
@@ -18,6 +18,74 @@ async def get_shop(request):
     except Exception as error:
         return JsonResponse({"ERROR": format(error)}, status=500)
 
+@csrf_exempt
+async def user(request):
+    try:
+        user1 = dict()
+        if request.method == 'POST':
+            body: dict = json.loads(request.body)
+
+            if body["register"]:
+                token = api.registration(body["register"]["nickname"],
+                                        body["register"]["password"],
+                                        body["register"]["email"])
+            elif body["login"]:
+                token = api.login(body["login"]["email"],
+                                body["login"]["password"])
+            elif body["unregister"]:
+                token = request.headers.get("Token")
+                api.unregister(token)
+            elif body["logout"]:
+                token = request.headers.get("Token")
+                api.logout(token)
+            elif body["add_money"]:
+                token = request.headers.get("Token")
+                api.add_money(token, body["add_money"]["money"])
+            else:
+                token = request.headers.get("Token")
+
+            user1 = api.get_user(token)
+
+        return JsonResponse({"OK": user1}, status=200)
+    except Exception as error:
+        return JsonResponse({"ERROR": format(error)}, status=500)
+
+
+@csrf_exempt
+async def basket(request):
+    try:
+        basket1 = None
+        if request.method == 'POST':
+            body: dict = json.loads(request.body)
+            token = request.headers.get("Token")
+
+            if body["add_product"]:
+                api.add_product_to_basket(token, body["add_product"]["product_id"])
+            elif body["delete_product"]:
+                api.delete_product_from_basket(token, body["delete_product"]["product_id"])
+            elif body["clear"]:
+                api.clear_basket(token)
+            elif body["buy_products"]:
+                api.buy_products(token)
+
+            products_id = api.get_products_id(token, "basket")
+            basket1 = api.get_products_by_id(products_id)
+
+        return JsonResponse({"OK": basket1}, status=200)
+    except Exception as error:
+        return JsonResponse({"error": format(error)}, status=500)
+
+@csrf_exempt
+async def history(request):
+    try:
+        histories = None
+        if request.method == 'POST':
+            token = request.headers.get("Token")
+            basket1 = api.get_histories_with_products(token)
+        return JsonResponse({"OK": histories}, status=200)
+    except Exception as error:
+        return JsonResponse({"error": format(error)}, status=500)
+
 @csrf_exempt
 async def login(request):
     try:
@@ -63,26 +131,50 @@ async def unregister(request):
         return JsonResponse({"error": format(error)}, status=500)
 
 @csrf_exempt
-async def get_basket(request):
+async def add_product_to_basket(request):
     try:
-        basket = None
+        token = None
         if request.method == 'POST':
+            body: dict = json.loads(request.body)
             token = request.headers.get("Token")
-            basket = api.get_basket(token)
-        return JsonResponse({"OK": basket}, status=200)
+            api.add_product_to_basket(token, body["product_id"])
+        return JsonResponse({"OK": token}, status=200)
     except Exception as error:
         return JsonResponse({"error": format(error)}, status=500)
 
 @csrf_exempt
-async def get_history(request):
+async def delete_product_from_basket(request):
     try:
-        histories = None
+        token = None
         if request.method == 'POST':
+            body: dict = json.loads(request.body)
             token = request.headers.get("Token")
-            histories = api.get_histories(token)
-        return JsonResponse({"OK": histories}, status=200)
+            api.delete_product_from_basket(token, body["product_id"])
+        return JsonResponse({"OK": token}, status=200)
     except Exception as error:
         return JsonResponse({"error": format(error)}, status=500)
 
+@csrf_exempt
+async def buy_products(request):
+    try:
+        token = None
+        if request.method == 'POST':
+            body: dict = json.loads(request.body)
+            token = request.headers.get("Token")
+            api.buy_products(token)
+        return JsonResponse({"OK": token}, status=200)
+    except Exception as error:
+        return JsonResponse({"error": format(error)}, status=500)
 
+@csrf_exempt
+async def clear_basket(request):
+    try:
+        token = None
+        if request.method == 'POST':
+            body: dict = json.loads(request.body)
+            token = request.headers.get("Token")
+            api.clear_basket(token)
+        return JsonResponse({"OK": token}, status=200)
+    except Exception as error:
+        return JsonResponse({"error": format(error)}, status=500)
 

app/tests/api_test.py

@@ -8,6 +8,7 @@ from app.api import (
     get_histories,
     unregister,
     get_products,
+    get_histories_with_products,
     add_money  # Импортируем напрямую
 )
 from app.api import add_product_to_shop, delete_product_from_shop
@@ -130,19 +131,30 @@ class TestFullUserFlow(unittest.TestCase):
         self.assertEqual(history['products_cost'], 130.0)
         print("✅ История покупки создана")
 
-    def test_10_check_basket_is_empty(self):
-        """10. Проверка, что корзина пуста"""
+    def test_10_get_history_with_products(self):
+        """10. test_16_get_history_with_products"""
+        conn = psycopg2.connect(**DB_CONFIG)
+        cur = conn.cursor()
+        try:
+            products = get_histories_with_products(self.token)
+            print(products)
+        finally:
+            cur.close()
+            conn.close()
+
+    def test_11_check_basket_is_empty(self):
+        """11. Проверка, что корзина пуста"""
         basket = get_basket(self.token)
         self.assertEqual(basket, '[]')
         print("✅ Корзина пуста")
 
-    def test_11_unregister_user(self):
-        """11. Удаление пользователя"""
+    def test_12_unregister_user(self):
+        """12. Удаление пользователя"""
         unregister(self.token)
         print("✅ Пользователь удалён")
 
-    def test_12_check_user_and_related_data_deleted(self):
-        """12. Проверка, что пользователь, корзина и история удалены"""
+    def test_13_check_user_and_related_data_deleted(self):
+        """13. Проверка, что пользователь, корзина и история удалены"""
         conn = psycopg2.connect(**DB_CONFIG)
         cur = conn.cursor()
         try:
@@ -164,14 +176,14 @@ class TestFullUserFlow(unittest.TestCase):
             cur.close()
             conn.close()
 
-    def test_13_delete_test_products(self):
-        """13. Удаление тестовых товаров"""
+    def test_14_delete_test_products(self):
+        """14. Удаление тестовых товаров"""
         for pid in self.product_ids:
             delete_product_from_shop(pid)
         print("✅ Тестовые товары удалены")
 
-    def test_14_check_products_deleted(self):
-        """14. Проверка, что товары удалены"""
+    def test_15_check_products_deleted(self):
+        """15. Проверка, что товары удалены"""
         conn = psycopg2.connect(**DB_CONFIG)
         cur = conn.cursor()
         try:
@@ -183,8 +195,8 @@ class TestFullUserFlow(unittest.TestCase):
             cur.close()
             conn.close()
 
-    def test_15_check_products(self):
-        """15. Проверка, что товары есть"""
+    def test_16_check_products(self):
+        """16. Проверка, что товары есть"""
         conn = psycopg2.connect(**DB_CONFIG)
         cur = conn.cursor()
         try:
@@ -232,4 +244,4 @@ class TestFullUserFlow(unittest.TestCase):
 
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()