|
1 | 1 | from pathlib import Path |
2 | 2 | import shutil |
3 | 3 | import os |
| 4 | +from contextlib import contextmanager |
| 5 | +import tempfile |
| 6 | +from typing import Iterator |
4 | 7 |
|
5 | 8 | from fileex import exception |
6 | 9 |
|
@@ -125,3 +128,79 @@ def merge( |
125 | 128 | moved_paths.append(dest_item) |
126 | 129 | source.rmdir() |
127 | 130 | return moved_paths |
| 131 | + |
| 132 | + |
| 133 | +@contextmanager |
| 134 | +def get_or_make( |
| 135 | + path: str | Path | None = None, |
| 136 | + ensure_empty: bool = False, |
| 137 | + suffix: str | None = None, |
| 138 | + prefix: str | None = None, |
| 139 | + dir: str | Path | None = None, |
| 140 | + ignore_cleanup_errors: bool = False, |
| 141 | + delete: bool = True |
| 142 | +) -> Iterator[Path]: |
| 143 | + """Provide a directory for I/O operations. |
| 144 | +
|
| 145 | + This context manager yields a `pathlib.Path` object |
| 146 | + pointing to an existing directory: |
| 147 | + - If `path` is provided, ensures the directory exists |
| 148 | + (creating it if necessary) and yields it. |
| 149 | + - If `path` is None, creates a temporary directory, |
| 150 | + yields it, and optionally removes it on exit. |
| 151 | +
|
| 152 | + Parameters |
| 153 | + ---------- |
| 154 | + path |
| 155 | + Filesystem path to use. If None, a temporary directory is created. |
| 156 | + ensure_empty |
| 157 | + If True, ensures that the directory is empty before yielding, |
| 158 | + raising an error if it is not. |
| 159 | + suffix |
| 160 | + Suffix for the temporary directory, if created. |
| 161 | + prefix |
| 162 | + Prefix for the temporary directory, if created. |
| 163 | + dir |
| 164 | + Directory in which to create the temporary directory, if created. |
| 165 | + ignore_cleanup_errors |
| 166 | + If True, ignore errors during cleanup of the temporary directory. |
| 167 | + delete |
| 168 | + If True, the temporary directory will be deleted on exit. |
| 169 | +
|
| 170 | + Yields |
| 171 | + ------ |
| 172 | + work_dir |
| 173 | + The directory to perform file operations in. |
| 174 | +
|
| 175 | + Examples |
| 176 | + -------- |
| 177 | + >>> with get_or_make('/tmp/data') as work_dir: |
| 178 | + ... (work_dir / 'file.txt').write_text('hello') |
| 179 | + >>> with get_or_make() as work_dir: |
| 180 | + ... # work_dir is a fresh temp dir, auto-cleaned |
| 181 | + ... (work_dir / 'temp.txt').write_text('world') |
| 182 | + """ |
| 183 | + if path: |
| 184 | + outdir = Path(path).resolve() |
| 185 | + if outdir.exists(): |
| 186 | + if not outdir.is_dir(): |
| 187 | + raise ValueError( |
| 188 | + f"The specified output path '{outdir}' is not a directory." |
| 189 | + ) |
| 190 | + if ensure_empty and any(outdir.iterdir()): |
| 191 | + raise ValueError( |
| 192 | + f"The specified output directory '{outdir}' is not empty." |
| 193 | + ) |
| 194 | + else: |
| 195 | + outdir.mkdir(parents=True, exist_ok=True) |
| 196 | + yield outdir |
| 197 | + else: |
| 198 | + with tempfile.TemporaryDirectory( |
| 199 | + suffix=suffix, |
| 200 | + prefix=prefix, |
| 201 | + dir=dir, |
| 202 | + ignore_cleanup_errors=ignore_cleanup_errors, |
| 203 | + delete=delete, |
| 204 | + ) as tmpdir: |
| 205 | + yield Path(tmpdir) |
| 206 | + return |
0 commit comments