backend get_products

backend/api.py

@@ -0,0 +1,40 @@
+from utils import *
+
+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

backend/main.py

@@ -0,0 +1,8 @@
+from api import *
+
+if __name__ == "__main__":
+    products_json = get_products()
+    print(json.dumps(products_json, ensure_ascii=False, indent=2))
+
+    product1_json = get_product(1)
+    print(json.dumps(product1_json, ensure_ascii=False, indent=2))

backend/requirements.txt

@@ -0,0 +1 @@
+psycopg2-binary

backend/utils.py

@@ -0,0 +1,35 @@
+import psycopg2
+from psycopg2 import sql
+import json
+
+# Параметры подключения
+DB_CONFIG = {
+    'host': '192.168.1.100',
+    'port': 5432,  # стандартный порт PostgreSQL
+    'database': 'game_shop',  # замените на имя вашей БД
+    'user': 'casaos',  # замените на ваше имя пользователя
+    'password': 'casaos'  # замените на ваш пароль
+}
+
+def fetch_table_as_json(cursor, table_name) -> list[dict]:
+    cursor.execute(sql.SQL("SELECT * FROM {}").format(sql.Identifier(table_name)))
+    columns = [desc[0] for desc in cursor.description]  # Имена столбцов
+    rows = cursor.fetchall()
+    # Преобразуем каждую строку в словарь
+    result = [dict(zip(columns, row)) for row in rows]
+    return result
+
+
+def fetch_row_as_json(cursor, table_name, row_id) -> dict | None:
+    query = sql.SQL("SELECT * FROM {} WHERE id = %s").format(
+        sql.Identifier(table_name)
+    )
+    cursor.execute(query, (row_id,))
+
+    columns = [desc[0] for desc in cursor.description]  # Имена столбцов
+    row = cursor.fetchone()  # Получаем одну строку
+
+    if row is None:
+        return None  # или raise Exception, если нужно
+
+    return dict(zip(columns, row))