Sqlalchemy获取行的时间

Question

我有一个名为ocation的模型，它具有列^{cd1>}这是^{{cd2>}字段和^{cd3>}字段，它是^{{cd4>}字段，以分钟为单位表示持续时间。现在我想检查^{cd5>}是否介于约会的^{{cd1>}和^{{cd3>}的总和之间

我目前正试图这样做，但我需要一个解决方案，它将同时适用于PostgreSQL和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

如果我正确地理解了这个问题。。。 像这样？在

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

我把时间转换成历元时间和秒数留给你

Answer 2

我不认为您可以在ORM中为sqlite和postgres直接做到这一点，但是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'

通过这样的设置，您应该能够将原始查询转换为跨方言查询，从而直接呈现为SQL，而不是使用Python进行持续时间计算：

^{pr2}$

Answer 3

免责声明1:首先，考虑一下实际使用postgresql代替sqlite是否“更便宜”。我假设你在开发/生产上有差异，你应该避免。在任何现代操作系统上安装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处理日期操作非常棘手，尤其是从日期中加/减间隔。因此，让我们以稍微不同的方式处理它，并创建自定义函数，以分钟为单位获取两个日期之间的间隔：

^{pr2}$

您已经可以使用以下查询实现检查（我没有在我的示例中添加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)
)

基本上，问题在这里得到了解决，解决方案应该适用于您使用的两个数据库引擎。在

奖金：

您可以使用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())

第二种方法允许您构造如下查询：

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

如果at_time未指定，则使用当前时间。当然，您可以根据需要修改查询：

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