RPUSH operation with optional TTL. Args: name: List key *values: Values to push ttl: Optional time-to-live in seconds Returns: Length of the list after push
(self, name: str, *values: FieldT, ttl: Optional[float] = None)
| 112 | self._zincrby = self.register_script(LUA_ZINCRBY) |
| 113 | |
| 114 | def rpush(self, name: str, *values: FieldT, ttl: Optional[float] = None) -> Union[Awaitable[int], int]: |
| 115 | """ |
| 116 | RPUSH operation with optional TTL. |
| 117 | |
| 118 | Args: |
| 119 | name: List key |
| 120 | *values: Values to push |
| 121 | ttl: Optional time-to-live in seconds |
| 122 | |
| 123 | Returns: |
| 124 | Length of the list after push |
| 125 | """ |
| 126 | if ttl is None: |
| 127 | return super().rpush(name, *values) |
| 128 | |
| 129 | with self.pipeline() as pipe: |
| 130 | pipe.multi() |
| 131 | pipe.rpush(name, *values) |
| 132 | pipe.expire(name, ttl) |
| 133 | result = pipe.execute() |
| 134 | return result[0] |
| 135 | |
| 136 | def zincrby( |
| 137 | self, |
no test coverage detected