Write a single image to cache. Args: image_path: Image path (used as key). img_bytes: Image data as bytes. Returns: bool: True if successful.
(self, image_path: str, img_bytes: bytes)
| 266 | return txn.get(key) is not None |
| 267 | |
| 268 | def put(self, image_path: str, img_bytes: bytes) -> bool: |
| 269 | """Write a single image to cache. |
| 270 | |
| 271 | Args: |
| 272 | image_path: Image path (used as key). |
| 273 | img_bytes: Image data as bytes. |
| 274 | |
| 275 | Returns: |
| 276 | bool: True if successful. |
| 277 | """ |
| 278 | shard_id = self._get_shard_id(image_path) |
| 279 | env = self._get_env(shard_id) |
| 280 | key = self._get_key(image_path) |
| 281 | |
| 282 | try: |
| 283 | with env.begin(write=True) as txn: |
| 284 | txn.put(key, img_bytes) |
| 285 | return True |
| 286 | except Exception as e: |
| 287 | if self.logger: |
| 288 | self.logger.error(f"Failed to write to LMDB [{image_path}]: {e}") |
| 289 | return False |
| 290 | |
| 291 | def get(self, image_path: str) -> Optional[bytes]: |
| 292 | """Read image data from cache. |
no test coverage detected