Merge pull request #66 from shunichironomura/feature/file-move

feature/file move

.gitignore

@@ -161,4 +161,3 @@ cython_debug/
 
 
 /vault
-/requirements.txt

capsula.toml

@@ -18,14 +18,14 @@ environment-variables = [
 [capture.files]
 "pyproject.toml" = { hash = "sha256", copy = true }
 "poetry.lock" = { hash = "sha256", copy = true }
-"requirements.txt" = { hash = "sha256", copy = true }
+"requirements.txt" = { hash = "sha256", move = true }
 
 [capture.git.repositories]
 capsula = '.'
 
 [monitor]
 [monitor.item.calculate-pi-cli.files]
-"examples/pi_cli.png" = { hash = "sha256" }
+"examples/pi_cli.png" = { hash = "sha256", move = true }
 
 [monitor.item.calculate-pi-dec.files]
-"examples/pi_dec.png" = { hash = "sha256" }
+"examples/pi_dec.png" = { hash = "sha256", move = true }

capsula/_monitor.py

@@ -23,7 +23,7 @@
 from datetime import datetime, timedelta
 from functools import wraps
 from pathlib import Path
-from shutil import copyfile
+from shutil import copyfile, move
 from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar
 
 if sys.version_info < (3, 10):
@@ -71,8 +71,8 @@ class PreRunInfoFunc(PreRunInfoBase):
 
 
 class OutputFileInfo(BaseModel):
-    hash_algorithm: Literal["md5", "sha1", "sha256", "sha3"]
-    file_hash: str = Field(..., alias="hash")
+    hash_algorithm: Literal["md5", "sha1", "sha256", "sha3"] | None
+    file_hash: str | None = Field(..., alias="hash")
 
 
 class PostRunInfoBase(BaseModel):
@@ -160,10 +160,12 @@ def teardown(self, post_run_info: _TPostRunInfo, *, items: Iterable[str]) -> _TP
                     continue
                 post_run_info.files[path] = OutputFileInfo(
                     hash_algorithm=file.hash_algorithm,
-                    hash=compute_hash(path, file.hash_algorithm),
+                    hash=compute_hash(path, file.hash_algorithm) if file.hash_algorithm else None,
                 )
                 if file.copy_:
                     copyfile(path, self.capture_config.capsule / path.name)
+                elif file.move:
+                    move(path, self.capture_config.capsule / path.name)
 
         with (self.capture_config.capsule / "post-run-info.json").open("w") as f:
             f.write(post_run_info.model_dump_json(indent=4))

capsula/context.py

@@ -5,7 +5,7 @@
 import sys
 from abc import ABC, abstractmethod
 from pathlib import Path
-from shutil import copyfile
+from shutil import copyfile, move
 from typing import TYPE_CHECKING, Any, Literal
 
 if sys.version_info < (3, 11):
@@ -124,19 +124,21 @@ def capture(cls, config: CaptureConfig) -> dict[str, Self]:
 
 
 class FileContext(ContextItem):
-    hash_algorithm: Literal["md5", "sha1", "sha256", "sha3"]
-    file_hash: str = Field(..., alias="hash")
+    hash_algorithm: Literal["md5", "sha1", "sha256", "sha3"] | None
+    file_hash: str | None = Field(..., alias="hash")
 
     @classmethod
     def capture(cls, config: CaptureConfig) -> dict[Path, Self]:
         files = {}
         for path, file_config in config.files.items():
             files[path] = cls(
-                hash=compute_hash(path, file_config.hash_algorithm),
+                hash=compute_hash(path, file_config.hash_algorithm) if file_config.hash_algorithm else None,
                 hash_algorithm=file_config.hash_algorithm,
             )
             if file_config.copy_:
                 copyfile(path, config.capsule / path.name)
+            elif file_config.move:
+                move(path, config.capsule / path.name)
 
         return files
 

capsula/file.py

@@ -1,8 +1,18 @@
+from __future__ import annotations
+
 from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 
 class CaptureFileConfig(BaseModel):
-    hash_algorithm: Literal["md5", "sha1", "sha256", "sha3"] = "sha256"
-    copy_: bool = Field(default=True, alias="copy")
+    hash_algorithm: Literal["md5", "sha1", "sha256", "sha3"] | None = None
+    copy_: bool = Field(default=False, alias="copy")
+    move: bool = False
+
+    @model_validator(mode="after")  # type: ignore
+    def exclusive_copy_move(cls, model: CaptureFileConfig) -> CaptureFileConfig:  # noqa: N805
+        if model.copy_ and model.move:
+            msg = "Only one of copy or move can be true"
+            raise ValueError(msg)
+        return model