cli: implement remote submission uploads Issue 1486

.github/workflows/testing.yml

@@ -106,6 +106,11 @@ jobs:
         if: matrix.package != 'emmet-core'
         run: python -m pip install --user --no-deps ./emmet-core
 
+      - name: Install editable emmet-archival for emmet-cli
+        shell: bash -l {0}
+        if: matrix.package == 'emmet-cli'
+        run: python -m pip install --user --no-deps ./emmet-archival
+
       - name: Install ${{ matrix.package }}
         shell: bash -l {0}
         run: python -m pip install --user --no-deps -e ./${{ matrix.package }}

emmet-cli/emmet/cli/state_manager.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
+import fcntl
 import json
 import logging
+import os
 from pathlib import Path
 from typing import Any, Callable, Self, TextIO
-import fcntl
+from uuid import uuid4
 
 logger = logging.getLogger("emmet")
 
@@ -32,13 +34,14 @@ class StateManager:
     """Manages persistent state for the CLI application."""
 
     def __init__(self, state_dir: Path | str = Path.home() / ".emmet"):
-        # Store only the state file path
-        self.state_file = str(Path(state_dir) / "state.json")
+        self.state_dir = Path(state_dir)
+        self.state_file = str(self.state_dir / "state.json")
         self._ensure_state_dir()
 
     def _ensure_state_dir(self) -> None:
         """Ensures the state directory exists."""
-        Path(self.state_file).parent.mkdir(parents=True, exist_ok=True)
+        self.state_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+        self.state_dir.chmod(0o700)
 
     def _load_state(self) -> dict[str, Any]:
         """Loads state from disk. Not thread safe."""
@@ -54,8 +57,37 @@ def _load_state(self) -> dict[str, Any]:
 
     def _save_state(self, state: dict[str, Any]) -> None:
         """Saves current state to disk. Not thread safe."""
-        with Path(self.state_file).open("w") as f:
-            json.dump(state, f, indent=2)
+        state_path = Path(self.state_file)
+        temporary_path = state_path.with_name(f".{state_path.name}.{uuid4().hex}.tmp")
+        descriptor = os.open(
+            temporary_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
+        )
+        try:
+            try:
+                state_file = os.fdopen(descriptor, "w")
+            except Exception:
+                os.close(descriptor)
+                raise
+            with state_file as f:
+                json.dump(state, f, indent=2)
+                f.flush()
+                os.fsync(f.fileno())
+            os.replace(temporary_path, state_path)
+            try:
+                directory_descriptor = os.open(
+                    state_path.parent, os.O_RDONLY | os.O_DIRECTORY
+                )
+                try:
+                    os.fsync(directory_descriptor)
+                finally:
+                    os.close(directory_descriptor)
+            except OSError:
+                logger.warning(
+                    "State was saved, but its directory entry could not be synced."
+                )
+        except Exception:
+            temporary_path.unlink(missing_ok=True)
+            raise
 
     def get(self, key: str, default: Any = None) -> Any:
         """Gets a value from state."""

emmet-cli/emmet/cli/submission.py

@@ -7,7 +7,7 @@
 from multiprocessing import get_context
 from os import PathLike, cpu_count
 from pathlib import Path
-from typing import ClassVar, Iterable
+from typing import ClassVar, Iterable, Literal, Protocol
 from uuid import UUID, uuid4
 
 from pydantic import BaseModel, Field, PrivateAttr
@@ -72,6 +72,35 @@ def refresh(self) -> None:
             self.calc_validation_errors.clear()
 
 
+class CalculationChange(BaseModel):
+    """A calculation-level change included in a submission snapshot."""
+
+    calculation_id: UUID
+    status: Literal["added", "changed", "removed"]
+    locator: CalculationLocator
+    calculation: CalculationMetadata | None = None
+    added_files: list[str] = Field(default_factory=list)
+    changed_files: list[str] = Field(default_factory=list)
+    removed_files: list[str] = Field(default_factory=list)
+
+
+class SubmissionChangeSet(BaseModel):
+    """The complete set of changes between two submission snapshots."""
+
+    current_calculations: list[tuple[CalculationLocator, CalculationMetadata]]
+    changes: list[CalculationChange]
+
+    @property
+    def has_changes(self) -> bool:
+        return bool(self.changes)
+
+
+class SubmissionUploader(Protocol):
+    """Upload a staged submission snapshot to a remote service."""
+
+    def upload(self, submission_id: UUID, changes: SubmissionChangeSet) -> None: ...
+
+
 def invoke_calc_refresh(args):
     path, cm = args
     cm.refresh()
@@ -110,9 +139,7 @@ class Submission(BaseModel):
         default=None,
     )
 
-    _pending_push: dict[CalculationLocator, FileMetadata] | None = PrivateAttr(
-        default=None
-    )
+    _pending_push: SubmissionChangeSet | None = PrivateAttr(default=None)
 
     def last_pushed(
         self,
@@ -315,8 +342,8 @@ def _create_calculations_copy(self, refresh: bool = False):
                     cm.refresh()
         return pending_calculations
 
-    def stage_for_push(self) -> list[FileMetadata]:
-        """Stages submission for push. Returns the list of files that will need to be (re)pushed."""
+    def stage_for_push(self) -> SubmissionChangeSet:
+        """Stage and validate the current snapshot for a remote push."""
         self.pending_calculations = self._create_calculations_copy()
 
         if not self.validate_submission():
@@ -327,46 +354,94 @@ def stage_for_push(self) -> list[FileMetadata]:
                 "Submission does not pass validation. Please fix validation errors prior to staging."
             )
 
-        changes = self.get_changed_files_per_calc_path(
+        self._pending_push = self.get_submission_changes(
             self.last_pushed(), self.pending_calculations
         )
-        self._pending_push = changes  # type: ignore[assignment]
+        return self._pending_push
 
-        return [item for sublist in changes.values() for item in sublist]
-
-    def get_changed_files_per_calc_path(
+    def get_submission_changes(
         self,
         previous: list[tuple[CalculationLocator, CalculationMetadata]] | None,
         current: list[tuple[CalculationLocator, CalculationMetadata]],
-    ) -> dict[CalculationLocator, list[FileMetadata]]:
-        changes: dict[CalculationLocator, list[FileMetadata]] = {}
-        if not previous:
-            changes = {k: v.files for k, v in current}
-        else:
-            for loc, cm in current:
-                prev_cm = next((cm_p for loc_p, cm_p in previous if loc_p == loc), None)
-                if prev_cm is None:
-                    changes[loc] = cm.files
-                else:
-                    file_changes = []
-                    for fm in cm.files:
-                        match = next(
-                            (item for item in prev_cm.files if item == fm), None
-                        )
-                        if match is None or fm.hash != match.hash:
-                            file_changes.append(fm)
-                    if file_changes:
-                        changes[loc] = file_changes
-        return changes
-
-    def push(self) -> None:
+    ) -> SubmissionChangeSet:
+        """Return added, changed, and removed calculations and files."""
+        previous_by_id = {
+            calculation.id: (locator, calculation)
+            for locator, calculation in (previous or [])
+        }
+        current_by_id = {
+            calculation.id: (locator, calculation) for locator, calculation in current
+        }
+        changes = []
+
+        for calculation_id in sorted(current_by_id, key=str):
+            locator, calculation = current_by_id[calculation_id]
+            previous_entry = previous_by_id.get(calculation_id)
+            current_files = {file.name: file for file in calculation.files}
+
+            if previous_entry is None:
+                changes.append(
+                    CalculationChange(
+                        calculation_id=calculation_id,
+                        status="added",
+                        locator=locator,
+                        calculation=calculation,
+                        added_files=sorted(current_files),
+                    )
+                )
+                continue
+
+            _, previous_calculation = previous_entry
+            previous_files = {file.name: file for file in previous_calculation.files}
+            added_files = sorted(current_files.keys() - previous_files.keys())
+            removed_files = sorted(previous_files.keys() - current_files.keys())
+            changed_files = sorted(
+                name
+                for name in current_files.keys() & previous_files.keys()
+                if current_files[name].hash != previous_files[name].hash
+            )
+            if added_files or changed_files or removed_files:
+                changes.append(
+                    CalculationChange(
+                        calculation_id=calculation_id,
+                        status="changed",
+                        locator=locator,
+                        calculation=calculation,
+                        added_files=added_files,
+                        changed_files=changed_files,
+                        removed_files=removed_files,
+                    )
+                )
+
+        for calculation_id in sorted(
+            previous_by_id.keys() - current_by_id.keys(), key=str
+        ):
+            locator, calculation = previous_by_id[calculation_id]
+            changes.append(
+                CalculationChange(
+                    calculation_id=calculation_id,
+                    status="removed",
+                    locator=locator,
+                    removed_files=sorted(file.name for file in calculation.files),
+                )
+            )
+
+        return SubmissionChangeSet(
+            current_calculations=current,
+            changes=changes,
+        )
+
+    def push(self, uploader: SubmissionUploader) -> None:
         """Performs the push. Returns info about the push"""
-        if not self.pending_calculations or not self._pending_push:
+        if (
+            self.pending_calculations is None
+            or self._pending_push is None
+            or not self._pending_push.has_changes
+        ):
             raise EmmetCliError("Nothing is staged. Please stage before pushing.")
 
-        if self.get_changed_files_per_calc_path(
-            self.pending_calculations, self._create_calculations_copy(refresh=True)
-        ):
+        current = self._create_calculations_copy(refresh=True)
+        if self.get_submission_changes(self.pending_calculations, current).has_changes:
             raise EmmetCliError(
                 "Files for submission have changed since staging. Please re-stage before pushing."
             )
@@ -378,12 +453,7 @@ def push(self) -> None:
                 "Submission does not pass validation. Please fix validation errors and re-stage."
             )
 
-        # TODO: do push
-        for k, _ in self._pending_push.items():
-            # call RawArchive static method to create file_paths from list of FileMetadata for the pending_calc[k]
-            # construct a RawArchive file for writing
-            # push that file to S3
-            pass
+        uploader.upload(self.id, self._pending_push)
 
         # do bookkeeping
         self.calc_history.append(self.pending_calculations)

emmet-cli/emmet/cli/submit.py

@@ -1,7 +1,11 @@
 import logging
 from pathlib import Path
+from uuid import UUID
+
 import click
-from emmet.cli.submission import Submission
+from emmet.cli.state_manager import StateManager
+from emmet.cli.submission import Submission, SubmissionUploader
+from emmet.cli.upload import HttpSubmissionUploader
 from emmet.cli.utils import EmmetCliError
 
 logger = logging.getLogger("emmet")
@@ -134,17 +138,26 @@ def validate(ctx: click.Context, submission: Path, check_all: bool) -> None:
     click.echo("Use 'emmet tasks status <task_id>' to check the status")
 
 
-def _push_submission(submission_path: Path) -> tuple[bool, str]:
+def _push_submission(
+    submission_path: Path,
+    state_dir: Path | None = None,
+    uploader: SubmissionUploader | None = None,
+) -> tuple[bool, str]:
     """Helper function to push a submission that can run in a separate process."""
     sub = Submission.load(submission_path)
-    updated_file_info = sub.stage_for_push()
-    if not updated_file_info:
+    changes = sub.stage_for_push()
+    if not changes.has_changes:
         return (
             False,
             "Files for submission have not changed since last update. Not pushing.",
         )
 
-    sub.push()
+    if uploader is None:
+        state_manager = StateManager(state_dir or Path.home() / ".emmet")
+        with HttpSubmissionUploader.from_environment(state_manager) as managed_uploader:
+            sub.push(managed_uploader)
+    else:
+        sub.push(uploader)
     sub.save(submission_path)
     return True, f"Successfully updated submission in {submission_path}"
 
@@ -157,6 +170,57 @@ def push(ctx: click.Context, submission: Path) -> None:
 
     Returns a task ID that can be used to check the status."""
     task_manager = ctx.obj["task_manager"]
-    task_id = task_manager.start_task(_push_submission, Path(submission))
+    state_dir = task_manager.state_manager.state_dir
+    task_id = task_manager.start_task(_push_submission, Path(submission), state_dir)
     click.echo(f"Push started. Task ID: {task_id}")
     click.echo("Use 'emmet tasks status <task_id>' to check the status")
+
+
+def _submission_id_from_target(target: str) -> UUID:
+    """Resolve a submission metadata path or UUID to a submission ID."""
+    target_path = Path(target)
+    if target_path.exists():
+        if not target_path.is_file():
+            raise EmmetCliError(f"Submission target is not a file: {target}")
+        try:
+            return Submission.load(target_path).id
+        except (OSError, ValueError):
+            raise EmmetCliError(
+                f"Could not load submission metadata from {target}."
+            ) from None
+    try:
+        return UUID(target)
+    except ValueError:
+        raise EmmetCliError(
+            f"Submission target must be an existing metadata file or UUID: {target}"
+        ) from None
+
+
+def _echo_object_ids(label: str, object_ids: list[str]) -> None:
+    click.echo(f"{label}: {len(object_ids)}")
+    for object_id in object_ids:
+        click.echo(f"  {object_id}")
+
+
+@submit.command("contributor-status")
+@click.pass_context
+def contributor_status(ctx: click.Context) -> None:
+    """Checks whether the current user can contribute submissions."""
+    state_manager = ctx.obj["task_manager"].state_manager
+    with HttpSubmissionUploader.from_environment(state_manager) as client:
+        status = client.contributor_status()
+    click.echo(f"Contributor status: {status}")
+
+
+@submit.command("status")
+@click.argument("target", nargs=1, type=str)
+@click.pass_context
+def submission_status(ctx: click.Context, target: str) -> None:
+    """Checks remote status using a submission metadata file or UUID."""
+    submission_id = _submission_id_from_target(target)
+    state_manager = ctx.obj["task_manager"].state_manager
+    with HttpSubmissionUploader.from_environment(state_manager) as client:
+        status = client.submission_status(submission_id)
+    click.echo(f"Submission {status['submission_id']}: {status['status']}")
+    _echo_object_ids("Completed objects", status["completed_object_ids"])
+    _echo_object_ids("In-progress objects", status["in_progress_object_ids"])