SQLAlchemy Fundamentals: Building a Simple Inventory System

This challenge will guide you through the fundamental concepts of SQLAlchemy, a powerful Python SQL toolkit and Object-Relational Mapper (ORM). You will create a basic inventory management system, learning how to define database models, establish connections, and perform common CRUD (Create, Read, Update, Delete) operations. Mastering these basics is crucial for efficiently managing data in Python applications.

Problem Description

Your task is to implement a simple inventory system using SQLAlchemy. This involves:

1. Defining Database Models: Create Python classes that represent your database tables. For this challenge, you'll focus on a single table: Products.
2. Setting up the Database: Configure SQLAlchemy to connect to an in-memory SQLite database for simplicity.
3. Creating Tables: Use SQLAlchemy to create the Products table in the database based on your defined model.
4. Performing CRUD Operations: Implement functions to:
   • Add new products to the inventory.
   • Retrieve all products from the inventory.
   • Update the quantity of an existing product.
   • Delete a product from the inventory.

Key Requirements:

• Use SQLAlchemy Core for defining the table structure.
• Use SQLAlchemy ORM for mapping Python classes to database tables.
• The Products table should have the following columns:
  • id: Integer, Primary Key, Auto-incrementing.
  • name: String, Not Null, Unique.
  • quantity: Integer, Not Null, Default to 0.
  • price: Numeric (Decimal), Not Null.
• The database should be an in-memory SQLite database (e.g., sqlite:///:memory:).
• Implement functions for add_product, get_all_products, update_product_quantity, and delete_product.

Expected Behavior:

• After setup, the Products table should be created.
• add_product should successfully insert a new product.
• get_all_products should return a list of product objects.
• update_product_quantity should modify the quantity of a specified product.
• delete_product should remove a specified product.

Edge Cases:

• Adding a product with a name that already exists should raise an error or be handled gracefully (e.g., print a message). For this challenge, assume names are unique and the database constraint will handle this.
• Attempting to update or delete a product that does not exist should be handled gracefully (e.g., print a message, return False).

Examples

Example 1: Adding and Retrieving Products

from sqlalchemy import create_engine, Column, Integer, String, Numeric
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from decimal import Decimal

# Assume the setup and Product model are defined as per requirements

# Setup (simplified for example)
engine = create_engine('sqlite:///:memory:')
Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False, unique=True)
    quantity = Column(Integer, nullable=False, default=0)
    price = Column(Numeric, nullable=False)

Base.metadata.create_all(engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# --- Your implementation functions would go here ---
def add_product(session, name: str, quantity: int, price: Decimal):
    new_product = Product(name=name, quantity=quantity, price=price)
    session.add(new_product)
    session.commit()
    session.refresh(new_product)
    print(f"Added product: {new_product.name} (ID: {new_product.id})")
    return new_product

def get_all_products(session):
    return session.query(Product).all()
# --- End of your implementation functions ---

session = SessionLocal()

# Add some products
add_product(session, "Laptop", 10, Decimal("1200.50"))
add_product(session, "Mouse", 50, Decimal("25.00"))

# Retrieve all products
products = get_all_products(session)

for product in products:
    print(f"ID: {product.id}, Name: {product.name}, Quantity: {product.quantity}, Price: {product.price}")

session.close()

Output:

Added product: Laptop (ID: 1)
Added product: Mouse (ID: 2)
ID: 1, Name: Laptop, Quantity: 10, Price: 1200.50
ID: 2, Name: Mouse, Quantity: 50, Price: 25.00

Explanation:

The code first sets up the database and the Product model. Then, add_product is used to insert two products. Finally, get_all_products retrieves these products, and they are printed to the console.

Example 2: Updating and Deleting Products

# ... (Assume setup from Example 1 is present, including add_product and get_all_products)
from sqlalchemy.orm import Session

# --- Your implementation functions ---
def update_product_quantity(session: Session, product_id: int, new_quantity: int) -> bool:
    product = session.query(Product).filter(Product.id == product_id).first()
    if product:
        product.quantity = new_quantity
        session.commit()
        session.refresh(product)
        print(f"Updated quantity for '{product.name}' to {new_quantity}")
        return True
    else:
        print(f"Product with ID {product_id} not found.")
        return False

def delete_product(session: Session, product_id: int) -> bool:
    product = session.query(Product).filter(Product.id == product_id).first()
    if product:
        session.delete(product)
        session.commit()
        print(f"Deleted product: {product.name} (ID: {product.id})")
        return True
    else:
        print(f"Product with ID {product_id} not found.")
        return False
# --- End of your implementation functions ---

session = SessionLocal()

# Add products (if not already added from Example 1)
if not get_all_products(session):
    add_product(session, "Keyboard", 20, Decimal("75.99"))
    add_product(session, "Monitor", 5, Decimal("300.00"))

# Update quantity
update_product_quantity(session, 1, 15) # Assuming Laptop has ID 1

# Attempt to update non-existent product
update_product_quantity(session, 99, 10)

# Delete a product
delete_product(session, 2) # Assuming Mouse has ID 2

# Retrieve and print remaining products
products = get_all_products(session)
print("\nCurrent inventory:")
for product in products:
    print(f"ID: {product.id}, Name: {product.name}, Quantity: {product.quantity}, Price: {product.price}")

session.close()

Output:

Updated quantity for 'Laptop' to 15
Product with ID 99 not found.
Deleted product: Mouse (ID: 2)

Current inventory:
ID: 1, Name: Laptop, Quantity: 15, Price: 1200.50
ID: 3, Name: Keyboard, Quantity: 20, Price: 75.99
ID: 4, Name: Monitor, Quantity: 5, Price: 300.00

Explanation:

This example demonstrates updating the quantity of an existing product and handling an attempt to update a non-existent one. It also shows how to delete a product and then lists the remaining inventory.

Constraints

• The entire database interaction must be performed using SQLAlchemy.
• Use an in-memory SQLite database (sqlite:///:memory:).
• The solution should be a single Python script.
• Functions should be well-defined and take the SQLAlchemy Session object as an argument where appropriate for performing database operations.
• The solution should be able to handle at least 100 product entries without significant performance degradation for basic CRUD operations.

Notes

• Start by importing necessary components from sqlalchemy.
• Define your database engine first.
• Use declarative_base to create a base class for your ORM models.
• Create a sessionmaker to generate Session objects.
• Remember to commit your changes to the database session.
• Consider the return values of your functions to indicate success or failure, especially for update and delete operations.
• For uniqueness constraints (like product name), SQLAlchemy will raise an IntegrityError if violated. You don't necessarily need to explicitly check before inserting, but be aware of this.