(
self, key: str, value: bytes, expires: int | datetime | None = None
)
| 21 | return self.conn.get(key) |
| 22 | |
| 23 | def set( |
| 24 | self, key: str, value: bytes, expires: int | datetime | None = None |
| 25 | ) -> None: |
| 26 | if not expires: |
| 27 | self.conn.set(key, value) |
| 28 | elif isinstance(expires, datetime): |
| 29 | now_utc = datetime.now(timezone.utc) |
| 30 | if expires.tzinfo is None: |
| 31 | now_utc = now_utc.replace(tzinfo=None) |
| 32 | delta = expires - now_utc |
| 33 | self.conn.setex(key, int(delta.total_seconds()), value) |
| 34 | else: |
| 35 | self.conn.setex(key, expires, value) |
| 36 | |
| 37 | def delete(self, key: str) -> None: |
| 38 | self.conn.delete(key) |
nothing calls this directly
no outgoing calls
no test coverage detected