| 2 | |
| 3 | |
| 4 | def du( |
| 5 | url: str, |
| 6 | path: Optional[str] = None, |
| 7 | rev: Optional[str] = None, |
| 8 | summarize: bool = False, |
| 9 | config: Union[dict[str, Any], str, None] = None, |
| 10 | remote: Optional[str] = None, |
| 11 | remote_config: Optional[dict] = None, |
| 12 | ): |
| 13 | from dvc.config import Config |
| 14 | |
| 15 | from . import Repo |
| 16 | |
| 17 | if config and not isinstance(config, dict): |
| 18 | config_dict = Config.load_file(config) |
| 19 | else: |
| 20 | config_dict = None |
| 21 | |
| 22 | with Repo.open( |
| 23 | url, |
| 24 | rev=rev, |
| 25 | subrepos=True, |
| 26 | uninitialized=True, |
| 27 | config=config_dict, |
| 28 | remote=remote, |
| 29 | remote_config=remote_config, |
| 30 | ) as repo: |
| 31 | path = path or "" |
| 32 | |
| 33 | fs = repo.dvcfs |
| 34 | |
| 35 | if summarize or not fs.isdir(path): |
| 36 | return [(path, fs.du(path, total=True))] |
| 37 | |
| 38 | ret = [ |
| 39 | (entry_path, fs.du(entry_path, total=True)) for entry_path in fs.ls(path) |
| 40 | ] |
| 41 | ret.append((path, sum(entry[1] for entry in ret))) |
| 42 | return ret |