Batch check existence of multiple images, one transaction per shard. Uses cursor.set_key() for fast key-only existence checks (avoids reading values), single MD5 per path, and parallel shard access. Args: image_paths: List of image paths to check. Retur
(self, image_paths: List[str])
| 305 | return txn.get(key) |
| 306 | |
| 307 | def batch_exists(self, image_paths: List[str]) -> Dict[str, bool]: |
| 308 | """Batch check existence of multiple images, one transaction per shard. |
| 309 | |
| 310 | Uses cursor.set_key() for fast key-only existence checks (avoids |
| 311 | reading values), single MD5 per path, and parallel shard access. |
| 312 | |
| 313 | Args: |
| 314 | image_paths: List of image paths to check. |
| 315 | |
| 316 | Returns: |
| 317 | Dict mapping image_path -> bool (exists or not). |
| 318 | """ |
| 319 | # Group paths by shard using single MD5 per path |
| 320 | shard_groups: Dict[int, List[Tuple[str, bytes]]] = defaultdict(list) |
| 321 | for path in image_paths: |
| 322 | shard_id, key = self._get_shard_id_and_key(path) |
| 323 | shard_groups[shard_id].append((path, key)) |
| 324 | |
| 325 | result = {} |
| 326 | result_lock = threading.Lock() |
| 327 | |
| 328 | def _check_shard(shard_id_and_items): |
| 329 | shard_id, items = shard_id_and_items |
| 330 | env = self._get_env(shard_id) |
| 331 | shard_result = {} |
| 332 | with env.begin(write=False, buffers=True) as txn: |
| 333 | cursor = txn.cursor() |
| 334 | for path, key in items: |
| 335 | shard_result[path] = cursor.set_key(key) |
| 336 | with result_lock: |
| 337 | result.update(shard_result) |
| 338 | |
| 339 | num_shards_to_check = len(shard_groups) |
| 340 | if num_shards_to_check > 1: |
| 341 | workers = min(num_shards_to_check, 8) |
| 342 | with ThreadPoolExecutor(max_workers=workers) as executor: |
| 343 | list(executor.map(_check_shard, shard_groups.items())) |
| 344 | else: |
| 345 | for item in shard_groups.items(): |
| 346 | _check_shard(item) |
| 347 | |
| 348 | return result |
| 349 | |
| 350 | def batch_get(self, image_paths: List[str]) -> Dict[str, Optional[bytes]]: |
| 351 | """Batch read multiple images, one transaction per shard. |
no test coverage detected