Sqlalchemy get row in timeslot

Question

I have a model called Appointment which has the columns datetime which is a DateTime field and duration which is an Integer field and represents duration in minutes. Now I want to check if func.now() is between the datetime of the appointment and the sum of the datetime and duration

I am currently to try to do it this way, but I need a solution that will work for both PostgreSQL and SQLite.

current_appointment = Appointment.query.filter(
    Appointment.datetime.between(
        func.now(),
        func.timestampadd(
            'MINUTES', Appointment.duration, func.now()
            )
        )
    ).limit(1).one_or_none()

Answer 1

I don't think you'll be able to do this directly in the ORM for both sqlite and postgres, but sqlalchemy lets you extend it in a cross-dialect way with Custom SQL Constructs and Compilation Extension.

This snippet might not be exactly right because I hacked at it with some different models and translated it over for this, but I got something very close to render the postgres SQL correctly:

from sqlalchemy import func
from sqlalchemy.sql import expression
from sqlalchemy.types import DateTime
from sqlalchemy.ext.compiler import compiles

class durationnow(expression.FunctionElement):
    type = DateTime()
    name = 'durationnow'

@compiles(durationnow, 'sqlite')
def sl_durationnow(element, compiler, **kw):
    return compiler.process(
        func.timestampadd('MINUTES', element.clauses, func.now())
    )

@compiles(durationnow, 'postgresql')
def pg_durationnow(element, compiler, **kw):
    return compiler.process(
        func.now() + func.make_interval(0, 0, 0, 0, 0, element.clauses)
    )

    # Or alternatively...
    # return "now() - make_interval(0, 0, 0, 0, 0, {})".format(compiler.process(element.clauses))
    # which is more in-line with how the documentation uses 'compiles'

With something like that set up you should be able to turn your original query into a cross-dialect one that renders to SQL directly instead of doing the duration computation in Python:

current_appointment = Appointment.query.filter(
    Appointment.datetime.between(
        func.now(),
        durationnow(Appointment.duration)
    ).limit(1).one_or_none()

Answer 2

Disclaimer 1: First of all, think if it is not "cheaper" to actually use postgresql instead of sqlite everywhere. I assume you have development/production differences, which you should avoid. Installation of postgresql on any modern OS is quite trivial.
Assuming above is not an option/desired, let's continue.

Disclaimer 2: The solution with the custom SQL construct (as per @Josh's answer) is really the only reasonable way to achieve this. Unfortunately, the proposed solution does not actually work for sqlite, and could not be fixed with just few lines, hence a separate answer.

Solution: Assuming you have the following model:

class Appointment(Base):
    __tablename__ = 'appointment'

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    datetime = Column(DateTime)  # @note: should be better named `start_date`?
    duration = Column(Integer)

sqlite is really tricky dealing with dates operations, especially adding/subtracting intervals from dates. Therefore, let's approach it somewhat differently and create custom functions to get an interval between two dates in minutes:

class diff_minutes(expression.FunctionElement):
    type = Integer()
    name = 'diff_minutes'

@compiles(diff_minutes, 'sqlite')
def sqlite_diff_minutes(element, compiler, **kw):
    dt1, dt2 = list(element.clauses)
    return compiler.process(
        (func.strftime('%s', dt1) - func.strftime('%s', dt2)) / 60
    )

@compiles(diff_minutes, 'postgresql')
def postgres_diff_minutes(element, compiler, **kw):
    dt1, dt2 = list(element.clauses)
    return compiler.process(func.extract('epoch', dt1 - dt2) / 60)

You can already implement your check using following query (i am not adding limit(1).one_or_none in my examples, which you can obviously do when you need it):

q = (
    session
    .query(Appointment)
    .filter(Appointment.datetime <= func.now())
    .filter(diff_minutes(func.now(), Appointment.datetime) <= Appointment.duration)
)

But now you are not limited by current time (func.now()), and you can check (and unit test) your data against any time:

# at_time = func.now()
at_time = datetime.datetime(2017, 11, 11, 17, 50, 0)
q = (
    session
    .query(Appointment)
    .filter(Appointment.datetime <= at_time)
    .filter(diff_minutes(at_time, Appointment.datetime) <= Appointment.duration)
)

Basically, problem is solved here, and the solution should work for both database engines you use.

BONUS:

You can hide the implementation of checking if the event is current using Hybrid Methods.

Lets add following to the Appointment class:

@hybrid_method
def is_current(self, at_time=None):
    if at_time is None:
        at_time = datetime.datetime.now()
    return self.datetime <= at_time <= self.datetime + datetime.timedelta(minutes=self.duration)

@is_current.expression
def is_current(cls, at_time=None):
    if at_time is None:
        at_time = datetime.datetime.now()

    stime = cls.datetime
    diffm = diff_minutes(at_time, cls.datetime)
    return and_(diffm >= 0, cls.duration >= diffm).label('is_current')

The first one allows you to run the check in memory (on python, not on SQL side):

print(my_appointment.is_current())

The second one allows you to construct query like below:

q = session.query(Appointment).filter(Appointment.is_current(at_time))

where if at_time it not specified, current time will be used. You can, of course then modify the query as you wish:

current_appointment = session.query(Appointment).filter(Appointment.is_current()).limit(1).one_or_none()