ls_files.py
python
| 1 | """muse plumbing ls-files — list tracked files in a snapshot. |
| 2 | |
| 3 | Lists every file tracked in a commit's snapshot, along with the SHA-256 |
| 4 | object ID of its content. Defaults to the HEAD commit of the current branch. |
| 5 | |
| 6 | Output (JSON, default):: |
| 7 | |
| 8 | { |
| 9 | "commit_id": "<sha256>", |
| 10 | "snapshot_id": "<sha256>", |
| 11 | "files": [ |
| 12 | {"path": "tracks/drums.mid", "object_id": "<sha256>"}, |
| 13 | ... |
| 14 | ] |
| 15 | } |
| 16 | |
| 17 | Output (--format text):: |
| 18 | |
| 19 | <object_id> <path> |
| 20 | ... |
| 21 | |
| 22 | Plumbing contract |
| 23 | ----------------- |
| 24 | |
| 25 | - Exit 0: manifest listed successfully. |
| 26 | - Exit 1: commit or snapshot not found. |
| 27 | """ |
| 28 | |
| 29 | from __future__ import annotations |
| 30 | |
| 31 | import json |
| 32 | import logging |
| 33 | import pathlib |
| 34 | |
| 35 | import typer |
| 36 | |
| 37 | from muse.core.errors import ExitCode |
| 38 | from muse.core.repo import require_repo |
| 39 | from muse.core.store import get_commit_snapshot_manifest, get_head_commit_id, read_commit, read_current_branch |
| 40 | |
| 41 | logger = logging.getLogger(__name__) |
| 42 | |
| 43 | app = typer.Typer() |
| 44 | |
| 45 | |
| 46 | def _current_branch(root: pathlib.Path) -> str: |
| 47 | return read_current_branch(root) |
| 48 | |
| 49 | |
| 50 | @app.callback(invoke_without_command=True) |
| 51 | def ls_files( |
| 52 | ctx: typer.Context, |
| 53 | commit: str | None = typer.Option( |
| 54 | None, "--commit", "-c", help="Commit ID to read (default: HEAD)." |
| 55 | ), |
| 56 | fmt: str = typer.Option("json", "--format", help="Output format: json or text."), |
| 57 | ) -> None: |
| 58 | """List all tracked files and their object IDs in a snapshot. |
| 59 | |
| 60 | Analogous to ``git ls-files --stage``. Reads the snapshot manifest of |
| 61 | the given commit (or HEAD) and prints each tracked file path together |
| 62 | with its content-addressed object ID. |
| 63 | """ |
| 64 | root = require_repo() |
| 65 | |
| 66 | if commit is None: |
| 67 | branch = _current_branch(root) |
| 68 | commit_id = get_head_commit_id(root, branch) |
| 69 | if commit_id is None: |
| 70 | typer.echo(json.dumps({"error": "No commits on current branch."})) |
| 71 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 72 | else: |
| 73 | commit_id = commit |
| 74 | |
| 75 | commit_record = read_commit(root, commit_id) |
| 76 | if commit_record is None: |
| 77 | typer.echo(json.dumps({"error": f"Commit not found: {commit_id}"})) |
| 78 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 79 | |
| 80 | manifest = get_commit_snapshot_manifest(root, commit_id) |
| 81 | if manifest is None: |
| 82 | typer.echo(json.dumps({"error": f"Snapshot not found for commit: {commit_id}"})) |
| 83 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 84 | |
| 85 | files = [ |
| 86 | {"path": p, "object_id": oid} |
| 87 | for p, oid in sorted(manifest.items()) |
| 88 | ] |
| 89 | |
| 90 | if fmt == "text": |
| 91 | for entry in files: |
| 92 | typer.echo(f"{entry['object_id']}\t{entry['path']}") |
| 93 | return |
| 94 | |
| 95 | typer.echo(json.dumps({ |
| 96 | "commit_id": commit_id, |
| 97 | "snapshot_id": commit_record.snapshot_id, |
| 98 | "file_count": len(files), |
| 99 | "files": files, |
| 100 | }, indent=2)) |