diff --git a/redis/commands/core.py b/redis/commands/core.py index 3595677de6..406a159538 100644 --- a/redis/commands/core.py +++ b/redis/commands/core.py @@ -1512,16 +1512,42 @@ def expire(self, name: KeyT, time: ExpiryT) -> ResponseT: time = int(time.total_seconds()) return self.execute_command("EXPIRE", name, time) - def expireat(self, name: KeyT, when: AbsExpiryT) -> ResponseT: + def expireat( + self, + name: KeyT, + when: AbsExpiryT, + nx: bool = False, + xx: bool = False, + gt: bool = False, + lt: bool = False, + ) -> ResponseT: """ - Set an expire flag on key ``name``. ``when`` can be represented - as an integer indicating unix time or a Python datetime object. + Set an expire flag on key ``name`` with given ``option``. ``when`` + can be represented as an integer indicating unix time or a Python + datetime object. + + Valid options are: + -> NX -- Set expiry only when the key has no expiry + -> XX -- Set expiry only when the key has an existing expiry + -> GT -- Set expiry only when the new expiry is greater than current one + -> LT -- Set expiry only when the new expiry is less than current one For more information check https://redis.io/commands/expireat """ if isinstance(when, datetime.datetime): when = int(time.mktime(when.timetuple())) - return self.execute_command("EXPIREAT", name, when) + + exp_option = list() + if nx: + exp_option.append("NX") + if xx: + exp_option.append("XX") + if gt: + exp_option.append("GT") + if lt: + exp_option.append("LT") + + return self.execute_command("EXPIREAT", name, when, *exp_option) def expiretime(self, key: str) -> int: """ diff --git a/tests/test_commands.py b/tests/test_commands.py index a73541ef10..a7b3a89c1c 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -1081,6 +1081,43 @@ def test_expiretime(self, r): r.expireat("a", 33177117420) assert r.expiretime("a") == 33177117420 + @skip_if_server_version_lt("7.0.0") + def test_expireat_option_nx(self, r): + assert r.set("key", "val") is True + expire_at = redis_server_time(r) + datetime.timedelta(minutes=1) + assert r.expireat("key", expire_at, nx=True) is True + expire_at = redis_server_time(r) + datetime.timedelta(minutes=2) + assert r.expireat("key", expire_at, nx=True) is False + + @skip_if_server_version_lt("7.0.0") + def test_expireat_option_xx(self, r): + assert r.set("key", "val") is True + expire_at = redis_server_time(r) + datetime.timedelta(minutes=1) + assert r.expireat("key", expire_at, xx=True) is False + assert r.expireat("key", expire_at) is True + expire_at = redis_server_time(r) + datetime.timedelta(minutes=2) + assert r.expireat("key", expire_at, xx=True) is True + + @skip_if_server_version_lt("7.0.0") + def test_expireat_option_gt(self, r): + expire_at = redis_server_time(r) + datetime.timedelta(minutes=2) + assert r.set("key", "val") is True + assert r.expireat("key", expire_at) is True + expire_at = redis_server_time(r) + datetime.timedelta(minutes=1) + assert r.expireat("key", expire_at, gt=True) is False + expire_at = redis_server_time(r) + datetime.timedelta(minutes=3) + assert r.expireat("key", expire_at, gt=True) is True + + @skip_if_server_version_lt("7.0.0") + def test_expireat_option_lt(self, r): + expire_at = redis_server_time(r) + datetime.timedelta(minutes=2) + assert r.set("key", "val") is True + assert r.expireat("key", expire_at) is True + expire_at = redis_server_time(r) + datetime.timedelta(minutes=3) + assert r.expireat("key", expire_at, lt=True) is False + expire_at = redis_server_time(r) + datetime.timedelta(minutes=1) + assert r.expireat("key", expire_at, lt=True) is True + def test_get_and_set(self, r): # get and set can't be tested independently of each other assert r.get("a") is None