Write an artifact to a file
(self, artifact_dir: str, key: str, value: Union[str, bytes])
| 2487 | logger.info(f"Cleaned up {deleted_count} old artifact directories.") |
| 2488 | |
| 2489 | def _write_artifact_file(self, artifact_dir: str, key: str, value: Union[str, bytes]) -> None: |
| 2490 | """Write an artifact to a file""" |
| 2491 | # Sanitize filename |
| 2492 | safe_key = "".join(c for c in key if c.isalnum() or c in "._-") |
| 2493 | if not safe_key: |
| 2494 | safe_key = "artifact" |
| 2495 | |
| 2496 | file_path = os.path.join(artifact_dir, safe_key) |
| 2497 | |
| 2498 | try: |
| 2499 | if isinstance(value, str): |
| 2500 | with open(file_path, "w", encoding="utf-8") as f: |
| 2501 | f.write(value) |
| 2502 | elif isinstance(value, bytes): |
| 2503 | with open(file_path, "wb") as f: |
| 2504 | f.write(value) |
| 2505 | else: |
| 2506 | # Convert to string and write |
| 2507 | with open(file_path, "w", encoding="utf-8") as f: |
| 2508 | f.write(str(value)) |
| 2509 | except Exception as e: |
| 2510 | logger.warning(f"Failed to write artifact {key} to {file_path}: {e}") |
| 2511 | |
| 2512 | def _load_artifact_dir(self, artifact_dir: str) -> Dict[str, Union[str, bytes]]: |
| 2513 | """Load artifacts from a directory""" |