Querying for overlapping time ranges in SQLAlchemy and Postgres

user_1749551 0
Viewed 725

I'm using Flask-SQLAlchemy to describe a Postgres database. Three related tables look like this (in part):

from sqlalchemy.dialects.postgresql import TSTZRANGE

class Shift(Base):
    __tablename__ = "shifts"

    id = db.Column(db.Integer, primary_key=True)
    hours = db.Column(TSTZRANGE, nullable=False)

class Volunteer(Base):
    __tablename__ = "volunteers"

    id = db.Column(db.Integer(), primary_key=True)

    shifts = db.relationship(
        "Shift",
        secondary="shift_assignments",
        backref=db.backref("volunteers", lazy="dynamic"),
    )

class ShiftAssignment(Base):
    __tablename__ = "shift_assignments"
    __table_args__ = (db.UniqueConstraint('shift_id', 'volunteer_id', name='_shift_vol_uc'),)

    id = db.Column(db.Integer, primary_key=True)
    shift_id = db.Column("shift_id", db.Integer(), db.ForeignKey("shifts.id"))
    volunteer_id = db.Column(
        "volunteer_id", db.Integer(), db.ForeignKey("volunteers.id")
    )

Now, I'm assigning a Volunteer to new Shift and want to make sure that the vol isn't already committed to a different Shift at the same time.

I have tried this in a Volunteer instance method, but it's not working:

new_shift = db.session.get(Shift, new_shift_id)

if new_shift not in self.shifts:
    for shift in self.shifts:
        overlap = db.session.scalar(shift.hours.overlaps(new_shift.hours))

This results in the following exception:

'DateTimeTZRange' object has no attribute 'overlaps'

It seems like I should probably not even be doing this by iterating over the list anyway, but should be directly querying the DB to do the date overlap math. So I guess I need to join the volunteers and shifts and then filter to find out if any shifts overlap with the target shift. But I can't figure out how to do that, and examples of overlaps and its RangeOperators friends is really thin on the ground.

Would appreciate a hand here.

flask-sqlalchemy sqlalchemy postgresql
1 Answers

It was much easier than I was making it. Again, this is in a Volunteer instance method.

new_shift = db.session.get(Shift, new_shift_id)

overlapping_shift = (
    db.session.query(Shift, ShiftAssignment)
    .join(ShiftAssignment)
    .filter(ShiftAssignment.volunteer_id == self.id)
    .filter(Shift.hours.overlaps(new_shift.hours))
    .first()
)

if overlapping_shift:
    print("overlap found")

Note that the query returns a (Shift, ShiftAssignment) tuple. We join the two appropriate tables and then filter twice, left with any overlapping shifts assigned to the current volunteer.

user_17495510
Related