SQLAlchemy hybrid_property / метод для запроса объекта datetime, который динамически создается на основе значений столбца

Question

Я пытаюсь управлять ключами API в приложении.

Предполагается, что эта модель:

class APIKey(db.Model):
    __tablename__ = 'api_keys'
    ...
    daily_reset_time = db.Column(db.Integer)  # daily reset hour maybe use Interval or Time type here?
    daily_limit = db.Column(db.Integer)
    per_second_limit = db.Column(db.Integer)
    _calls_made = db.Column(db.Integer, default=0)
    last_started = db.Column(db.DateTime)
    last_ended = db.Column(db.DateTime)
    ...

    @property
    def last_used(self):
        return self.last_ended if self.last_ended is not None else self.last_started

Некоторые API имеют дневные ограничения, которые сбрасываются в определенное время.В настоящее время я сохраняю это время как целое число, которое представляет ежедневный час сброса (возможно, было бы более уместно сохранить его как интервал или тип времени).Я хотел бы иметь возможность выполнять запросы к calls_made, а еще лучше к calls_remaining полю.Для этого мне нужно будет рассчитать время последнего сброса и посмотреть, произошло ли свойство last_used до daily_reset_time, чтобы определить, произошел ли сброс, и, следовательно, отразить это в моей модели, установив для _calls_made значение 0.Свойство экземпляра делает это с помощью метода .replace для объекта datetime.

@hybrid_property
def previous_reset(self):
    if self.daily_reset_time is not None:
        now = datetime.utcnow()
        if now.hour >= self.daily_reset_time:
            return now.replace(
                hour=self.daily_reset_time,
                minute=0,
                second=0,
                microsecond=0
            )
        else:
            return now.replace(
                hour=self.daily_reset_time,
                minute=0,
                second=0,
                microsecond=0
            ) - timedelta(days=1)

Это явно не будет работать в выражении гибридного свойства, поскольку значение, используемое в методе замены datetime, будетColumn instance:

@previous_reset.expression
def previous_reset(cls):
    now = datetime.utcnow()
    return case(
        [
            (  # its later than the daily reset time
                and_(cls.daily_reset_time.isnot(None),
                     cls.daily_reset_time <= now.hour),
                now.replace(
                    hour=cls.daily_reset_time,  # this is a valueerror
                    minute=0,
                    second=0,
                    microsecond=0
                )
            ),
            (
                and_(cls.daily_reset_time.isnot(None),
                     cls.daily_reset_time >= now.hour),
                (now - timedelta(days=1)).replace(
                    hour=cls.daily_reset_time,
                    minute=0,
                    second=0,
                    microsecond=0
                )
            ),
        ],
        else_=None
    )

После просмотра функций dateg в postgres мне, вероятно, понадобится нечто более подобное вместо вызова now.replace в качестве возвращаемого значения для выражения case:

func.date_trunc('days', func.timezone('utc', func.current_timestamp)) + f"interval '{cls.daily_reset_time}' hours'"

Но это также чего-то не хватает ...

Как бы я вычислил это в выражении гибридного свойства?Есть ли лучший подход для решения этой проблемы, поскольку мне, вероятно, также необходимо добавить сигнал, который проверяет, следует ли сбрасывать атрибут _calls_made?

Answer 1

Вы можете использовать тот факт, что интервалы поддерживают умножение:

from sqlalchemy import Interval, literal

...

(func.date_trunc('days', func.timezone('utc', func.current_timestamp())) +
 literal('1 hour').cast(Interval) * cls.daily_reset_time)

Вы также можете заменить выражение CASE на арифметику.Идея состоит в том, чтобы вычесть часы сброса из текущего времени, переместив их на предыдущий день, если сброс еще не достигнут, обрезать до точности дня и добавить часы назад:

@previous_reset.expression
def previous_reset(cls):
    utc_now = func.timezone('utc', func.current_timestamp())
    hour_delta = literal('1 hour').cast(Interval) * cls.daily_reset_time
    # The CASE expression is not needed even for the NULL handling, as the
    # original would produce NULL in case `daily_reset_time` was NULL, as
    # does the below expression.
    return func.date_trunc('day', utc_now - hour_delta) + hour_delta