CODEWITHSUNDEEP
pythonpostgresqlsqlalchemy
Verbal_Kint
Verbal_Kint

Reputation: 1416

SQLAlchemy hybrid_property/method to query against datetime object which is dynamically built based on column values

I am trying to manage api keys in an application.

Assuming this model:

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

Some of the APIs have daily limits which reset at a certain time. Currently I am storing this time as an Integer which represents the daily reset hour (perhaps storing it as an interval or time type would be more appropriate). I would like to be able to query against a calls_made, or even better a calls_remaining field. To do this I would have to calculate the last reset time and see if the last_used property happened prior to the daily_reset_time to determine if a reset has happened and therefore to reflect that in my model by setting _calls_made to 0. The instance property does this by using the .replace method on a datetime object.

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

This would clearly not work in a hybrid property expression since the value used in the datetime replace method would be a 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
    )

After looking at postgres's datetime functions I would probably need something more like this in place of the now.replace call as the return value for the case expression:

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

But that's also missing something...

How would I calculate this in an expression of a hybrid property? Is there a better approach to solve this seeing as I would probably also have to add a signal which checks if the _calls_made attribute should be reset?

Upvotes: 0

Views: 765

Answers (1)

Ilja Everil&#228;
Ilja Everil&#228;

Reputation: 52929

You could use the fact that intervals support multiplication:

from sqlalchemy import Interval, literal

...

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

You could also replace the CASE expression with arithmetic. The idea is to subtract the reset hours from the current time, moving it to previous day if reset has not yet been reached, truncate to day precision, and add the hours back:

@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

Upvotes: 2

Related Questions