Sqlalchemy nhận hàng trong khoảng thời gian

Question

Tôi có một mô hình được gọi là Cuộc hẹn có các cột datetime là trường DateTime và duration là trường Integer và thể hiện thời lượng tính bằng phút. Bây giờ tôi muốn kiểm tra xem func.now() là giữa datetime của việc bổ nhiệm và tổng các datetime và duration

Tôi hiện đang cố gắng để làm điều đó theo cách này, nhưng tôi cần một giải pháp mà sẽ làm việc cho cả hai PostgreSQL và 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

Tôi không nghĩ rằng bạn sẽ có thể làm điều này trực tiếp trong ORM cho cả sqlite và postgres , nhưng sqlalchemy cho phép bạn mở rộng nó theo một phương thức đa phương ngữ với Custom SQL Constructs and Compilation Extension.

đoạn này có thể không được chính xác đúng bởi vì tôi bị hack vào nó với một số mô hình khác nhau và dịch nó qua cho điều này, nhưng tôi đã nhận một cái gì đó rất gần với render postgres SQL một cách chính xác:

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' 

Với một cái gì đó như mà thiết lập bạn sẽ có thể để biến truy vấn ban đầu của bạn vào một cross-phương ngữ mà ám vào SQL trực tiếp thay vì làm việc tính toán thời gian bằng Python:

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

Answer 2

Nếu tôi hiểu đúng câu hỏi ... Điều gì đó tương tự?

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 

Tôi sẽ để lại nó cho bạn để chuyển đổi sang kỷ nguyên thời gian và giây trong suốt thời gian

Answer 3

Từ những gì tôi hiểu về nó, PostgreSQL sử dụng timestamps unix trong khi Sqlite sử dụng iso- 8601 dấu thời gian được lưu trữ dưới dạng chuỗi. Vì vậy, nếu bạn thay đổi cấu trúc tổng thể của cơ sở dữ liệu của bạn để sử dụng định dạng Sqlite, nó sẽ cung cấp cho bạn chức năng bạn muốn, bạn có thể chuyển đổi datetime với hàm .isoformat(). Thật không may nếu bạn không làm việc với dữ liệu thử nghiệm duy nhất, bạn sẽ phải lặp qua tất cả các hàng để thay đổi chúng. Không chắc chắn nếu điều này là chấp nhận được với bạn nhưng là một cách dễ dàng để làm điều đó.

Dựa trên phần datetime của http://docs.sqlalchemy.org/en/latest/core/type_basics.html

Answer 4

Disclaimer 1:Trước hết, nghĩ rằng nếu nó không phải là "rẻ hơn" thực sự sử dụng postgresql thay vì sqlite ở khắp mọi nơi. Tôi cho rằng bạn có sự khác biệt về phát triển/sản xuất, mà bạn nên tránh. Cài đặt postgresql trên bất kỳ hệ điều hành hiện đại nào là khá tầm thường.
Giả sử ở trên không phải là một tùy chọn/mong muốn, hãy tiếp tục.

Tuyên bố từ chối trách nhiệm 2: Giải pháp với cấu trúc SQL tùy chỉnh (theo câu trả lời của @ Josh) thực sự là cách hợp lý duy nhất để đạt được điều này. Thật không may, giải pháp được đề xuất không thực sự hoạt động cho sqlite và không thể khắc phục được chỉ với một vài dòng, do đó có một câu trả lời riêng biệt.

Giải pháp: Giả sử bạn có mô hình sau:

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 là thực sự khó khăn đối phó với hoạt động ngày, đặc biệt là thêm/trừ khoảng thời gian từ ngày. Vì vậy, chúng ta hãy tiếp cận nó một chút khác nhau và tạo ra các chức năng tùy chỉnh để có được một khoảng thời gian giữa hai thời điểm trong vài phút:

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) 

Bạn đã có thể thực hiện kiểm tra bằng cách sử dụng truy vấn sau đây (tôi không thêm limit(1).one_or_none trong các ví dụ của tôi, mà bạn rõ ràng có thể làm khi bạn cần nó):

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

Nhưng bây giờ bạn không bị giới hạn bởi thời gian hiện tại (func.now()), và bạn có thể kiểm tra (và kiểm tra đơn vị) dữ liệu của bạn chống lại bất cứ lúc nào:

012.

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

Về cơ bản, sự cố được giải quyết tại đây và giải pháp sẽ hoạt động cho cả hai công cụ cơ sở dữ liệu mà bạn sử dụng.

BONUS:

Bạn có thể ẩn việc thực hiện kiểm tra nếu sự kiện là hiện tại sử dụng Hybrid Methods.

Cho phép thêm sau vào lớp 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') 

Người đầu tiên cho phép bạn chạy kiểm tra trong bộ nhớ (trên python, không nằm về phía SQL):

print(my_appointment.is_current()) 

thứ hai cho phép bạn tạo truy vấn như dưới đây:

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

nơi nếu at_time nó không được chỉ định, thời gian hiện tại sẽ được sử dụng. Tất nhiên, bạn có thể sửa đổi truy vấn theo ý muốn:

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