File size: 3,708 Bytes
588f657 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
from sqlalchemy.orm import Session
from database import Product, Transaction
import uuid
from datetime import datetime
def add_product(db: Session, name: str, current_stock: float, safety_stock: float, lead_time: float, monthly_demand: float, product_class: str):
product = Product(name=name, current_stock=current_stock, safety_stock=safety_stock, lead_time=lead_time, monthly_demand=monthly_demand, product_class=product_class) # Create new product
db.add(product)
db.commit()
db.refresh(product)
return product
def get_all_products(db: Session):
return db.query(Product).all() # Fetch all products
def get_product_by_id(db: Session, product_id: int):
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise ValueError("Product not found")
return product
# In-memory session-based functions
def add_product_memory(session_products: list, name: str, current_stock: float, safety_stock: float, lead_time: float, monthly_demand: float, product_class: str):
if any(p["name"].lower() == name.lower() for p in session_products):
raise ValueError("Product name already exists in your session")
product = {"id": str(uuid.uuid4()), "name": name, "current_stock": current_stock, "safety_stock": safety_stock, "lead_time": lead_time, "monthly_demand": monthly_demand, "product_class": product_class} # Create new in-memory product
session_products.append(product)
return product
def update_product_memory(session_products: list, product_id: str, name: str, current_stock: float, safety_stock: float, lead_time: float, monthly_demand: float, product_class: str):
if any(p["name"].lower() == name.lower() and p["id"] != product_id for p in session_products):
raise ValueError("Product name already exists in your session")
for product in session_products:
if product["id"] == product_id:
product.update({"name": name, "current_stock": current_stock, "safety_stock": safety_stock, "lead_time": lead_time, "monthly_demand": monthly_demand, "product_class": product_class})
return product
raise ValueError("Product not found")
def delete_product_memory(session_products: list, product_id: str):
for i, product in enumerate(session_products):
if product["id"] == product_id:
session_products.pop(i)
return
raise ValueError("Product not found")
def add_stock_memory(session_products: list, session_transactions: list, product_id: str, quantity: float):
for product in session_products:
if product["id"] == product_id:
product["current_stock"] += quantity
session_transactions.append({"id": str(uuid.uuid4()), "product_id": product_id, "quantity": quantity, "transaction_type": "ADD", "timestamp": datetime.now()}) # Record stock addition
return
raise ValueError("Product not found")
def remove_stock_memory(session_products: list, session_transactions: list, product_id: str, quantity: float):
for product in session_products:
if product["id"] == product_id:
if product["current_stock"] < quantity:
raise ValueError("Insufficient stock")
product["current_stock"] -= quantity
session_transactions.append({"id": str(uuid.uuid4()), "product_id": product_id, "quantity": quantity, "transaction_type": "REMOVE", "timestamp": datetime.now()}) # Record stock removal
return
raise ValueError("Product not found")
def get_transactions_memory(session_transactions: list):
return session_transactions # Fetch all transactions |