Sqlalchemy gets string in timeslot

Question

I have a model called "Destination" that has datetime columns, which is the datetime field and duration , which is the Integer field and represents the duration in minutes. Now I want to check if func.now() between the destination datetime and the sum of datetime and duration

I'm currently trying to do this, 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 can do it directly in ORM for both sqlite and postgres, but sqlalchemy allows you to extend it cross-dialect using Custom SQL Constructors and a compilation extension .

This snippet may not be entirely right, because I cracked it with several different models and translated it for this, but I have something very close to correctly processing postgres SQL:

 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 similar in mind, you will need to turn your initial query into a cross-dialect query that maps directly to SQL, rather than performing a duration calculation 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, consider whether it is generally “cheaper” to use postgresql instead of sqlite everywhere. I assume that you have development / production differences that should be avoided. Installing postgresql on any modern OS is pretty trivial.
Assuming this is not an option / desired, continue.

Failure 2: A solution with a custom SQL construct (as per @Josh's answer) is indeed the only reasonable way to achieve this. Unfortunately, the proposed solution does not actually work for sqlite and cannot be fixed with just a few lines, therefore, 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 really difficult to use date operations, especially adding / subtracting intervals from dates. Therefore, let's approach and create custom functions in a slightly different way to get the 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 verification using the following query (I do not add limit(1).one_or_none to 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 the current time ( func.now() ), and you can check (and unit test) your data at 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 the problem is solved here , and the solution should work for both of your database engines.

BONUS:

You can hide the implementation of checking whether the event is current using Hybrid methods .

Let's add the 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 allows you to run a check in memory (on python, not on the SQL side):

 print(my_appointment.is_current()) 

The second method allows you to build a query, as shown below:

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

If if at_time not specified, the current time will be used. You can, of course, change the request as you wish:

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

Answer 3

If I understand the question correctly ... Something like this?

 def check_for_current_appt(appt_epoch, appt_duration): '''INPUT : appt_timestamp (int (epoch time)): start time for appointment appt_duration (int): duration of appointment in seconds OUTPUT : appt_underway (bool): True if appointment is currently underway''' now = time.time() appt_underway = 0 < (now - appt_epoch) < appt_duration return appt_underway 

I will leave it to transform into an era and seconds for a duration

Answer 4

From what I understand, PostgreSQL uses unix timestamps, while Sqlite uses iso-8601 timestamps stored as strings. Therefore, if you change the general structure of your database to use the Sqlite format, it should provide you with the necessary functionality, you can convert the date and time using the .isoformat () function. Unfortunately, if you are not working with only test data, you will have to iterate over all the rows to change them. Not sure if this is acceptable to you, but this is an easy way to do this.

Based on the datetime section http://docs.sqlalchemy.org/en/latest/core/type_basics.html