Retrieve all artifacts for a program Args: program_id: ID of the program Returns: Dictionary of artifact name to content
(self, program_id: str)
| 2380 | logger.debug(f"Stored {len(large_artifacts)} large artifacts for program {program_id}") |
| 2381 | |
| 2382 | def get_artifacts(self, program_id: str) -> Dict[str, Union[str, bytes]]: |
| 2383 | """ |
| 2384 | Retrieve all artifacts for a program |
| 2385 | |
| 2386 | Args: |
| 2387 | program_id: ID of the program |
| 2388 | |
| 2389 | Returns: |
| 2390 | Dictionary of artifact name to content |
| 2391 | """ |
| 2392 | program = self.get(program_id) |
| 2393 | if not program: |
| 2394 | return {} |
| 2395 | |
| 2396 | artifacts = {} |
| 2397 | |
| 2398 | # Load small artifacts from JSON |
| 2399 | if program.artifacts_json: |
| 2400 | try: |
| 2401 | small_artifacts = json.loads(program.artifacts_json) |
| 2402 | artifacts.update(small_artifacts) |
| 2403 | except json.JSONDecodeError as e: |
| 2404 | logger.warning(f"Failed to decode artifacts JSON for program {program_id}: {e}") |
| 2405 | |
| 2406 | # Load large artifacts from disk |
| 2407 | if program.artifact_dir and os.path.exists(program.artifact_dir): |
| 2408 | disk_artifacts = self._load_artifact_dir(program.artifact_dir) |
| 2409 | artifacts.update(disk_artifacts) |
| 2410 | |
| 2411 | return artifacts |
| 2412 | |
| 2413 | def _get_artifact_size(self, value: Union[str, bytes]) -> int: |
| 2414 | """Get size of an artifact value in bytes""" |
no test coverage detected