Sqlalchemyがタイムスロットで行を取得する

Question

私はDateTimeフィールドとdurationフィールドのdatetimeフィールドを持ち、継続時間を分単位で表すAppointmentというモデルを持っています。今私はfunc.now()が任命のdatetimeとdatetimeの合計と、私はこのようにそれを実行しようとする現在、午前duration

の間にあるかどうかを確認したいが、私は、PostgreSQLやSQLiteの両方のために動作しますソリューションが必要です。

Answer 1

のためのエポック時間と秒に変換するためにあなたにそれを残しておきますsqlalchemyを使用すると、Custom SQL Constructs and Compilation Extensionを使用してクロスダイヤル方法で拡張することができます。

私はいくつかの異なるモデルとそれをハッキングし、このためにそれを超える翻訳が、私は正しく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' 

をPostgreSQLはUNIXタイムスタンプを使用して、私はそれを理解して何から

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

Answer 2

質問が正しく理解されている場合... これは何か？

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 

私は私はあなたがsqliteのとpostgresの両方のためのORMで直接これを行うことができるだろうとは思わない期間

Answer 3

：それはあなたの代わりにPythonでデュレーションの計算を行うの直接SQLにレンダリングするクロス方言一つにあなたの元のクエリを有効にすることができるはず設定しましたSqliteはiso-文字列として保存される8601タイムスタンプ。したがって、データベースの全体的な構造をSqlite形式を使用するように変更する場合は、必要な機能を提供する必要があります。datetimeを.isoformat（）関数で変換できます。残念ながら、テストデータだけで作業していない場合は、すべての行を繰り返して変更する必要があります。これがあなたに受け入れられるかどうかは分かりませんが、簡単な方法です。

http://docs.sqlalchemy.org/en/latest/core/type_basics.html

Answer 4

免責事項1の日時セクションに基づいて：実際にはどこにでもpostgresql代わりのsqliteのを使用して「安い」でない場合すべてのまず、と思います。私は開発/生産上の違いがあると思いますが、これは避けるべきです。最新のOSにpostgresqlをインストールすることは非常に簡単です。
上記はオプションではないと思っていますが、これからも続けましょう。

免責事項2：カスタムSQL構成（@ Joshの回答に従って）を使用したソリューションは、これを実現する実際の唯一の方法です。 残念なことに、提案された解決策は実際にはsqliteでは機能せず、ほんの数行で解決できなかったため、別個の回答となりました。

ソリューション： 次のモデルを持っていると仮定すると：

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は特に追加/日付から間隔を減じ、日付操作を扱う本当に難しいです。したがって、のは多少異なるそれに近づくと数分で2つの日付の間の間隔を取得するカスタム関数を作成してみましょう：、

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) 

あなたは既に次のクエリを使用して、チェックを実装することができます（私は私の例でlimit(1).one_or_noneを追加していないですがこれを使用しますあなたのデータは、任意の時間に対して）

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

しかし、今は、現在の時刻（func.now()）によって限定されるものではなく、あなたはチェック（およびユニットテストできます：

あなたはそれを必要とするとき、明らかに）を行うことができます

# 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) 
) 

ここでは、問題はで解決され、使用する両方のデータベースエンジンで解決するはずです。

BONUS：

あなたはにイベントがHybrid Methodsを使用して、現在あるかどうかをチェックするのインプリメンテーションを非表示にすることができます。

Appointmentクラスに以下を追加できます：

@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') 

最初のものは、あなたが（のpythonではなく、SQL側の）メモリのチェックを実行することができます：

print(my_appointment.is_current()) 

秒1を

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

ここで、の場合、次のようなクエリを作成できます。 0を指定しないと、現在の時刻が使用されます。あなたの例では

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