Step 2: Database Integration¶

Overview¶

This step extends Step 1 by integrating SQLAlchemy ORM for data storage and demonstrates LumiX’s from_model() method for direct ORM integration.

What’s New in Step 2:

SQLite database for persistent storage
SQLAlchemy declarative ORM models
LumiX’s from_model() for automatic data loading
Solution persistence to database
Cached compatibility checkers for performance

Prerequisites:

pip install lumix ortools sqlalchemy

Problem Description¶

Same as Step 1 - assign lectures to timeslots and classrooms while respecting scheduling constraints. The key difference is ORM integration:

SQLAlchemy declarative models instead of Python lists
LumiX queries database directly using from_model(session)
Solution saved back to database using ORM
Type-safe database operations with IDE support

Key Features Demonstrated¶

1. ORM Integration with SQLAlchemy¶

Using declarative models for type-safe database operations.

2. from_model() Usage¶

LumiX queries the database directly:

assignment = (
    LXVariable[Tuple[Lecture, TimeSlot, Classroom], int]("assignment")
    .binary()
    .indexed_by_product(
        LXIndexDimension(Lecture, lambda lec: lec.id).from_model(session),
        LXIndexDimension(TimeSlot, lambda ts: ts.id).from_model(session),
        LXIndexDimension(Classroom, lambda room: room.id).from_model(session),
    )
    # Filter: classroom must fit the class (using cached checker for performance)
    .where_multi(
        lambda lec, ts, room: fits_checker(lec.class_id, room.id)
    )

3. Solution Persistence¶

Save optimization results back to the database via ORM session.

4. Cached Compatibility Checker¶

Avoid redundant database queries with caching:

def create_cached_class_fits_checker(session: Session):
    """Create a cached checker function for class-classroom compatibility.

    Queries all classes and classrooms once and caches the results for
    efficient repeated lookups during variable creation. This avoids
    redundant database queries when called thousands of times in where_multi().

    Performance: Reduces from O(n) queries to O(1) lookups after initial setup.
    For a typical timetabling problem with 2,400 variable combinations, this
    reduces from 4,800 database queries to just 8 queries (4 classes + 4 classrooms).

    Args:
        session: SQLAlchemy Session

    Returns:
        A checker function with signature (class_id, classroom_id) -> bool

    Example:
        >>> checker = create_cached_class_fits_checker(session)
        >>> fits = checker(class_id=1, classroom_id=2)  # Fast cached lookup
        >>> if fits:
        ...     print("Class fits in classroom")
    """
    # Query all data once upfront
    classes_dict = {c.id: c.size for c in session.query(SchoolClass).all()}
    classrooms_dict = {r.id: r.capacity for r in session.query(Classroom).all()}

    # Return closure with cached data
    def check(class_id: int, classroom_id: int) -> bool:
        """Check if class fits in classroom using cached data."""
        class_size = classes_dict.get(class_id)
        room_capacity = classrooms_dict.get(classroom_id)
        if class_size is not None and room_capacity is not None:
            return class_size <= room_capacity
        return False

    return check

Database Schema¶

The database contains 7 tables:

teachers - Teacher information
classrooms - Classroom with capacity
classes - Student classes with size
subjects - Course subjects
lectures - Individual teaching sessions
timeslots - Available scheduling slots
schedule_assignments - Optimized schedule solutions

ORM Models¶

class Teacher(Base):
    """Teacher ORM model.

    Attributes:
        id: Primary key
        name: Teacher's name
    """
    __tablename__ = 'teachers'

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)

    def __repr__(self):
        return f"<Teacher(id={self.id}, name='{self.name}')>"


class Classroom(Base):
    """Classroom ORM model.

    Attributes:
        id: Primary key
        name: Classroom name/number
        capacity: Maximum student capacity
    """
    __tablename__ = 'classrooms'

    id = Column(Integer, primary_key=True)

Running the Example¶

Step 1: Populate Database¶

cd tutorials/timetabling/step2_database_integration
python sample_data.py

Step 2: Run Optimization¶

python timetabling_db.py

Expected Output¶

Database initialization messages
Data loading confirmation
Model building progress
Solution status
Teacher and class timetables
Solution saved to database confirmation

Code Walkthrough¶

1. Initialize Database¶

# Print timetable
print(
    f"\n{'Period':<8} {'Monday':<20} {'Tuesday':<20} {'Wednesday':<20} {'Thursday':<20} {'Friday':<20}"
)

2. Create Variables with from_model()¶

assignment = (
    LXVariable[Tuple[Lecture, TimeSlot, Classroom], int]("assignment")
    .binary()
    .indexed_by_product(
        LXIndexDimension(Lecture, lambda lec: lec.id).from_model(session),
        LXIndexDimension(TimeSlot, lambda ts: ts.id).from_model(session),
        LXIndexDimension(Classroom, lambda room: room.id).from_model(session),
    )
    # Filter: classroom must fit the class (using cached checker for performance)
    .where_multi(
        lambda lec, ts, room: fits_checker(lec.class_id, room.id)
    )

3. Build Constraints (Same as Step 1)¶

Constraint logic remains the same, but data comes from database instead of Python lists.

4. Save Solution to Database¶

def save_solution_to_db(solution, session):
    """Save the optimization solution to the database using ORM.

    Extracts all schedule assignments from the solution and stores
    them in the database using SQLAlchemy ORM.

    Args:
        solution: LXSolution object.
        session: SQLAlchemy Session instance.
    """
    if not solution.is_optimal() and not solution.is_feasible():
        print("\n❌ No solution to save!")
        return

    print("\nSaving solution to database using ORM...")

    # Delete existing assignments
    session.query(ScheduleAssignment).delete()

    count = 0
    assignments = []
    for (lecture_id, timeslot_id, classroom_id), value in solution.variables[
        "assignment"
    ].items():
        if value > 0.5:  # Binary variable is 1
            assignment = ScheduleAssignment(
                lecture_id=lecture_id,
                timeslot_id=timeslot_id,
                classroom_id=classroom_id,
            )
            assignments.append(assignment)
            count += 1

    session.add_all(assignments)
    session.commit()
    print(f"  Saved {count} schedule assignments")

Key Learnings¶

Benefits of ORM Integration¶

Type Safety: IDE autocomplete for model attributes
Data Persistence: Solutions saved automatically
Scalability: Handle larger datasets efficiently
Maintainability: Schema changes managed via migrations
Testability: Mock database sessions for unit tests

Performance Optimization¶

The cached compatibility checker provides significant performance improvements:

Without caching: O(n) database queries per variable
With caching: O(1) lookups after initial load
Speedup: 100x+ for large problems

Next Steps¶

After completing Step 2, proceed to:

Step 3 (Step 3: Goal Programming with Teacher Preferences) - Add teacher preferences using goal programming
Step 4 (Step 4: Large-Scale Optimization with Room Types) - Scale to production-ready size