This file contains the inital Orders management system microservice

Order_Microservice_Group3/app/__init__.py

+from flask import Flask
+
+from .controllers.AddToCartController import AddToCart_bp
+from .controllers.CheckOutController import checkout_bp
+from .controllers.DeleteFromItemController import DeleteCart_bp
+from .controllers.FetchCartController import cart_bp
+
+def create_app():
+    app = Flask(__name__)
+    app.register_blueprint(AddToCart_bp)
+    app.register_blueprint(checkout_bp)
+    app.register_blueprint(DeleteCart_bp)
+    app.register_blueprint(cart_bp)
+    return app
\ No newline at end of file

Order_Microservice_Group3/app/config.py

+import os
+
+DEBUG = True
+SECRET_KEY = "Group3"
+
+PORT = 5001
+
+KAFKA_SERVER= "localhost:9092"

Order_Microservice_Group3/app/controllers/AddToCartController.py

+from flask import Blueprint, jsonify, request, json, session
+from app.models.AddToCart import add_item_shopping_cart
+import requests
+
+AddToCart_bp = Blueprint("AddToCart", __name__)
+
+# @AddToCart_bp.route("/cart/AddToCart", methods=["POST"])
+# def add_item():
+#     user_id = session.get("user_id")
+
+#     if user_id:
+#         data = request.get_json()
+#         ProductID = data.get("ProductID")
+#         CartQuantity = data.get("CartQuantity")
+#         UnitPrice = data.get("UnitPrice")
+
+#         if not ProductID or not CartQuantity:
+#             return jsonify({"error": "Product ID and quantity are required."}), 400
+
+#         cart_item_data = {
+#             "UserID": user_id,
+#             "ProductID": ProductID,
+#             "UnitPrice": UnitPrice,
+#             CartQuantity": CartQuantity
+#         }
+
+#         add_cart_item_message = add_item_shopping_cart(cart_item_data)
+#         return jsonify(add_cart_item_message)
+
+#     else:
+#         return jsonify({"error": "You need to be logged in to add items to the cart."}), 
+
+#This one to test
+
+@AddToCart_bp.route("/cart/AddToCart", methods=["POST"])
+def add_item():
+    
+    user_id = 1  
+    data = request.get_json()
+    ProductID = data.get("ProductID")
+    CartQuantity = data.get("CartQuantity")
+    UnitPrice = data.get("UnitPrice")
+
+    if not ProductID or not CartQuantity or UnitPrice is None:
+        return jsonify({"error": "Product ID, quantity, and unit price are required."}), 400
+
+    cart_item_data = {
+        "UserID": user_id,
+        "ProductID": ProductID,
+        "UnitPrice": UnitPrice,
+        "CartQuantity": CartQuantity
+    }
+
+    add_cart_item_message = add_item_shopping_cart(cart_item_data)
+    return jsonify(add_cart_item_message)

Order_Microservice_Group3/app/controllers/CheckOutController.py

+from flask import Blueprint, request, jsonify, session
+from app.models.CheckOut import checkout_cart  
+from app.models.FetchKart import get_user_cart_items
+
+checkout_bp = Blueprint('checkout', __name__)
+
+@checkout_bp.route('/cart/checkout', methods=['POST'])
+def handle_checkout():
+
+    user_id = 1
+    if not user_id:
+        return jsonify({"error": "User not logged in"}), 401
+
+    data = request.get_json()
+    
+    if not data or 'cart_items' not in data or 'location' not in data:
+        return jsonify({"error": "Missing cart items or location"}), 400
+
+    cart_items = data['cart_items']
+    location = data['location']
+    
+    total_price = data.get('total_price', 0)
+
+    checkout_result = checkout_cart(user_id, cart_items, location, total_price)
+
+    if 'error' in checkout_result:
+        return jsonify({"error": checkout_result["error"]}), 500
+
+    return jsonify(checkout_result), 200
\ No newline at end of file

Order_Microservice_Group3/app/controllers/DeleteFromItemController.py

+from app.models.DeleteItemCartModel import delete_item_from_cart
+from flask import Blueprint, jsonify, request, json, session
+
+DeleteCart_bp = Blueprint("delete", __name__)
+
+# @DeleteCart_bp.route("/cart/delete", methods=['DELETE'])
+# def delete_item(cart_item_id):
+#     user_id = session.get("user_id")
+#     if not user_id:
+#         return jsonify({"error": "You need to be logged in to delete items from the cart."}), 401
+
+#     # Call the function to delete the item from the cart
+#     result = delete_item_from_cart(user_id, cart_item_id)
+
+#     if "error" in result:
+#         return jsonify(result), 404  
+#     return jsonify(result), 200
+
+@DeleteCart_bp.route("/cart/delete/<int:cart_item_id>", methods=['DELETE'])
+def delete_item(cart_item_id):
+    user_id = 5
+    if not user_id:
+        return jsonify({"error": "You need to be logged in to delete items from the cart."}), 401
+
+    # Call the function to delete the item from the cart
+    result = delete_item_from_cart(user_id, cart_item_id)
+
+    if "error" in result:
+        return jsonify(result), 404  
+    return jsonify(result), 200
\ No newline at end of file

Order_Microservice_Group3/app/controllers/FetchCartController.py

+from flask import Blueprint, jsonify, request, json, session
+from app.models.FetchKart import get_user_cart_items
+import requests
+
+# cart_bp = Blueprint("cart", __name__)
+
+# @cart_bp.route("/cart", methods=["GET"])
+# def fetch_user_cart():
+#     user_id = session.get("user_id")
+
+#     if not user_id:
+#         return jsonify({"error": "You need to be logged in to view the cart."}), 401
+
+#     cart_data = get_user_cart_items(user_id)
+#     if "error" in cart_data:
+#         return jsonify({"error": cart_data["error"]}),
+
+#     return jsonify(cart_data)
+
+#This code is to test
+
+cart_bp = Blueprint("cart", __name__)
+
+@cart_bp.route("/cart", methods=["GET"])
+def fetch_user_cart():
+    
+    user_id = request.args.get("user_id")
+
+    if not user_id:
+        return jsonify({"error": "You need to be logged in to view the cart."}), 400
+
+    cart_data = get_user_cart_items(user_id)
+    if "error" in cart_data:
+        return jsonify({"error": cart_data["error"]}), 500
+
+    return jsonify(cart_data)
\ No newline at end of file

Order_Microservice_Group3/app/models/AddToCart.py

+import pyodbc
+from flask import jsonify
+
+import sys
+sys.path.append("c:/Users/Fin/University of surrey/Advanced Web Technology/advanced-web-project/Orders_Microservice_Group3")
+
+from app.models.database_connection import connect_db
+
+def add_item_shopping_cart(data):
+
+    try: #error handling
+        connection = connect_db()
+        cursor = connection.cursor()
+
+        insert_cart_item_query = '''INSERT INTO dbo.ShoppingCart (UserID, ProductID, UnitPrice, CartQuantity)
+VALUES (?, ?, ?, ?);'''
+        cursor.execute(insert_cart_item_query, (data["UserID"], data["ProductID"], data["UnitPrice"], data["CartQuantity"]))
+ 
+        connection.commit()
+
+        return {"message": "Item added to cart successfully"}
+
+    except pyodbc.Error as e:
+        print(f"Database error in add_item_to_cart: {e}")
+        connection.rollback()
+        return {"Error": f"Unexpected error: {str(e)}"}
+
+    except Exception as e:
+        print(f"Unexpected error occurred in add_item_to_cart: {e}")
+        connection.rollback()
+        return {"Error": "Unexpected error"}
+
+    finally:
+        if cursor:
+            cursor.close()
+        if connection:
+            connection.close()
+

Order_Microservice_Group3/app/models/CheckOut.py

+from app.models.database_connection import connect_db
+import pyodbc
+from datetime import datetime
+
+# def checkout_cart(user_id, address, delivery):
+#     try:
+#         connection = connect_db()
+#         # Ensure that all operations are executed as a single transaction
+#         connection.autocommit = False
+#         cursor = connection.cursor()
+
+#         # Fetch cart items for the user
+#         cursor.execute("SELECT ProductID, CartQuantity, UnitPrice, CartQuantity * UnitPrice AS TotalItemPrice FROM ShoppingCart WHERE UserID = ?", (user_id,))
+#         cart_items = cursor.fetchall()
+
+#         if not cart_items:
+#             return {"error": "No items in the cart to checkout."}
+
+#         # Insert a new order group
+#         total_price = sum(item.TotalItemPrice for item in cart_items)
+#         cursor.execute("INSERT INTO OrderGroup (Transaction_date, Address, Delivery, CustomerID, TotalPrice) VALUES (?, ?, ?, ?, ?)", (datetime.now(), address, delivery, user_id, total_price))
+#         transaction_id = cursor.lastrowid  
+
+#         # Insert cart items into OrderItem table and link to the new order group
+#         for item in cart_items:
+#             cursor.execute("INSERT INTO OrderItem (Transaction_ID, StockCode, OrdersStatus, Quantity, TotalItemPrice) VALUES (?, ?, ?, ?, ?)", (transaction_id, item.ProductID, 'Processing', item.CartQuantity, item.TotalItemPrice))
+
+#         # Delete the cart items
+#         cursor.execute("DELETE FROM ShoppingCart WHERE UserID = ?", (user_id,))
+
+#         # Commit the transaction
+#         connection.commit()
+
+#         return {"success": "Checkout completed successfully.", "TransactionID": transaction_id}
+
+#     except pyodbc.Error as e:
+#         print(f"Database error during checkout: {e}")
+#         connection.rollback()  # Rollback in case of error
+#         return {"error": "Database error during checkout"}
+
+#     finally:
+#         if cursor:
+#             cursor.close()
+#         if connection:
+#             connection.close()
+
+
+def checkout_cart(user_id, cart_items, Location, total_price):
+    try:
+        connection = connect_db()
+        # Ensures all operations are executed as a single transaction
+        connection.autocommit = False
+        cursor = connection.cursor()
+
+        insert_order_group_stmt = """
+            INSERT INTO OrderGroup (CustomerID, TotalPrice, TransactionDate, Location, OrdersStatus)
+            OUTPUT INSERTED.OrderGroupID
+            VALUES (?, ?, ?, ?, ?)
+        """
+        cursor.execute(insert_order_group_stmt, (user_id, total_price, datetime.now(), Location, 'Processing'))
+        OrderGroupID = cursor.fetchone()[0]  
+
+        # Insert cart items into the OrderItem table and link them to the new order group
+        for item in cart_items:
+            cursor.execute("""
+                INSERT INTO OrderItem (OrderGroupID,  ProductID, Quantity, UnitPrice, TotalItemPrice)
+                VALUES (?, ?, ?, ?, ?)
+            """, (OrderGroupID, item['ProductID'], item['CartQuantity'], item['UnitPrice'], item['TotalItemPrice']))
+
+        # Delete the cart items from the ShoppingCart table
+        cursor.execute("DELETE FROM ShoppingCart WHERE UserID = ?", (user_id,))
+
+        # Commit the transaction
+        connection.commit()
+
+        return {"success": "Checkout completed successfully.", "OrderID": OrderGroupID}
+
+    except pyodbc.Error as e:
+        print(f"Database error during checkout: {e}")
+        connection.rollback()  # Rollback in case of an error
+        return {"error": "Database error during checkout"}
+
+    finally:
+        if cursor:
+            cursor.close()
+        if connection:
+            connection.close()
+
+

Order_Microservice_Group3/app/models/DeleteItemCartModel.py

+from flask import jsonify
+from app.models.database_connection import connect_db
+import pyodbc
+
+def delete_item_from_cart(user_id, cart_item_id):
+
+    try:  # error handling
+        connection = connect_db()
+        cursor = connection.cursor()
+
+        # SQL query to delete the item from the cart based on cart_item_id and UserID
+        delete_cart_item_query = '''DELETE FROM dbo.ShoppingCart WHERE UserID = ? AND CartItemID = ?;'''
+        cursor.execute(delete_cart_item_query, (user_id, cart_item_id))
+
+        # Committing the transaction to the database
+        connection.commit()
+
+        # Check if the delete operation was successful
+        if cursor.rowcount == 0:
+            return {"error": "Item not found or could not be deleted"}
+
+        return {"message": "Item deleted successfully"}
+
+    except pyodbc.Error as e:
+        print(f"Database error in delete_item_from_cart: {e}")
+        # Rollback the transaction in case of an error
+        connection.rollback()
+        return {"error": "Database error during item deletion"}
+
+    except Exception as e:
+        print(f"Unexpected error occurred in delete_item_from_cart: {e}")
+        # Rollback the transaction in case of an unexpected error
+        connection.rollback()
+        return {"error": "Unexpected error during item deletion"}
+
+    finally:
+        if cursor:
+            cursor.close()
+        if connection:
+            connection.close()
\ No newline at end of file

Order_Microservice_Group3/app/models/FetchKart.py

+import pyodbc
+from datetime import datetime
+from app.models.database_connection import connect_db
+
+def checkout_cart(user_id, cart_items, Location, total_price):
+    try:
+        connection = connect_db()
+        # Ensures all operations are executed as a single transaction
+        connection.autocommit = False
+        cursor = connection.cursor()
+
+        # Insert a new order group and capture the Transaction_ID
+        insert_order_group_stmt = """
+            INSERT INTO OrderGroup (CustomerID, TotalPrice, TransactionDate, Location)
+            OUTPUT INSERTED.OrderGroupID
+            VALUES (?, ?, ?, ?, ?)
+        """
+        cursor.execute(insert_order_group_stmt, (user_id, total_price, datetime.now(), Location))
+        OrderGroupID = cursor.fetchone()[0]  # Fetch the Transaction_ID 
+
+        # Insert cart items into the OrderItem table and link them to the new order group
+        for item in cart_items:
+            cursor.execute("""
+                INSERT INTO OrderItem (OrderGroupID, StockCode, OrdersStatus, Quantity, TotalItemPrice)
+                VALUES (?, ?, ?, ?, ?)
+            """, (OrderGroupID, item['ProductID'], 'Processing', item['CartQuantity'], item['TotalItemPrice']))
+
+        # Delete the cart items from the ShoppingCart table
+        cursor.execute("DELETE FROM ShoppingCart WHERE UserID = ?", (user_id,))
+
+        # Commit the transaction
+        connection.commit()
+
+        return {"success": "Checkout completed successfully.", "OrderID": OrderGroupID}
+
+    except pyodbc.Error as e:
+        print(f"Database error during checkout: {e}")
+        connection.rollback()  # Rollback in case of an error
+        return {"error": "Database error during checkout"}
+
+    finally:
+        if cursor:
+            cursor.close()
+        if connection:
+            connection.close()
+
+
+
+
+
+
+
+
+
+
+
+
+
+from app.models.database_connection import connect_db
+from flask import jsonify
+import pyodbc
+
+def get_user_cart_items(user_id):
+    try:
+        connection = connect_db()
+        cursor = connection.cursor()
+
+        query = "SELECT * FROM ShoppingCart WHERE UserID = ?"
+        cursor.execute(query, (user_id,))
+
+        # Fetch all rows for the user_id
+        rows = cursor.fetchall()
+
+        # Prepare a list of dictionaries to hold cart item data
+        cart_items = []
+        Total_Price = 0
+        for row in rows:
+            Total_item_price = row.CartQuantity * row.UnitPrice
+            cart_item = {
+                "CartItemID": row.CartItemID, 
+                "UserID": row.UserID,
+                "ProductID": row.ProductID,
+                "UnitPrice": row.UnitPrice,
+                "CartQuantity": row.CartQuantity,
+                "AddedDate": row.AddedDate,
+                "TotalItemPrice": Total_item_price
+            }
+            cart_items.append(cart_item)
+            Total_Price += Total_item_price
+
+        return {"CartItems": cart_items, "TotalPrice": Total_Price}
+
+    except pyodbc.Error as e:
+        print(f"Database error in get_user_cart_items: {e}")
+        return {"error": "Database error"}
+
+    finally:
+        if cursor:
+            cursor.close()
+        if connection:
+            connection.close()
+
+
+
+
+#for interacting with the message broker, also i need to do it for price
+
+# def can_fulfill_order(user_id):
+#     try:
+#         connection = connect_db()
+#         cursor = connection.cursor()
+
+#         # Assuming you have a function that gets the latest stock level for a given product ID
+#         def get_stock_level(product_id):
+#             # This function should query your cache or database that is kept up to date with Kafka messages
+#             # For demonstration, let's assume it queries a database table `ProductStock`
+#             cursor.execute("SELECT StockLevel FROM ProductStock WHERE ProductID = ?", (product_id,))
+#             stock_info = cursor.fetchone()
+#             return stock_info.StockLevel if stock_info else 0
+
+#         query = "SELECT * FROM ShoppingCart WHERE UserID = ?"
+#         cursor.execute(query, (user_id,))
+#         rows = cursor.fetchall()
+
+#         cart_items = []
+#         Total_Price = 0
+#         for row in rows:
+#             current_stock_level = get_stock_level(row.ProductID)
+#             if row.CartQuantity > current_stock_level:
+#                 # If any item exceeds the available stock, return immediately indicating the order cannot be fulfilled
+#                 return {"error": f"Cannot fulfill order for ProductID {row.ProductID}. Only {current_stock_level} left in stock."}
+
+#             Total_item_price = row.CartQuantity * row.UnitPrice
+#             cart_item = {
+#                 "CartItemID": row.CartItemID, 
+#                 "UserID": row.UserID,
+#                 "ProductID": row.ProductID,
+#                 "UnitPrice": row.UnitPrice,
+#                 "CartQuantity": row.CartQuantity,
+#                 "AddedDate": row.AddedDate,
+#                 "TotalItemPrice": Total_item_price
+#             }
+#             cart_items.append(cart_item)
+#             Total_Price += Total_item_price
+
+#         # If all items can be fulfilled, return the cart items and total price
+#         return {"CartItems": cart_items, "TotalPrice": Total_Price}
+
+#     except pyodbc.Error as e:
+#         print(f"Database error in can_fulfill_order: {e}")
+#         return {"error": "Database error"}
+
+#     finally:
+#         if cursor:
+#             cursor.close()
+#         if connection:
+#             connection.close()
\ No newline at end of file

Order_Microservice_Group3/app/models/UpdateProduct.py

+#from confluent_kafka import Producer
+import json
+import pyodbc
+#from flask import jsonify
+from database_connection import connect_db
+
+def UpdateProduct(product_id, quantity, price):
+    try:
+        connection = connect_db()
+        connection.autocommit = False
+        cursor = connection.cursor()
+
+        # SQL MERGE statement to insert or update
+        sql = """
+        MERGE INTO ProductData AS target
+        USING (SELECT ? AS ProductID, ? AS QuantityAvailable, ? AS UnitPrice) AS source
+        ON target.ProductID = source.ProductID
+        WHEN MATCHED THEN
+            UPDATE SET QuantityAvailable = source.QuantityAvailable, UnitPrice = source.UnitPrice
+        WHEN NOT MATCHED BY TARGET THEN
+            INSERT (ProductID, QuantityAvailable, UnitPrice)
+            VALUES (source.ProductID, source.QuantityAvailable, source.UnitPrice);
+        """
+        cursor.execute(sql, (product_id, quantity, price))
+        connection.commit()
+        return {"success": "Product information updated successfully."}
+
+    except pyodbc.Error as e:
+        print(f"Database error during product update: {e}")
+        connection.rollback()
+        return {"error": f"Database error during product update: {e}"}
+
+    except Exception as e:
+        print(f"Unexpected error occurred: {e}")
+        connection.rollback()
+        return {"error": f"Unexpected error: {e}"}
+
+    finally:
+        if cursor:
+            cursor.close()
+        if connection:
+            connection.close()
+

Order_Microservice_Group3/app/models/__init__.py

+from flask import Flask
+from flask_cors import CORS
+
+
+#from app.models import models
+
+app = Flask(__name__)
+CORS(app)
+
+#db = sqlAlchemy
+
+#from app import routes
+
+if __name__ == '__main__':
+    app.run(debug=True)

Order_Microservice_Group3/app/models/database_connection.py

+import pyodbc
+
+#Connect to database
+def connect_db():
+
+    server = 
+    database = 
+
+    connection_string = f'DRIVER={{SQL Server}};SERVER={server};DATABASE={database};Trusted_Connection=yes;'
+    
+    return pyodbc.connect(connection_string)
\ No newline at end of file

Order_Microservice_Group3/app/publishers/kafkaPublishers.py

+from kafka import KafkaProducer
+from kafka.admin import KafkaAdminClient, NewTopic
+
+import json
+
+producer = KafkaProducer(bootstrap_servers="localhost:9092")
+
+
+#Creates the topic
+def create_Quantity_updated_topic():
+    # Create KafkaAdminClient instance
+    admin_client = KafkaAdminClient(bootstrap_servers="localhost:9092")
+
+    # Define the topic name and configuration
+    topic_name = "QuantityUpdateFromOrders"
+    num_partitions = 1
+    replication_factor = 1
+
+    # Retrieve the list of existing topics
+    topic_metadata = admin_client.list_topics()
+
+    # Check if the topic already exists
+    if topic_name not in topic_metadata:
+        # Define the configuration for the new topic
+        new_topic = NewTopic(name=topic_name, num_partitions=num_partitions, replication_factor=replication_factor)
+        # Create the new topic
+        admin_client.create_topics(new_topics=[new_topic], validate_only=False)
+
+
+#Function is called in updateProfileControllers.py
+def publish_product_updated_event(event_data):
+    # Serialize the event data to JSON
+    event_json = json.dumps(event_data)
+    # Publish the event to the Kafka topic
+    data_to_send = producer.send("QuantityUpdateFromOrders", value=event_json.encode("utf-8"))
+    try:
+        record_metadata = data_to_send.get(timeout=10)
+        print("Message sent successfully!")
+        print("Topic:", record_metadata.topic)
+        print("Partition:", record_metadata.partition)
+        print("Offset:", record_metadata.offset)
+    except Exception as e:
+        print("Failed to send message:", e)
+    producer.flush()

Order_Microservice_Group3/app/subscriber/UpdateProductSubscriber.py

+
+
+
+import sys
+print(sys.path)
+from threading import Thread
+from kafka import KafkaConsumer
+import json
+import logging
+KAFKA_SERVER= "localhost:9092"
+
+from  app.models.UpdateProduct import UpdateProduct
+
+def consume_product_updated_event():
+    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+    logging.info("Consuming product updated event...")
+    
+    # Create KafkaConsumer listening on the product_update_topic
+    consumer = KafkaConsumer("product_update_topic", bootstrap_servers=KAFKA_SERVER)
+    
+    for message in consumer:
+        product_data_str = message.value.decode("utf-8")
+        product_data = json.loads(product_data_str)
+        logging.info("Received product update: {}".format(product_data))
+        
+        # Call your UpdateProduct function passing the dictionary containing product details
+        UpdateProduct(product_id=product_data['product_id'], 
+                      quantity=product_data['quantity'], 
+                      price=product_data['price'])
+
+def start_kafka_consumer():
+    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+    logging.info("Starting Kafka consumer...")
+
+    # Starting the consumer in a separate thread
+    kafka_thread = Thread(target=consume_product_updated_event)
+    kafka_thread.daemon = True  
+    kafka_thread.start()
\ No newline at end of file

Order_Microservice_Group3/app/subscriber/import pyodbc.txt

+import pyodbc
+from datetime import datetime
+from app.models.database_connection import connect_db
+from your_kafka_module import kafka_produce  # Import the Kafka producer function
+
+def order_service_consumer():
+    consumer = Consumer({
+        'bootstrap.servers': 'localhost:9092',
+        'group.id': 'orders-group',
+        'auto.offset.reset': 'earliest'
+    })
+    consumer.subscribe(['product_changes'])
+
+    while True:
+        message = consumer.poll(1.0)
+        if message is None:
+            continue
+        if message.error():
+            print("Consumer error: {}".format(message.error()))
+            continue
+
+        data = json.loads(message.value().decode('utf-8'))
+        if data['type'] == 'quantity_update':
+            # Update orders database based on quantity change
+            update_orders_database(data['product_id'], data['quantity_sold'])
+
+def update_orders_database(product_id, quantity_sold):
+    connection = connect_db()
+    try:
+        cursor = connection.cursor()
+        # Update logic here, e.g., flagging orders as fulfilled
+        cursor.execute("UPDATE Orders SET Status = 'Updated' WHERE product_id = ?", (product_id,))
+        connection.commit()
+    finally:
+        cursor.close()
+        connection.close()
+
+
+
+
+
+
+
+
+kafka-console-producer.bat --broker-list localhost:9092 --topic product_update_topic

Order_Microservice_Group3/requirements.txt

+flask
+requests
+random
\ No newline at end of file