new endpoints

flask_app/logic.py

@@ -97,4 +97,158 @@ def get_user_by_params(user_id: str = None, wallet_address: str = None, nickname
 
     except sqlite3.Error as e:
         LOGGER.error(f"Database error: {e}")
+        raise Exception("An error occurred with the database operation") from e
+
+def add_license_to_user(license_id: str, user_id: str = None, wallet_address: str = None, nickname: str = None) -> str:
+    LOGGER.info("add_license_to_user_db called")
+
+    if not (user_id or wallet_address or nickname):
+        LOGGER.error("No valid user parameter provided")
+        raise ValueError("No valid user parameter provided")
+
+    try:
+        with sqlite3.connect(DB_PATH) as conn:
+            c = conn.cursor()
+
+            # Определение используемого параметра пользователя
+            if user_id:
+                param = user_id
+                column = "userid"
+            elif wallet_address:
+                param = wallet_address
+                column = "wallet_address"
+            elif nickname:
+                param = nickname
+                column = "nickname"
+
+            # Получение user_id по заданному параметру
+            c.execute(f"SELECT userid FROM users WHERE {column} = ?", (param,))
+            row = c.fetchone()
+            if not row:
+                LOGGER.info("No user found with the given parameters")
+                return "user_not_found"
+            user_id = row[0]
+
+            # Проверка наличия лицензии с таким же license_id
+            c.execute("SELECT COUNT(*) FROM licenses WHERE licenseid = ?", (license_id,))
+            count = c.fetchone()[0]
+            if count > 0:
+                LOGGER.info(f"License {license_id} already exists")
+                return "license_already_exists"
+
+            # Добавление лицензии в таблицу licenses
+            c.execute("INSERT INTO licenses (licenseid) VALUES (?)", (license_id,))
+
+            # Связывание лицензии с пользователем в таблице user_licenses
+            c.execute("INSERT INTO user_licenses (userid, licenseid) VALUES (?, ?)", (user_id, license_id))
+
+            conn.commit()
+            LOGGER.info(f"License {license_id} added and linked to user {user_id}")
+            return "success"
+
+    except sqlite3.Error as e:
+        LOGGER.error(f"Database error: {e}")
+        conn.rollback()
+        raise Exception("An error occurred with the database operation") from e
+
+def get_user_by_license_id(license_id: str) -> dict:
+    LOGGER.info("get_user_by_license_id_db called")
+
+    try:
+        with sqlite3.connect(DB_PATH) as conn:
+            c = conn.cursor()
+            c.execute("""
+                SELECT u.userid, u.wallet_address, u.created_at, u.nickname, l.licenseid
+                FROM users u
+                JOIN user_licenses ul ON u.userid = ul.userid
+                JOIN licenses l ON ul.licenseid = l.licenseid
+                WHERE l.licenseid = ?
+            """, (license_id,))
+            row = c.fetchone()
+
+            if row:
+                user_data = {
+                    'user_id': row[0],
+                    'wallet_address': row[1],
+                    'created_at': row[2],
+                    'nickname': row[3]
+                }
+                LOGGER.info(f"User found for license {license_id}")
+                return user_data
+            else:
+                LOGGER.info(f"No user found for license {license_id}")
+                return None
+
+    except sqlite3.Error as e:
+        LOGGER.error(f"Database error: {e}")
+        raise Exception("An error occurred with the database operation") from e
+
+def get_licenses_by_user_params(user_id: str = None, wallet_address: str = None, nickname: str = None) -> list:
+    LOGGER.info("get_licenses_by_user_params_db called")
+
+    if not (user_id or wallet_address or nickname):
+        LOGGER.error("No valid user parameter provided")
+        raise ValueError("No valid user parameter provided")
+
+    try:
+        with sqlite3.connect(DB_PATH) as conn:
+            c = conn.cursor()
+
+            if user_id:
+                param = user_id
+                column = "userid"
+            elif wallet_address:
+                param = wallet_address
+                column = "wallet_address"
+            elif nickname:
+                param = nickname
+                column = "nickname"
+
+            c.execute(f"""
+                SELECT l.licenseid
+                FROM licenses l
+                JOIN user_licenses ul ON l.licenseid = ul.licenseid
+                JOIN users u ON ul.userid = u.userid
+                WHERE u.{column} = ?
+            """, (param,))
+            rows = c.fetchall()
+
+            if rows:
+                licenses_data = [{'license_id': row[0]} for row in rows]
+                LOGGER.info(f"Licenses found for user with {column} = {param}")
+                return licenses_data
+            else:
+                LOGGER.info(f"No licenses found for user with {column} = {param}")
+                return None
+
+    except sqlite3.Error as e:
+        LOGGER.error(f"Database error: {e}")
+        raise Exception("An error occurred with the database operation") from e
+
+def delete_license_by_id(license_id: str) -> bool:
+    LOGGER.info("delete_license_by_id_db called")
+
+    try:
+        with sqlite3.connect(DB_PATH) as conn:
+            c = conn.cursor()
+
+            # Удаление связи из таблицы user_licenses
+            c.execute("DELETE FROM user_licenses WHERE licenseid = ?", (license_id,))
+
+            # Удаление лицензии из таблицы licenses
+            c.execute("DELETE FROM licenses WHERE licenseid = ?", (license_id,))
+            licenses_deleted = c.rowcount
+
+            conn.commit()
+
+            if licenses_deleted > 0:
+                LOGGER.info(f"License {license_id} and associated user licenses deleted")
+                return True
+            else:
+                LOGGER.info(f"License {license_id} not found")
+                return False
+
+    except sqlite3.Error as e:
+        LOGGER.error(f"Database error: {e}")
+        conn.rollback()
         raise Exception("An error occurred with the database operation") from e

flask_app/views.py

@@ -52,5 +52,100 @@ def get_user():
         else:
             return jsonify({'error': 'User not found'}), 404
 
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+@app.route('/api/addLicense', methods=['POST'])
+@limiter.limit("4 per minute")
+@token_required
+def add_license():
+    LOGGER.info('add_license called')
+
+    data = request.get_json()
+    license_id = data.get('license_id')
+    user_id = data.get('user_id')
+    wallet_address = data.get('wallet_address')
+    nickname = data.get('nickname')
+
+    if not license_id:
+        return jsonify({'error': 'license_id is required'}), 400
+
+    if sum(bool(param) for param in [user_id, wallet_address, nickname]) != 1:
+        return jsonify({'error': 'Exactly one user parameter (user_id, wallet_address, or nickname) is required'}), 400
+
+    try:
+        result = logic.add_license_to_user(license_id, user_id, wallet_address, nickname)
+        if result == "success":
+            return jsonify({'message': 'License added successfully'}), 201
+        elif result == "user_not_found":
+            return jsonify({'error': 'User not found'}), 404
+        elif result == "license_already_exists":
+            return jsonify({'error': 'License already exists'}), 409
+
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+@app.route('/api/getUserByLicense', methods=['GET'])
+@limiter.limit("10 per minute")
+@token_required
+def get_user_by_license():
+    LOGGER.info('get_user_by_license called')
+
+    license_id = request.args.get('license_id')
+
+    if not license_id:
+        return jsonify({'error': 'license_id is required'}), 400
+
+    try:
+        user_data = logic.get_user_by_license_id(license_id)
+        if user_data:
+            return jsonify(user_data), 200
+        else:
+            return jsonify({'error': 'User not found for the given license'}), 404
+
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+@app.route('/api/getLicensesByUser', methods=['GET'])
+@limiter.limit("10 per minute")
+@token_required
+def get_licenses_by_user():
+    LOGGER.info('get_licenses_by_user called')
+
+    user_id = request.args.get('user_id')
+    wallet_address = request.args.get('wallet_address')
+    nickname = request.args.get('nickname')
+
+    if sum(bool(param) for param in [user_id, wallet_address, nickname]) != 1:
+        return jsonify({'error': 'Exactly one user parameter (user_id, wallet_address, or nickname) is required'}), 400
+
+    try:
+        licenses_data = logic.get_licenses_by_user_params(user_id, wallet_address, nickname)
+        if licenses_data:
+            return jsonify(licenses_data), 200
+        else:
+            return jsonify({'error': 'No licenses found for the given user'}), 404
+
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+@app.route('/api/deleteLicense', methods=['DELETE'])
+@limiter.limit("4 per minute")
+@token_required
+def delete_license():
+    LOGGER.info('delete_license called')
+
+    license_id = request.args.get('license_id')
+
+    if not license_id:
+        return jsonify({'error': 'license_id is required'}), 400
+
+    try:
+        success = logic.delete_license_by_id(license_id)
+        if success:
+            return jsonify({'message': 'License deleted successfully'}), 200
+        else:
+            return jsonify({'error': 'License not found'}), 404
+
     except Exception as e:
         return jsonify({'error': str(e)}), 500