Python program: Aggregation queries with SQLAlchemy

Python SQLAlchemy: Exercise-14 with Solution

Write a Python program that performs aggregation queries to get the total number of orders, total sales amount, minimum item price, maximum item price and average order quantity using SQLAlchemy.

Sample Solution:

Code:

from sqlalchemy import create_engine, Column, Integer, Float, String, ForeignKey
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy import func
# Create a SQLite database named shop.db
engine = create_engine('sqlite:///shop2.db', echo=False)
# Create a base class for declarative models
Base = declarative_base()
# Define the User model
class User(Base):
    __tablename__ = 'users'  
    user_id = Column(Integer, primary_key=True)
    user_name = Column(String, nullable=False)
    user_email = Column(String, nullable=False)
# Define the Item model
class Item(Base):
    __tablename__ = 'items'    
    item_id = Column(Integer, primary_key=True)
    item_name = Column(String, nullable=False)
    item_price = Column(Float, nullable=False)
    item_quantity = Column(Integer, nullable=False)    
    orders = relationship("Order", back_populates="item")  # Define the relationship to Order
# Define the Order model
class Order(Base):
    __tablename__ = 'orders'    
    order_id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.user_id'), nullable=False)
    item_id = Column(Integer, ForeignKey('items.item_id'), nullable=False)
    order_quantity = Column(Integer, nullable=False)    
    item = relationship("Item", back_populates="orders")  # Define the relationship to Item
# Create a session to interact with the database
Session = sessionmaker(bind=engine)
session = Session()

# Aggregation queries
def perform_aggregation_queries():
    # Count the number of orders
    orders_count = session.query(func.count(Order.order_id)).scalar()
    print(f"Total number of orders: {orders_count}")
    # Calculate the total sales amount
    total_sales = session.query(func.sum(Order.order_quantity * Item.item_price)).scalar()
    print(f"Total sales amount: ${total_sales:.2f}")
    # Find the minimum and maximum item price
    min_item_price = session.query(func.min(Item.item_price)).scalar()
    max_item_price = session.query(func.max(Item.item_price)).scalar()
    print(f"Minimum item price: ${min_item_price:.2f}")
    print(f"Maximum item price: ${max_item_price:.2f}")
    # Calculate the average order quantity
    avg_order_quantity = session.query(func.avg(Order.order_quantity)).scalar()
    print(f"Average order quantity: {avg_order_quantity:.2f}")
# Perform aggregation queries
perform_aggregation_queries()
# Close the session
session.close()

Output:

Total number of orders: 3
Total sales amount: $343.62
Minimum item price: $0.45
Maximum item price: $15.82
Average order quantity: 6.00

Flowchart:

Previous: Calculate total order cost with SQLAlchemy.